diff --git a/specs/personal-knowledge-graph.yaml b/specs/personal-knowledge-graph.yaml
new file mode 100644
index 0000000..b882914
--- /dev/null
+++ b/specs/personal-knowledge-graph.yaml
@@ -0,0 +1,141 @@
+name: prophet-mesh-personal-knowledge-graph
+version: 0.1.0
+product: Prophet Mesh
+purpose: >-
+  The default graph structure for a person — the seed the Memory Steward builds
+  over. A schema over HellGraph (NodeAtom + AssertionClass + required provenance
+  + SHACL shapes), scoped by memory-scope.yaml, owned by the memory-steward
+  agent, populated by prophet-workspace (contacts / calendar / mail / drive), and
+  the per-person graph the mesh grounds retrieval on. Not a new store — a schema.
+
+invariants:
+  - self_is_the_single_anchor
+  - every_node_and_edge_has_provenance
+  - schema_links_are_structural
+  - world_claims_are_assertions
+  - open_world
+  - external_links_are_reference_only
+  - private_by_default
+  - inference_is_labeled
+
+# ── Node types (entity classes) — the vocabulary a person is built from ────────
+node_types:
+  Self:
+    class: Person
+    cardinality: exactly_one
+    role: "the graph anchor (the user)"
+  Person:
+    class: Person
+    qualifiers: [relationship]
+    examples: ["Mom", "Jamie", "colleague", "Mom's dentist"]
+  Place:
+    class: Place
+    examples: ["Hometown", "residence", "workplace location"]
+  Organization:
+    class: Organization
+    examples: ["High school", "employer", "dental practice"]
+  Thing:
+    class: Thing
+    aka: Possession
+    examples: ["Electric guitar", "Acoustic guitar", "devices"]
+  Interest:
+    class: Topic
+    aka: [Skill, Hobby, Domain]
+    examples: ["guitar", "music", "a profession"]
+  Event:
+    class: Event
+    source: [calendar]
+  Document:
+    class: CreativeWork
+    source: [drive, notes]
+  Communication:
+    class: Message
+    source: [mail, chat]
+    note: "often modeled as an edge with content"
+  Account:
+    class: ExternalIdentity
+    role: "anchor for external-KG links"
+    examples: ["social handle", "login"]
+
+# ── Edge types (typed relations) — solid edges inside the PKG ──────────────────
+edge_types:
+  relatedTo:        { subtypes: [parentOf, childOf, siblingOf, spouseOf], class: Assertion }
+  knows:            { subtypes: [friendOf, colleagueOf, acquaintanceOf], class: Assertion }
+  homeTown:         { from: Self, to: Place, class: Assertion }
+  residesIn:        { to: Place, class: Assertion }
+  attended:         { to: Organization, class: Assertion }
+  worksAt:          { to: Organization, class: Assertion }
+  memberOf:         { to: Organization, class: Assertion }
+  providerFor:      { subtypes: [dentistOf, doctorOf, lawyerOf], class: Assertion }
+  owns:             { to: Thing, class: Assertion }
+  uses:             { to: Thing, class: Assertion }
+  interestedIn:     { to: Interest, class: Assertion }
+  skilledIn:        { to: Interest, class: Assertion }
+  participatedIn:   { to: Event, class: Assertion }
+  authored:         { to: Document, class: Assertion }
+  communicatedWith: { to: Person, carries: [Communication], class: Assertion }
+  hasAccount:       { from: [Self, Person], to: Account, class: Structural }
+
+# ── External links (the dashed edges) — the cross-KG grounding contract ────────
+# A PKG node may point at an entity in an external graph. One-way by invariant:
+# the mesh RESOLVES/reads the external entity for grounding; nothing private
+# crosses outward. Every external link is an Assertion with full provenance.
+external_link:
+  edge: sameAs
+  class: Assertion
+  direction: reference_only
+  required_fields: [target_kg, target_id, confidence, trust_class, provenance]
+external_kgs:
+  general_purpose:
+    role: "entity resolution and world facts"
+    examples: [Wikidata, "KBpedia/KKO"]
+    links_from: [Place, Organization, Interest]
+  social_network:
+    role: "people and relationships"
+    links_from: [Person, Account]
+    privacy: high
+  domain_specific:
+    role: "domain KGs — music, medical, legal, finance"
+    links_from: [Interest, Thing]
+  ecommerce_catalog:
+    role: "products and listings"
+    links_from: [Thing]
+
+# ── Provenance + scope carried by EVERY node and edge ──────────────────────────
+required_on_every_element:
+  provenance: [source, captured_at, method]
+  epistemic_mode: [asserted, imported_rdf, identity]
+  assertion_class: [Structural, Assertion, Executable]
+  provenance_tag: [P-RET, P-GEN]
+  inference_type: [I-NON, I-DED, I-IND, I-ABD]
+  memory_scope: "relationship_context:approved"
+  confidence: "number in [0,1]"
+
+# ── The SEED — what a brand-new person's graph is, before anything is learned ──
+# Self + this schema + the external-KG link-points. prophet-workspace ingestion
+# then fills the typed slots: contacts → Person + knows/relatedTo; calendar →
+# Event + participatedIn; mail/chat → communicatedWith (+ Communication);
+# drive/notes → Document + authored.
+seed:
+  nodes:
+    - type: Self
+      id: self
+      provenance: { source: onboarding, method: declared }
+  registered_external_kgs: [general_purpose, social_network, domain_specific, ecommerce_catalog]
+  empty_slots:
+    people: []
+    places: []
+    organizations: []
+    things: []
+    interests: []
+    events: []
+    documents: []
+    accounts: []
+
+governance:
+  owner_agent: memory-steward
+  scope_policy: memory-scope.yaml
+  substrate: hellgraph
+  grounding: >-
+    This graph is the per-person hellgraph the mesh grounds on. Exam / holdout
+    content is NEVER ingested (Rule #0 — no contamination via retrieval).
diff --git a/src/prophet_mesh/pkg.py b/src/prophet_mesh/pkg.py
new file mode 100644
index 0000000..f86a842
--- /dev/null
+++ b/src/prophet_mesh/pkg.py
@@ -0,0 +1,174 @@
+"""Personal Knowledge Graph — the default graph a person is built over.
+
+Materializes the seed from specs/personal-knowledge-graph.yaml, ingests
+prophet-workspace data (contacts / calendar / mail / drive) into typed nodes +
+edges, and validates every element against the spec vocabulary + provenance
+requirement (a SHACL-lite check). Brain-agnostic: this is the data model; the
+mesh grounds on it, the Memory Steward owns it.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass, field, replace
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+SPEC_PATH = Path("specs/personal-knowledge-graph.yaml")
+
+
+def load_spec(path: str | Path = SPEC_PATH) -> dict[str, Any]:
+    return yaml.safe_load(Path(path).read_text())
+
+
+def valid_node_types(spec: dict) -> set[str]:
+    return set(spec["node_types"].keys())
+
+
+def valid_relations(spec: dict) -> set[str]:
+    """Top-level edge types + all declared subtypes."""
+    rels: set[str] = set()
+    for name, body in spec["edge_types"].items():
+        rels.add(name)
+        for sub in (body or {}).get("subtypes", []) or []:
+            rels.add(sub)
+    return rels
+
+
+@dataclass(frozen=True)
+class Provenance:
+    source: str                 # e.g. "onboarding", "workspace:contacts"
+    method: str                 # declared | imported | inferred | ingested
+    captured_at: str = ""       # ISO-8601, optional
+
+
+@dataclass(frozen=True)
+class ExternalLink:
+    target_kg: str              # general_purpose | social_network | domain_specific | ecommerce_catalog
+    target_id: str              # e.g. a Wikidata Q-id
+    confidence: float = 0.0
+    trust_class: str = "unverified"
+
+
+@dataclass(frozen=True)
+class Node:
+    id: str
+    type: str
+    label: str
+    provenance: Provenance
+    assertion_class: str = "Assertion"   # Structural | Assertion | Executable
+    provenance_tag: str = "P-RET"        # P-RET (pointer-bound) | P-GEN
+    inference_type: str = "I-NON"        # I-NON | I-DED | I-IND | I-ABD
+    memory_scope: str = "relationship_context:approved"
+    confidence: float = 1.0
+    external: tuple[ExternalLink, ...] = ()
+
+
+@dataclass(frozen=True)
+class Edge:
+    src: str
+    dst: str
+    rel: str
+    provenance: Provenance
+    assertion_class: str = "Assertion"
+    provenance_tag: str = "P-RET"
+    inference_type: str = "I-NON"
+    memory_scope: str = "relationship_context:approved"
+    confidence: float = 1.0
+
+
+@dataclass
+class PKG:
+    self_id: str
+    nodes: dict[str, Node] = field(default_factory=dict)
+    edges: list[Edge] = field(default_factory=list)
+    external_kgs: list[str] = field(default_factory=list)
+
+    def add_node(self, n: Node) -> None:
+        self.nodes[n.id] = n
+
+    def add_edge(self, e: Edge) -> None:
+        self.edges.append(e)
+
+
+# ── Seed ──────────────────────────────────────────────────────────────────────
+def seed_graph(self_id: str = "self", spec: dict | None = None) -> PKG:
+    """A brand-new person's graph: the Self anchor + registered external KGs."""
+    spec = spec or load_spec()
+    g = PKG(self_id=self_id, external_kgs=list(spec["seed"]["registered_external_kgs"]))
+    g.add_node(Node(
+        id=self_id, type="Self", label="Self",
+        provenance=Provenance(source="onboarding", method="declared"),
+        assertion_class="Structural",
+    ))
+    return g
+
+
+# ── Ingestion: prophet-workspace → typed nodes + edges ────────────────────────
+def _prov(app: str) -> Provenance:
+    return Provenance(source=f"workspace:{app}", method="ingested")
+
+
+def ingest_contact(g: PKG, name: str, relationship: str | None = None) -> str:
+    """A contact → a Person node + an edge from Self (relatedTo family / knows social)."""
+    nid = f"person:{name.strip().lower().replace(' ', '_')}"
+    g.add_node(Node(id=nid, type="Person", label=name, provenance=_prov("contacts")))
+    family = {"parentOf", "childOf", "siblingOf", "spouseOf", "mom", "dad", "mother", "father", "sister", "brother"}
+    rel = "relatedTo" if (relationship or "").lower() in family else "knows"
+    g.add_edge(Edge(src=g.self_id, dst=nid, rel=rel, provenance=_prov("contacts")))
+    return nid
+
+
+def ingest_event(g: PKG, title: str) -> str:
+    nid = f"event:{title.strip().lower().replace(' ', '_')}"
+    g.add_node(Node(id=nid, type="Event", label=title, provenance=_prov("calendar")))
+    g.add_edge(Edge(src=g.self_id, dst=nid, rel="participatedIn", provenance=_prov("calendar")))
+    return nid
+
+
+def ingest_message(g: PKG, peer_node_id: str) -> None:
+    """A message → a communicatedWith edge (the peer Person must already exist)."""
+    g.add_edge(Edge(src=g.self_id, dst=peer_node_id, rel="communicatedWith", provenance=_prov("mail")))
+
+
+def ingest_document(g: PKG, title: str) -> str:
+    nid = f"doc:{title.strip().lower().replace(' ', '_')}"
+    g.add_node(Node(id=nid, type="Document", label=title, provenance=_prov("drive")))
+    g.add_edge(Edge(src=g.self_id, dst=nid, rel="authored", provenance=_prov("drive")))
+    return nid
+
+
+def link_external(g: PKG, node_id: str, link: ExternalLink) -> None:
+    """Attach a reference-only external-KG link to a node (privacy: read, never leak)."""
+    n = g.nodes[node_id]
+    g.nodes[node_id] = replace(n, external=n.external + (link,))
+
+
+# ── Validation (SHACL-lite): vocabulary + provenance + invariants ─────────────
+def validate(g: PKG, spec: dict | None = None) -> list[str]:
+    spec = spec or load_spec()
+    ntypes, rels = valid_node_types(spec), valid_relations(spec)
+    kgs = set(spec["external_kgs"].keys())
+    errs: list[str] = []
+
+    selves = [n for n in g.nodes.values() if n.type == "Self"]
+    if len(selves) != 1:
+        errs.append(f"invariant self_is_the_single_anchor: found {len(selves)} Self nodes")
+
+    for n in g.nodes.values():
+        if n.type not in ntypes:
+            errs.append(f"node {n.id}: unknown type {n.type!r}")
+        if not n.provenance.source:
+            errs.append(f"node {n.id}: missing provenance.source")
+        for x in n.external:
+            if x.target_kg not in kgs:
+                errs.append(f"node {n.id}: external link to unknown KG {x.target_kg!r}")
+
+    for e in g.edges:
+        if e.rel not in rels:
+            errs.append(f"edge {e.src}->{e.dst}: unknown relation {e.rel!r}")
+        if e.src not in g.nodes or e.dst not in g.nodes:
+            errs.append(f"edge {e.src}->{e.dst} ({e.rel}): dangling endpoint")
+        if not e.provenance.source:
+            errs.append(f"edge {e.src}->{e.dst}: missing provenance.source")
+    return errs
diff --git a/src/prophet_mesh/pkg_gate.py b/src/prophet_mesh/pkg_gate.py
new file mode 100644
index 0000000..82d9eac
--- /dev/null
+++ b/src/prophet_mesh/pkg_gate.py
@@ -0,0 +1,155 @@
+"""CRDT Slice 4 — the locus / attestation merge gate.
+
+Slices 1–3 make the graph *converge* (everyone sees the same ops). Slice 4 decides
+*which converged sub-graph is authoritative*. This is where the correctness moat
+plugs in: TEE/confidentiality proves an op ran privately; it says nothing about
+whether the computation behind it is correct. The gate admits an op into the
+**canonical view** iff:
+
+  * its ``locus`` is ``local`` or ``trusted_private`` (authored where we already
+    trust the execution locus), OR
+  * its ``locus`` is ``attested_fog`` / ``burst_cloud`` **and** its
+    ``attestationRef`` resolves to a **passing correctness receipt**.
+
+Everything else lands in the **quarantine view** — still converged and retained
+(longevity), just not authoritative. Fail-closed: a gated-locus op with no /
+unresolved attestation is quarantined, and an untrusted retract therefore cannot
+silently delete canonical data.
+
+Emits a ``SyncCycleReceipt``-shaped receipt (sourceos-spec v2) per gate cycle.
+Accretive: layers over pkg_ops, touches nothing below it.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Callable, Iterable
+
+from .pkg import PKG
+from .pkg_ops import materialize, OpLog
+
+# Loci that require a passing correctness receipt to reach the canonical view.
+GATED_LOCI = ("attested_fog", "burst_cloud")
+TRUSTED_LOCI = ("local", "trusted_private")
+
+# A resolver answers: does this attestationRef name a *passing* correctness receipt?
+Resolver = Callable[[str], bool]
+
+
+def deny_all(_ref: str) -> bool:
+    """Default resolver — fail-closed. Nothing from a gated locus is admitted."""
+    return False
+
+
+def passing(*refs: str) -> Resolver:
+    """Resolver that admits exactly the given (passing) attestation refs."""
+    allowed = set(refs)
+    return lambda ref: ref in allowed
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _canon(obj) -> str:
+    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def graph_version(g: PKG) -> str:
+    """Content hash of materialized graph state — the SyncCycleReceipt version."""
+    nodes = sorted((n.id, n.label) for n in g.nodes.values())
+    edges = sorted((e.src, e.rel, e.dst) for e in g.edges)
+    return "sha256:" + hashlib.sha256(_canon({"nodes": nodes, "edges": edges}).encode("utf-8")).hexdigest()
+
+
+def admit(env: dict, resolve: Resolver) -> bool:
+    """Admission rule for a single op (fail-closed on gated loci)."""
+    crdt = env["payload"]["crdt"]
+    locus = crdt["locus"]
+    if locus in TRUSTED_LOCI:
+        return True
+    if locus in GATED_LOCI:
+        ref = crdt.get("attestationRef")
+        return bool(ref) and resolve(ref)
+    return False  # unknown locus → fail-closed
+
+
+@dataclass
+class GateResult:
+    canonical: PKG                       # authoritative graph
+    quarantine: PKG                      # held-back view (retained, not authoritative)
+    canonical_ops: list[dict]
+    quarantine_ops: list[dict]
+    receipt: dict                        # SyncCycleReceipt-shaped
+
+    @property
+    def admitted(self) -> int:
+        return len(self.canonical_ops)
+
+    @property
+    def quarantined(self) -> int:
+        return len(self.quarantine_ops)
+
+
+def gate(
+    ops: Iterable[dict],
+    *,
+    resolve: Resolver = deny_all,
+    self_id: str = "self",
+    org: str = "self",
+    content_view: str | None = None,
+    lifecycle_env: str = "candidate",
+    run_locus: str = "trusted_private",
+    from_version: str | None = None,
+    cycle_id: str | None = None,
+    agentplane_run_ref: str | None = None,
+) -> GateResult:
+    """Split a converged op-log into canonical vs. quarantine and emit a receipt."""
+    ops = ops.ops if isinstance(ops, OpLog) else list(ops)
+    canonical_ops = [e for e in ops if admit(e, resolve)]
+    quarantine_ops = [e for e in ops if not admit(e, resolve)]
+
+    canonical = materialize(canonical_ops, self_id)
+    quarantine = materialize(quarantine_ops, self_id)
+    to_version = graph_version(canonical)
+
+    n_admit, n_quar = len(canonical_ops), len(quarantine_ops)
+    if not ops:
+        outcome, gate_val, reason = "skipped", "no-op", "empty op-log"
+    elif n_admit == 0:
+        outcome, gate_val, reason = "denied", "denied", f"all {n_quar} op(s) held pending correctness attestation"
+    elif n_quar:
+        outcome, gate_val, reason = "applied", "partial", f"{n_admit} admitted, {n_quar} quarantined pending attestation"
+    else:
+        outcome, gate_val, reason = "applied", "allowed", f"{n_admit} op(s) admitted"
+
+    steps = [{"step": "admit", "status": "ok", "reason": f"{n_admit} op(s) → canonical"}]
+    if n_quar:
+        steps.append({"step": "quarantine", "status": "skipped",
+                      "reason": f"{n_quar} op(s) → quarantine (unattested gated-locus)"})
+
+    receipt = {
+        "id": f"urn:srcos:sync-receipt:{uuid.uuid4()}",
+        "type": "SyncCycleReceipt",
+        "specVersion": "2.0.0",
+        "cycleId": cycle_id or f"pkg-gate-{uuid.uuid4()}",
+        "engineId": "sourceos.sync.pkg-crdt-gate",
+        "org": org,
+        "contentView": content_view or f"pkg-{self_id}",
+        "fromVersion": from_version,
+        "toVersion": to_version,
+        "lifecycleEnv": lifecycle_env,
+        "locus": run_locus,
+        "outcome": outcome,
+        "policyGate": gate_val,
+        "policyReason": reason,
+        "steps": steps,
+        "issuedAt": _now(),
+        "auditId": f"urn:srcos:audit:{uuid.uuid4()}",
+        "agentplaneRunRef": agentplane_run_ref,
+    }
+
+    return GateResult(canonical, quarantine, canonical_ops, quarantine_ops, receipt)
diff --git a/src/prophet_mesh/pkg_hellgraph.py b/src/prophet_mesh/pkg_hellgraph.py
new file mode 100644
index 0000000..f634e63
--- /dev/null
+++ b/src/prophet_mesh/pkg_hellgraph.py
@@ -0,0 +1,121 @@
+"""PKG → HellGraph writer.
+
+Projects a Personal Knowledge Graph (pkg.PKG) into HellGraph's canonical wire
+shape so it persists on the real substrate. HellGraph's write path is the TS
+``HellGraphStore`` façade (``@socioprophet/hellgraph``):
+
+    g.addNode(id, labels[], properties)
+    g.addEdge(label, from, to, properties)
+
+with GraphNode = {id, labels[], properties, createdAt} and every edge carrying
+the mandatory epistemic fields {epistemicClass, confidence, promotionState,
+createdAt}. This module emits exactly those payloads (JSON-serializable dicts)
+from the Python side; a thin TS ingester replays them through the façade. We do
+NOT touch the Rust kernel — external writers go through the TS projection.
+
+Idempotent by construction: HellGraph structural atoms are content-addressed, so
+re-emitting the same PKG collapses to the same atoms.
+"""
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+from prophet_mesh.pkg import PKG, Edge, ExternalLink, Node
+
+# ── Epistemic mapping: PKG inference/method → HellGraph EpistemicClass ─────────
+# HellGraph EpistemicClass (ts/src/types.ts): extracted_relation | inferred_relation
+# | confirmed_relation | graph_extraction | semantic.
+# A PKG element sourced from onboarding or workspace ingestion is a confirmed
+# import (source of truth); a labelled inference is inferred_relation.
+_INFERRED = {"I-DED", "I-IND", "I-ABD"}
+
+
+def _epistemic_class(inference_type: str, method: str) -> str:
+    if inference_type in _INFERRED:
+        return "inferred_relation"
+    if method in {"declared", "imported", "ingested"}:
+        return "confirmed_relation"
+    return "extracted_relation"
+
+
+def _promotion_state(inference_type: str) -> str:
+    # Confirmed imports are accepted; inferences enter as candidates for review.
+    return "candidate" if inference_type in _INFERRED else "confirmed"
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _created_at(captured_at: str) -> str:
+    return captured_at or _now()
+
+
+def _external_props(external: tuple[ExternalLink, ...]) -> dict[str, Any]:
+    """External links are reference-only; carry them as flat, auditable props."""
+    props: dict[str, Any] = {}
+    for i, x in enumerate(external):
+        props[f"sameAs.{i}.kg"] = x.target_kg
+        props[f"sameAs.{i}.id"] = x.target_id
+        props[f"sameAs.{i}.confidence"] = x.confidence
+        props[f"sameAs.{i}.trust"] = x.trust_class
+        props[f"sameAs.{i}.direction"] = "reference_only"
+    return props
+
+
+def node_payload(n: Node) -> dict[str, Any]:
+    """One GraphNode. Labels = [type, assertion_class]; provenance in properties."""
+    props: dict[str, Any] = {
+        "label": n.label,
+        "provenance.source": n.provenance.source,
+        "provenance.method": n.provenance.method,
+        "provenance.captured_at": n.provenance.captured_at,
+        "assertion_class": n.assertion_class,
+        "provenance_tag": n.provenance_tag,       # P-RET | P-GEN
+        "inference_type": n.inference_type,        # I-NON | I-DED | I-IND | I-ABD
+        "memory_scope": n.memory_scope,
+        "confidence": n.confidence,
+    }
+    props.update(_external_props(n.external))
+    return {
+        "id": n.id,
+        "labels": [n.type, n.assertion_class],
+        "properties": props,
+        "createdAt": _created_at(n.provenance.captured_at),
+    }
+
+
+def edge_payload(e: Edge) -> dict[str, Any]:
+    """One GraphEdge. rel → label; PKG provenance → mandatory epistemic fields."""
+    return {
+        "label": e.rel,
+        "from": e.src,
+        "to": e.dst,
+        "properties": {
+            # HellGraph-mandatory epistemic fields:
+            "epistemicClass": _epistemic_class(e.inference_type, e.provenance.method),
+            "confidence": e.confidence,
+            "promotionState": _promotion_state(e.inference_type),
+            "createdAt": _created_at(e.provenance.captured_at),
+            # PKG provenance / scope carried through for audit + replay:
+            "provenance.source": e.provenance.source,
+            "provenance.method": e.provenance.method,
+            "assertion_class": e.assertion_class,
+            "provenance_tag": e.provenance_tag,
+            "inference_type": e.inference_type,
+            "memory_scope": e.memory_scope,
+        },
+    }
+
+
+def to_hellgraph(g: PKG) -> dict[str, list[dict[str, Any]]]:
+    """The full PKG as a HellGraph ingest bundle: {nodes:[...], edges:[...]}.
+
+    Feed to the TS ingester, which replays each through HellGraphStore.addNode /
+    .addEdge. Deterministic order (nodes before edges) so edge endpoints exist.
+    """
+    return {
+        "nodes": [node_payload(n) for n in g.nodes.values()],
+        "edges": [edge_payload(e) for e in g.edges],
+    }
diff --git a/src/prophet_mesh/pkg_ops.py b/src/prophet_mesh/pkg_ops.py
new file mode 100644
index 0000000..6194212
--- /dev/null
+++ b/src/prophet_mesh/pkg_ops.py
@@ -0,0 +1,299 @@
+"""CRDT over the Personal Knowledge Graph — Slices 1–3.
+
+Dual-writes every PKG mutation as a sourceos-spec v2 ``EventEnvelope`` op onto an
+append-only, hash-linked log (the Figure-18 "railroad track"), then **reconstructs
+and converges** graph state from those logs across replicas.
+
+  * Slice 1 — emit: an EventEnvelope-conformant op per add/retract, content-hash DAG.
+  * Slice 2 — reconstruct: deterministic ``materialize`` folds a log back to a PKG,
+    with add-wins OR-Set semantics + observed-remove retracts.
+  * Slice 3 — converge: ``merge`` unions two replicas' logs; ``materialize`` over the
+    union is commutative / associative / idempotent → strong eventual consistency
+    with zero coordination. This is what closes local-first ideals 2 (multi-device)
+    and 4 (collaboration) without touching 5/6/7.
+
+Causal + trust metadata (parents, lamport, locus, attestationRef, OR-Set tags/removes)
+rides in the open ``payload`` — EventEnvelope keeps ``additionalProperties: false`` at
+the top level, so no schema change. The locus/attestation *merge gate* (which converged
+sub-graph is authoritative) is Slice 4; here every op still converges.
+
+Conforms to: EventEnvelope.json; reuses the SyncCycleReceipt.locus vocabulary.
+Accretive: does not modify pkg.py.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import uuid
+from dataclasses import asdict
+from datetime import datetime, timezone
+from typing import Any, Iterable
+
+from .pkg import PKG, Node, Edge, Provenance, ExternalLink
+
+SPEC_VERSION = "2.0.0"
+
+# SyncCycleReceipt.locus vocabulary — the "distance-to-execution" axis.
+LOCI = ("local", "trusted_private", "attested_fog", "burst_cloud")
+
+EdgeKey = tuple[str, str, str]  # (src, rel, dst)
+
+
+def _now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _canonical(obj: Any) -> str:
+    """Stable JSON for content hashing (sorted keys, no whitespace drift)."""
+    return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def _content_hash(event_type: str, object_id: str, payload: dict) -> str:
+    """Merkle-DAG node hash. Covers semantics + causal parents, NOT wall-clock
+    (occurredAt is display-only and never trusted for ordering)."""
+    material = _canonical({"eventType": event_type, "objectId": object_id, "payload": payload})
+    return "sha256:" + hashlib.sha256(material.encode("utf-8")).hexdigest()
+
+
+class OpLog:
+    """Append-only EventEnvelope log with a single-replica hash chain.
+
+    Single-replica ``parents`` is the linear chain head; after a ``merge`` the DAG
+    is multi-head (``verify_chain`` is therefore a per-replica, pre-merge check)."""
+
+    def __init__(self, replica_id: str, subject_id: str, locus: str = "local") -> None:
+        if locus not in LOCI:
+            raise ValueError(f"locus must be one of {LOCI}, got {locus!r}")
+        self.replica_id = replica_id
+        self.subject_id = subject_id
+        self.locus = locus
+        self.ops: list[dict] = []
+        self._head: str | None = None
+        self._lamport = 0
+
+    def append(
+        self,
+        event_type: str,
+        object_id: str,
+        core: dict,
+        *,
+        attestation_ref: str | None = None,
+        trust_level: str = "trusted-workspace-source",
+        removes: Iterable[str] | None = None,
+    ) -> dict:
+        self._lamport += 1
+        event_id = f"urn:srcos:event:{uuid.uuid4()}"
+        crdt = {
+            "replicaId": self.replica_id,
+            "lamport": self._lamport,
+            "parents": [self._head] if self._head else [],
+            "locus": self.locus,
+            "attestationRef": attestation_ref,   # resolved by the Slice-4 merge gate
+            "trustLevel": trust_level,
+            "tag": event_id,                      # OR-Set unique add-tag
+        }
+        if removes is not None:
+            crdt["removes"] = list(removes)       # OR-Set observed-remove tags
+        payload = {"op": event_type, **core, "crdt": crdt}
+        event_hash = _content_hash(event_type, object_id, payload)
+        envelope = {
+            "eventId": event_id,
+            "eventType": event_type,
+            "specVersion": SPEC_VERSION,
+            "occurredAt": _now(),
+            "actor": {"subjectId": self.subject_id},
+            "objectId": object_id,
+            "payload": payload,
+            "integrity": {"eventHash": event_hash, "signature": None},
+        }
+        self._head = event_hash
+        self.ops.append(envelope)
+        return envelope
+
+
+def _node_urn(node_id: str) -> str:
+    return f"urn:srcos:pkg-node:{node_id}"
+
+
+def _edge_urn(e: Edge) -> str:
+    return f"urn:srcos:pkg-edge:{e.src}|{e.rel}|{e.dst}"
+
+
+def _node_from(nd: dict) -> Node:
+    nd = dict(nd)
+    nd["provenance"] = Provenance(**nd["provenance"])
+    nd["external"] = tuple(ExternalLink(**x) for x in nd.get("external", ()))
+    return Node(**nd)
+
+
+def _edge_from(ed: dict) -> Edge:
+    ed = dict(ed)
+    ed["provenance"] = Provenance(**ed["provenance"])
+    return Edge(**ed)
+
+
+class EmittingPKG:
+    """Composes a PKG with an OpLog: every mutation is applied AND emitted.
+
+    Non-invasive wrapper — the underlying PKG is untouched, so all existing
+    ingestion (ingest_contact, workspace adapters, hellgraph writer) can be driven
+    through this to start producing ops with zero changes to pkg.py. Tracks the
+    live add-tags per element so retracts are *observed-removes* (add-wins)."""
+
+    def __init__(self, graph: PKG, *, replica_id: str, locus: str = "local") -> None:
+        self.g = graph
+        self.log = OpLog(replica_id=replica_id, subject_id=f"urn:srcos:replica:{replica_id}", locus=locus)
+        self._node_tags: dict[str, set[str]] = {}
+        self._edge_tags: dict[EdgeKey, set[str]] = {}
+
+    @classmethod
+    def seeded(cls, self_id: str = "self", *, replica_id: str, locus: str = "local") -> "EmittingPKG":
+        """Fresh graph whose Self anchor is emitted as the genesis op, so the op-log
+        is a *complete* source of truth (a fresh replica can materialize it and
+        recover the anchor). external_kgs registration as an op is a later slice."""
+        epkg = cls(PKG(self_id=self_id), replica_id=replica_id, locus=locus)
+        epkg.add_node(Node(
+            id=self_id, type="Self", label="Self",
+            provenance=Provenance(source="onboarding", method="declared"),
+            assertion_class="Structural",
+        ))
+        return epkg
+
+    # ── local mutations (apply + emit) ──────────────────────────────────────
+    def add_node(self, n: Node, *, attestation_ref: str | None = None) -> dict:
+        self.g.add_node(n)
+        env = self.log.append("GraphNodeAsserted", _node_urn(n.id), {"node": asdict(n)}, attestation_ref=attestation_ref)
+        self._node_tags.setdefault(n.id, set()).add(env["payload"]["crdt"]["tag"])
+        return env
+
+    def add_edge(self, e: Edge, *, attestation_ref: str | None = None) -> dict:
+        self.g.add_edge(e)
+        env = self.log.append("GraphEdgeAsserted", _edge_urn(e), {"edge": asdict(e)}, attestation_ref=attestation_ref)
+        self._edge_tags.setdefault((e.src, e.rel, e.dst), set()).add(env["payload"]["crdt"]["tag"])
+        return env
+
+    def retract_node(self, node_id: str, *, attestation_ref: str | None = None) -> dict:
+        """Observed-remove: retract exactly the add-tags this replica has seen, so a
+        concurrent re-add elsewhere (a tag we didn't observe) survives — add-wins."""
+        observed = sorted(self._node_tags.get(node_id, set()))
+        env = self.log.append("GraphNodeRetracted", _node_urn(node_id), {"nodeId": node_id},
+                              attestation_ref=attestation_ref, removes=observed)
+        self.g.nodes.pop(node_id, None)
+        self._node_tags.pop(node_id, None)
+        return env
+
+    def retract_edge(self, key: EdgeKey, *, attestation_ref: str | None = None) -> dict:
+        observed = sorted(self._edge_tags.get(key, set()))
+        env = self.log.append("GraphEdgeRetracted", f"urn:srcos:pkg-edge:{key[0]}|{key[1]}|{key[2]}",
+                              {"edgeKey": list(key)}, attestation_ref=attestation_ref, removes=observed)
+        self.g.edges = [e for e in self.g.edges if (e.src, e.rel, e.dst) != key]
+        self._edge_tags.pop(key, None)
+        return env
+
+    # ── remote op ingestion (apply without re-emitting) ─────────────────────
+    def apply_remote(self, env: dict) -> None:
+        """Ingest another replica's op into local view + tag trackers, WITHOUT
+        appending it to our own log. Lets this replica observe a remote add so a
+        later local retract removes the right tag."""
+        p, crdt = env["payload"], env["payload"]["crdt"]
+        op = p["op"]
+        if op == "GraphNodeAsserted":
+            self.g.add_node(_node_from(p["node"]))
+            self._node_tags.setdefault(p["node"]["id"], set()).add(crdt["tag"])
+        elif op == "GraphEdgeAsserted":
+            ed = p["edge"]
+            key = (ed["src"], ed["rel"], ed["dst"])
+            self.g.add_edge(_edge_from(ed))
+            self._edge_tags.setdefault(key, set()).add(crdt["tag"])
+        elif op == "GraphNodeRetracted":
+            nid, rem = p["nodeId"], set(crdt.get("removes", []))
+            self._node_tags[nid] = self._node_tags.get(nid, set()) - rem
+            if not self._node_tags.get(nid):
+                self.g.nodes.pop(nid, None)
+                self._node_tags.pop(nid, None)
+        elif op == "GraphEdgeRetracted":
+            key = tuple(p["edgeKey"])
+            rem = set(crdt.get("removes", []))
+            self._edge_tags[key] = self._edge_tags.get(key, set()) - rem
+            if not self._edge_tags.get(key):
+                self.g.edges = [e for e in self.g.edges if (e.src, e.rel, e.dst) != key]
+                self._edge_tags.pop(key, None)
+
+
+# ── Slice 2/3: reconstruct + converge ──────────────────────────────────────
+def _dedupe(ops: Iterable[dict]) -> list[dict]:
+    """Idempotent op-set keyed by eventId (re-applying the same op is a no-op)."""
+    seen: dict[str, dict] = {}
+    for env in ops:
+        seen[env["eventId"]] = env
+    return list(seen.values())
+
+
+def materialize(ops: Iterable[dict], self_id: str = "self") -> PKG:
+    """Deterministically reconstruct graph state from an op-log (Slice 2).
+
+    Add-wins OR-Set: an element is present iff it has ≥1 add-tag not covered by an
+    observed-remove. Among surviving adds, value is last-writer-wins by
+    (lamport, eventId) — a total, replica-independent tiebreak. Pure fold over the
+    op *set*, so it is order-independent → the property Slice-3 merge relies on."""
+    events = _dedupe(ops)
+    node_adds: dict[str, list[tuple[int, str, dict, str]]] = {}
+    edge_adds: dict[EdgeKey, list[tuple[int, str, dict, str]]] = {}
+    removed: set[str] = set()
+
+    for env in events:
+        p, crdt = env["payload"], env["payload"]["crdt"]
+        op = p["op"]
+        if op == "GraphNodeAsserted":
+            node_adds.setdefault(p["node"]["id"], []).append(
+                (crdt["lamport"], env["eventId"], p["node"], crdt["tag"]))
+        elif op == "GraphEdgeAsserted":
+            e = p["edge"]
+            key = (e["src"], e["rel"], e["dst"])
+            edge_adds.setdefault(key, []).append(
+                (crdt["lamport"], env["eventId"], e, crdt["tag"]))
+        elif op in ("GraphNodeRetracted", "GraphEdgeRetracted"):
+            removed.update(crdt.get("removes", []))
+
+    g = PKG(self_id=self_id)
+    for adds in node_adds.values():
+        survivors = [a for a in adds if a[3] not in removed]
+        if survivors:
+            g.add_node(_node_from(max(survivors, key=lambda a: (a[0], a[1]))[2]))
+    for adds in edge_adds.values():
+        survivors = [a for a in adds if a[3] not in removed]
+        if survivors:
+            g.add_edge(_edge_from(max(survivors, key=lambda a: (a[0], a[1]))[2]))
+    return g
+
+
+def fold(ops: list[dict], self_id: str = "self") -> PKG:
+    """Slice-1 alias — reconstruct from the log (now backed by ``materialize``)."""
+    return materialize(ops, self_id=self_id)
+
+
+def merge(*logs: "OpLog | list[dict]") -> list[dict]:
+    """Union replicas' op-logs into one deduped, deterministically-ordered log (Slice 3).
+
+    Commutative / associative / idempotent by construction (set union keyed by
+    eventId). Ordering by (lamport, eventId) is only for a stable *serialized* form;
+    ``materialize`` is order-independent, so convergence does not depend on it."""
+    merged: list[dict] = []
+    for log in logs:
+        merged.extend(log.ops if isinstance(log, OpLog) else log)
+    return sorted(_dedupe(merged), key=lambda e: (e["payload"]["crdt"]["lamport"], e["eventId"]))
+
+
+def verify_chain(ops: list[dict]) -> bool:
+    """Every op's recorded eventHash matches its recomputed content hash, and parents
+    point at the immediately preceding head. Single-replica / pre-merge check — after
+    a merge the DAG is multi-head and this no longer applies."""
+    prev: str | None = None
+    for env in ops:
+        p = env["payload"]
+        if _content_hash(env["eventType"], env["objectId"], p) != env["integrity"]["eventHash"]:
+            return False
+        if p["crdt"]["parents"] != ([prev] if prev else []):
+            return False
+        prev = env["integrity"]["eventHash"]
+    return True
diff --git a/src/prophet_mesh/pkg_replica.py b/src/prophet_mesh/pkg_replica.py
new file mode 100644
index 0000000..3ab89b8
--- /dev/null
+++ b/src/prophet_mesh/pkg_replica.py
@@ -0,0 +1,129 @@
+"""CRDT Slice 5 — live replicas + the HellGraph ingester.
+
+Slices 1–4 are pure functions over op-logs. Slice 5 makes them *live*:
+
+  * ``Replica`` — a device/agent that holds a PKG + op-log and syncs with peers by
+    **anti-entropy** (exchange only the ops the other side is missing), converging
+    two live participants without any central coordinator.
+  * ``replay_to_hellgraph`` — the ingester the writer docstring promised but never
+    had (gap #4): replays a materialized PKG through a ``HellGraphStore``-shaped
+    sink (``addNode`` / ``addEdge``), so the op-log is no longer emit-only — it
+    lands on the real substrate.
+
+The two compose into the punchline: sync converges *everything* (longevity), the
+Slice-4 gate keeps only the proven-correct sub-graph, and **only that canonical
+graph is replayed to HellGraph.** Unattested cloud ops live in every replica's log
+but never reach the authoritative store.
+
+Accretive: reuses pkg_ops (merge/materialize/apply_remote), pkg_gate (gate), and
+pkg_hellgraph (to_hellgraph). Nothing below is modified.
+"""
+from __future__ import annotations
+
+from typing import Iterable, Protocol, runtime_checkable
+
+from .pkg import PKG, Node, Edge
+from .pkg_ops import EmittingPKG, EdgeKey, materialize
+from .pkg_gate import gate, GateResult, Resolver, deny_all
+from .pkg_hellgraph import to_hellgraph
+
+
+class Replica:
+    """A live CRDT participant. Local edits emit ops; ``sync`` reconciles with a
+    peer via anti-entropy. The authoritative view is always ``gate(self.ops)``."""
+
+    def __init__(self, replica_id: str, *, locus: str = "local", self_id: str = "self", seed: bool = True) -> None:
+        self.replica_id = replica_id
+        if seed:
+            self.epkg = EmittingPKG.seeded(self_id, replica_id=replica_id, locus=locus)
+        else:
+            self.epkg = EmittingPKG(PKG(self_id=self_id), replica_id=replica_id, locus=locus)
+        self._seen: set[str] = {e["eventId"] for e in self.epkg.log.ops}
+
+    # ── local mutations (delegate + track) ──────────────────────────────────
+    def _track(self, env: dict) -> dict:
+        self._seen.add(env["eventId"])
+        return env
+
+    def add_node(self, n: Node, **kw) -> dict:
+        return self._track(self.epkg.add_node(n, **kw))
+
+    def add_edge(self, e: Edge, **kw) -> dict:
+        return self._track(self.epkg.add_edge(e, **kw))
+
+    def retract_node(self, node_id: str, **kw) -> dict:
+        return self._track(self.epkg.retract_node(node_id, **kw))
+
+    def retract_edge(self, key: EdgeKey, **kw) -> dict:
+        return self._track(self.epkg.retract_edge(key, **kw))
+
+    @property
+    def ops(self) -> list[dict]:
+        return self.epkg.log.ops
+
+    # ── anti-entropy ─────────────────────────────────────────────────────────
+    def delta_for(self, have: set[str]) -> list[dict]:
+        """Ops this replica holds that the peer (``have`` = its seen-set) lacks."""
+        return [e for e in self.ops if e["eventId"] not in have]
+
+    def receive(self, ops: Iterable[dict]) -> int:
+        """Ingest a peer's ops we haven't seen: apply to view + tag trackers and
+        incorporate into our log (multi-head DAG). Returns the count newly applied."""
+        new = [e for e in ops if e["eventId"] not in self._seen]
+        max_lamport = self.epkg.log._lamport
+        for env in new:
+            self.epkg.apply_remote(env)
+            self.epkg.log.ops.append(env)
+            self._seen.add(env["eventId"])
+            max_lamport = max(max_lamport, env["payload"]["crdt"]["lamport"])
+        self.epkg.log._lamport = max_lamport  # Lamport advance so later local edits win
+        return len(new)
+
+    def materialized(self, self_id: str = "self") -> PKG:
+        """Fully converged view (pre-gate)."""
+        return materialize(self.ops, self_id=self_id)
+
+    def gated(self, *, resolve: Resolver = deny_all, **kw) -> GateResult:
+        """Authoritative view + quarantine + receipt (Slice-4 gate over our log)."""
+        return gate(self.ops, resolve=resolve, self_id=self.epkg.g.self_id, **kw)
+
+
+def sync(a: Replica, b: Replica) -> tuple[int, int]:
+    """Bidirectional anti-entropy: after this, both replicas hold the same op-set
+    and therefore ``materialize`` to the same state. Returns (a→b, b→a) counts."""
+    to_b = a.delta_for(b._seen)
+    to_a = b.delta_for(a._seen)
+    n_ab = b.receive(to_b)
+    n_ba = a.receive(to_a)
+    return n_ab, n_ba
+
+
+# ── the live HellGraph ingester (closes gap #4) ─────────────────────────────
+@runtime_checkable
+class HellGraphSink(Protocol):
+    """The subset of the TS ``HellGraphStore`` façade an ingester needs. The real
+    implementation is an HTTP client to ``@socioprophet/hellgraph``; tests use an
+    in-memory double."""
+
+    def addNode(self, id: str, labels: list[str], properties: dict, createdAt: str | None = None) -> None: ...
+    def addEdge(self, label: str, frm: str, to: str, properties: dict) -> None: ...
+
+
+def replay_to_hellgraph(g: PKG, sink: HellGraphSink) -> dict[str, int]:
+    """Replay a materialized PKG through the façade sink (nodes before edges, so
+    endpoints exist). Idempotent because HellGraph atoms are content-addressed."""
+    bundle = to_hellgraph(g)
+    for n in bundle["nodes"]:
+        sink.addNode(n["id"], n["labels"], n["properties"], n.get("createdAt"))
+    for e in bundle["edges"]:
+        sink.addEdge(e["label"], e["from"], e["to"], e["properties"])
+    return {"nodes": len(bundle["nodes"]), "edges": len(bundle["edges"])}
+
+
+def persist_canonical(replica: Replica, sink: HellGraphSink, *, resolve: Resolver = deny_all, **gate_kw) -> tuple[GateResult, dict[str, int]]:
+    """End-to-end: gate the replica's converged log, then replay ONLY the canonical
+    (proven-correct) graph to HellGraph. The quarantine view is retained in the log
+    but never persisted. Returns (gate result incl. receipt, replay counts)."""
+    res = replica.gated(resolve=resolve, **gate_kw)
+    counts = replay_to_hellgraph(res.canonical, sink)
+    return res, counts
diff --git a/src/prophet_mesh/pkg_workspace.py b/src/prophet_mesh/pkg_workspace.py
new file mode 100644
index 0000000..f454c04
--- /dev/null
+++ b/src/prophet_mesh/pkg_workspace.py
@@ -0,0 +1,148 @@
+"""prophet-workspace → PKG adapters.
+
+Maps real prophet-workspace records into the Personal Knowledge Graph, one
+adapter per contract:
+
+    contact.schema.json        → Person (+ worksAt Organization, social sameAs)
+    calendar-event.schema.json → Event + participatedIn (Self + attendees)
+    mail-message.schema.json   → communicatedWith (Self ↔ the correspondent)
+    office-artifact.schema.json→ Document + authored
+
+Every emitted node/edge carries workspace provenance: source =
+``workspace:<surface>:<adapter>``, method ``ingested``, captured_at from the
+record's own timestamp (updatedAt / receivedAt / createdAt). External identities
+(socialProfiles) become reference-only links to the social_network KG — never a
+data copy. These consume the canonical records; they do not invent shape.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from prophet_mesh.pkg import PKG, Edge, ExternalLink, Node, Provenance, link_external
+
+# Contact.labels / groupRefs hints that reclassify knows → relatedTo (family).
+FAMILY_HINTS = {
+    "family", "mom", "dad", "mother", "father", "sister", "brother",
+    "parent", "child", "sibling", "spouse", "wife", "husband", "son", "daughter",
+}
+
+
+def _prov(rec: dict[str, Any], surface: str, when_key: str = "updatedAt") -> Provenance:
+    adapter = rec.get("sourceAdapter") or rec.get("accountRef") or rec.get("workroomId") or "unknown"
+    captured = rec.get(when_key) or rec.get("createdAt") or ""
+    return Provenance(source=f"workspace:{surface}:{adapter}", method="ingested", captured_at=captured)
+
+
+def _person_id(ref: str) -> str:
+    return f"person:{ref}"
+
+
+# ── contact.schema.json → Person ──────────────────────────────────────────────
+def adapt_contact(g: PKG, rec: dict[str, Any]) -> str:
+    if rec.get("contactClass") == "organization":
+        return adapt_organization(g, rec)
+    cid = rec["contactId"]
+    nid = _person_id(cid)
+    label = rec.get("displayName") or " ".join(
+        p for p in (rec.get("givenName"), rec.get("familyName")) if p
+    ) or cid
+    prov = _prov(rec, "contacts")
+    g.add_node(Node(id=nid, type="Person", label=label, provenance=prov))
+
+    hints = {h.lower() for h in (rec.get("labels") or [])} | {h.lower() for h in (rec.get("groupRefs") or [])}
+    rel = "relatedTo" if hints & FAMILY_HINTS else "knows"
+    g.add_edge(Edge(src=g.self_id, dst=nid, rel=rel, provenance=prov))
+
+    if rec.get("organizationRef"):
+        oid = f"org:{rec['organizationRef']}"
+        if oid not in g.nodes:
+            g.add_node(Node(id=oid, type="Organization", label=rec["organizationRef"], provenance=prov))
+        g.add_edge(Edge(src=nid, dst=oid, rel="worksAt", provenance=prov))
+
+    for sp in rec.get("socialProfiles") or []:
+        target = sp.get("url") or f"{sp.get('platform')}:{sp.get('handle')}"
+        link_external(g, nid, ExternalLink(
+            target_kg="social_network", target_id=target,
+            confidence=0.6, trust_class="platform",
+        ))
+    return nid
+
+
+def adapt_organization(g: PKG, rec: dict[str, Any]) -> str:
+    oid = f"org:{rec['contactId']}"
+    prov = _prov(rec, "contacts")
+    g.add_node(Node(id=oid, type="Organization", label=rec.get("displayName") or rec["contactId"], provenance=prov))
+    return oid
+
+
+# ── calendar-event.schema.json → Event + participatedIn ───────────────────────
+def adapt_event(g: PKG, rec: dict[str, Any]) -> str:
+    eid = rec["eventId"]
+    nid = f"event:{eid}"
+    prov = _prov(rec, "calendar")
+    g.add_node(Node(id=nid, type="Event", label=rec.get("title") or eid, provenance=prov))
+    g.add_edge(Edge(src=g.self_id, dst=nid, rel="participatedIn", provenance=prov))
+
+    for att in rec.get("attendees") or []:
+        if att.get("isSelf"):
+            continue
+        cref = att.get("contactRef")
+        if not cref:
+            continue
+        pid = _person_id(cref)
+        if pid not in g.nodes:
+            g.add_node(Node(id=pid, type="Person", label=att.get("name") or cref, provenance=prov))
+        g.add_edge(Edge(src=pid, dst=nid, rel="participatedIn", provenance=prov))
+    return nid
+
+
+# ── mail-message.schema.json → communicatedWith ───────────────────────────────
+def adapt_message(g: PKG, rec: dict[str, Any]) -> str | None:
+    """A message → a communicatedWith edge Self↔correspondent (the `from` party)."""
+    prov = _prov(rec, "mail", when_key="receivedAt")
+    frm = rec.get("from") or {}
+    if frm.get("isSelf"):
+        return None  # outbound; the correspondent is on `to` — kept simple for now
+    cref = frm.get("contactRef")
+    if cref:
+        pid = _person_id(cref)
+        label = frm.get("name") or cref
+    else:
+        email = frm.get("email", "unknown")
+        pid = _person_id(f"email:{email}")
+        label = frm.get("name") or email
+    if pid not in g.nodes:
+        g.add_node(Node(id=pid, type="Person", label=label, provenance=prov))
+    g.add_edge(Edge(src=g.self_id, dst=pid, rel="communicatedWith", provenance=prov))
+    return pid
+
+
+# ── office-artifact.schema.json → Document + authored ─────────────────────────
+def adapt_artifact(g: PKG, rec: dict[str, Any]) -> str:
+    aid = rec["artifactId"]
+    nid = f"doc:{aid}"
+    prov = _prov(rec, "office", when_key="createdAt")
+    g.add_node(Node(id=nid, type="Document", label=rec.get("title") or aid, provenance=prov))
+    g.add_edge(Edge(src=g.self_id, dst=nid, rel="authored", provenance=prov))
+    return nid
+
+
+# ── batch ingest ──────────────────────────────────────────────────────────────
+def ingest_workspace(
+    g: PKG,
+    contacts: list[dict] | None = None,
+    events: list[dict] | None = None,
+    messages: list[dict] | None = None,
+    artifacts: list[dict] | None = None,
+) -> PKG:
+    """Ingest whole workspace exports in dependency order (people first so mail/
+    calendar edges resolve to existing Person nodes)."""
+    for rec in contacts or []:
+        adapt_contact(g, rec)
+    for rec in artifacts or []:
+        adapt_artifact(g, rec)
+    for rec in events or []:
+        adapt_event(g, rec)
+    for rec in messages or []:
+        adapt_message(g, rec)
+    return g
diff --git a/tests/test_pkg.py b/tests/test_pkg.py
new file mode 100644
index 0000000..505702e
--- /dev/null
+++ b/tests/test_pkg.py
@@ -0,0 +1,72 @@
+"""Tests for the Personal Knowledge Graph seed + ingestion + validation."""
+from __future__ import annotations
+
+from prophet_mesh.pkg import (
+    ExternalLink,
+    ingest_contact,
+    ingest_document,
+    ingest_event,
+    ingest_message,
+    link_external,
+    load_spec,
+    seed_graph,
+    validate,
+)
+
+SPEC = load_spec()
+
+
+def test_seed_is_single_self_anchor_plus_external_kgs():
+    g = seed_graph()
+    assert len(g.nodes) == 1
+    self_node = g.nodes["self"]
+    assert self_node.type == "Self"
+    assert self_node.assertion_class == "Structural"
+    assert self_node.provenance.source == "onboarding"
+    assert set(g.external_kgs) == set(SPEC["seed"]["registered_external_kgs"])
+    assert validate(g, SPEC) == []
+
+
+def test_ingest_contact_family_vs_social_relation():
+    g = seed_graph()
+    mom = ingest_contact(g, "Mom", relationship="mother")
+    jamie = ingest_contact(g, "Jamie", relationship="friend")
+    assert g.nodes[mom].type == "Person"
+    rels = {(e.src, e.dst): e.rel for e in g.edges}
+    assert rels[("self", mom)] == "relatedTo"   # family
+    assert rels[("self", jamie)] == "knows"      # social
+    # every ingested element carries provenance
+    assert all(e.provenance.source.startswith("workspace:") for e in g.edges)
+    assert validate(g, SPEC) == []
+
+
+def test_full_workspace_ingest_validates():
+    g = seed_graph()
+    jamie = ingest_contact(g, "Jamie", relationship="friend")
+    ingest_message(g, jamie)
+    ingest_event(g, "Band practice")
+    ingest_document(g, "Setlist")
+    errs = validate(g, SPEC)
+    assert errs == [], errs
+    # spot-check the typed shape
+    types = sorted({n.type for n in g.nodes.values()})
+    assert types == ["Document", "Event", "Person", "Self"]
+
+
+def test_external_link_is_reference_only_and_kg_checked():
+    g = seed_graph()
+    p = ingest_contact(g, "Jamie", relationship="friend")
+    link_external(g, p, ExternalLink(target_kg="social_network", target_id="sn:jamie", confidence=0.9, trust_class="platform"))
+    assert validate(g, SPEC) == []
+    # an unknown external KG is rejected by the validator
+    link_external(g, p, ExternalLink(target_kg="darkweb", target_id="x"))
+    assert any("unknown KG" in e for e in validate(g, SPEC))
+
+
+def test_validator_catches_missing_provenance_and_bad_vocab():
+    from dataclasses import replace
+    from prophet_mesh.pkg import Provenance
+    g = seed_graph()
+    # break provenance on a node
+    g.nodes["self"] = replace(g.nodes["self"], provenance=Provenance(source="", method="declared"))
+    assert any("missing provenance.source" in e for e in validate(g, SPEC))
diff --git a/tests/test_pkg_gate.py b/tests/test_pkg_gate.py
new file mode 100644
index 0000000..d3d4096
--- /dev/null
+++ b/tests/test_pkg_gate.py
@@ -0,0 +1,125 @@
+"""CRDT Slice 4 — locus / attestation merge gate.
+
+Proves: trusted loci admitted, gated loci fail-closed without a passing receipt,
+attested gated ops admitted, an untrusted retract cannot delete canonical data,
+nothing is lost, and the emitted receipt conforms to SyncCycleReceipt v2.
+"""
+from __future__ import annotations
+
+from prophet_mesh.pkg import PKG, Node, Provenance
+from prophet_mesh.pkg_ops import EmittingPKG, merge
+from prophet_mesh.pkg_gate import gate, passing, deny_all
+
+PASS_REF = "urn:srcos:reasoning-receipt:pass-001"
+
+
+def _prov(src="workspace:contacts"):
+    return Provenance(source=src, method="ingested")
+
+
+def _node(nid, label, src="workspace:contacts"):
+    return Node(id=nid, type="Person", label=label, provenance=_prov(src))
+
+
+def _trusted():
+    t = EmittingPKG.seeded("self", replica_id="device-a", locus="trusted_private")
+    t.add_node(_node("person:ada", "Ada"))
+    return t
+
+
+def _cloud():
+    """A burst_cloud replica emitting one attested + one unattested claim."""
+    c = EmittingPKG(PKG(self_id="self"), replica_id="cloud-1", locus="burst_cloud")
+    c.add_node(_node("claim:cpi", "CPI figure", src="cloud:analysis"), attestation_ref=PASS_REF)
+    c.add_node(_node("claim:gdp", "GDP figure", src="cloud:analysis"))  # no attestation
+    return c
+
+
+def _ids(g: PKG):
+    return {n.id for n in g.nodes.values()}
+
+
+# ── admission ───────────────────────────────────────────────────────────────
+def test_trusted_loci_are_admitted():
+    t = _trusted()
+    res = gate(t.log.ops)  # default deny_all resolver — but these are trusted_private
+    assert _ids(res.canonical) == {"self", "person:ada"}
+    assert res.quarantined == 0
+
+
+def test_unattested_gated_op_is_quarantined_fail_closed():
+    merged = merge(_trusted().log, _cloud().log)
+    res = gate(merged, resolve=deny_all)  # nothing from cloud can pass
+    assert "claim:gdp" not in _ids(res.canonical)
+    assert "claim:cpi" not in _ids(res.canonical)   # even attested — resolver denies
+    assert "claim:gdp" in _ids(res.quarantine)
+
+
+def test_attested_gated_op_is_admitted():
+    merged = merge(_trusted().log, _cloud().log)
+    res = gate(merged, resolve=passing(PASS_REF))
+    assert "claim:cpi" in _ids(res.canonical)        # attested → canonical
+    assert "claim:gdp" not in _ids(res.canonical)    # unattested → still quarantined
+    assert "claim:gdp" in _ids(res.quarantine)
+
+
+# ── security property ─────────────────────────────────────────────────────────
+def test_untrusted_retract_cannot_delete_canonical_data():
+    t = _trusted()                                   # self + Ada (trusted_private)
+    ada_add = [e for e in t.log.ops if e["payload"].get("node", {}).get("id") == "person:ada"][0]
+
+    attacker = EmittingPKG(PKG(self_id="self"), replica_id="cloud-evil", locus="burst_cloud")
+    attacker.apply_remote(ada_add)                   # observes Ada's tag
+    attacker.retract_node("person:ada")              # unattested burst_cloud retract
+
+    merged = merge(t.log, attacker.log)
+    res = gate(merged, resolve=deny_all)
+    assert "person:ada" in _ids(res.canonical)       # the untrusted retract was quarantined
+    assert res.quarantined >= 1
+
+
+# ── longevity: nothing lost ──────────────────────────────────────────────────
+def test_nothing_is_lost():
+    merged = merge(_trusted().log, _cloud().log)
+    res = gate(merged, resolve=passing(PASS_REF))
+    assert res.admitted + res.quarantined == len(merged)
+
+
+# ── receipt conformance ──────────────────────────────────────────────────────
+_ALLOWED = {  # SyncCycleReceipt.json top-level keys (additionalProperties:false)
+    "id", "type", "specVersion", "cycleId", "engineId", "org", "contentView",
+    "fromVersion", "toVersion", "lifecycleEnv", "locus", "outcome", "policyGate",
+    "policyReason", "steps", "nixCacheUrl", "flakeRef", "durationMs", "issuedAt",
+    "auditId", "agentplaneRunRef",
+}
+_REQUIRED = {"id", "type", "specVersion", "cycleId", "engineId", "org", "contentView",
+             "toVersion", "lifecycleEnv", "locus", "outcome", "steps", "issuedAt", "auditId"}
+_OUTCOMES = {"planned", "dry_run", "applied", "skipped", "denied", "failed"}
+_STEP_STATUS = {"dry_run", "ok", "failed", "skipped", "timeout"}
+_LOCI = {"local", "trusted_private", "attested_fog", "burst_cloud"}
+
+
+def test_receipt_is_synccyclereceipt_conformant():
+    res = gate(merge(_trusted().log, _cloud().log), resolve=passing(PASS_REF))
+    r = res.receipt
+    assert set(r.keys()) <= _ALLOWED
+    assert _REQUIRED <= set(r.keys())
+    assert r["type"] == "SyncCycleReceipt"
+    assert r["id"].startswith("urn:srcos:sync-receipt:")
+    assert r["engineId"].startswith("sourceos.sync.")
+    assert r["auditId"].startswith("urn:srcos:audit:")
+    assert r["outcome"] in _OUTCOMES
+    assert r["locus"] in _LOCI
+    assert r["toVersion"].startswith("sha256:")
+    for step in r["steps"]:
+        assert {"step", "status"} <= set(step.keys())
+        assert step["status"] in _STEP_STATUS
+
+
+def test_receipt_outcome_reflects_gate_decision():
+    # all quarantined → denied; mixed → applied/partial
+    denied = gate(_cloud().log, resolve=deny_all).receipt
+    assert denied["outcome"] == "denied" and denied["policyGate"] == "denied"
+
+    mixed = gate(merge(_trusted().log, _cloud().log), resolve=passing(PASS_REF)).receipt
+    assert mixed["outcome"] == "applied" and mixed["policyGate"] == "partial"
diff --git a/tests/test_pkg_hellgraph.py b/tests/test_pkg_hellgraph.py
new file mode 100644
index 0000000..b0e5134
--- /dev/null
+++ b/tests/test_pkg_hellgraph.py
@@ -0,0 +1,62 @@
+"""Tests for the PKG → HellGraph writer (canonical wire-shape conformance)."""
+from __future__ import annotations
+
+from prophet_mesh.pkg import ExternalLink, ingest_contact, link_external, seed_graph
+from prophet_mesh.pkg_hellgraph import edge_payload, node_payload, to_hellgraph
+
+REQUIRED_EDGE_EPISTEMIC = {"epistemicClass", "confidence", "promotionState", "createdAt"}
+EPISTEMIC_CLASSES = {
+    "extracted_relation", "inferred_relation", "confirmed_relation",
+    "graph_extraction", "semantic",
+}
+PROMOTION_STATES = {"candidate", "confirmed", "contested", "superseded", "vetoed"}
+
+
+def test_node_payload_has_graphnode_shape():
+    g = seed_graph()
+    p = node_payload(g.nodes["self"])
+    assert set(p.keys()) == {"id", "labels", "properties", "createdAt"}
+    assert p["id"] == "self"
+    assert p["labels"][0] == "Self" and "Structural" in p["labels"]
+    assert p["properties"]["provenance.source"] == "onboarding"
+    assert p["createdAt"]  # never empty
+
+
+def test_edge_payload_carries_mandatory_epistemic_fields():
+    g = seed_graph()
+    ingest_contact(g, "Mom", relationship="mother")
+    e = g.edges[0]
+    p = edge_payload(e)
+    assert set(p.keys()) == {"label", "from", "to", "properties"}
+    assert REQUIRED_EDGE_EPISTEMIC <= set(p["properties"])
+    assert p["properties"]["epistemicClass"] in EPISTEMIC_CLASSES
+    assert p["properties"]["promotionState"] in PROMOTION_STATES
+
+
+def test_ingested_workspace_edge_is_confirmed_import():
+    g = seed_graph()
+    ingest_contact(g, "Jamie", relationship="friend")
+    props = edge_payload(g.edges[0])["properties"]
+    # method="ingested", I-NON → a confirmed import, not a speculative candidate
+    assert props["epistemicClass"] == "confirmed_relation"
+    assert props["promotionState"] == "confirmed"
+
+
+def test_external_link_is_reference_only_in_props():
+    g = seed_graph()
+    p = ingest_contact(g, "Jamie", relationship="friend")
+    link_external(g, p, ExternalLink("social_network", "sn:jamie", 0.9, "platform"))
+    props = node_payload(g.nodes[p])["properties"]
+    assert props["sameAs.0.kg"] == "social_network"
+    assert props["sameAs.0.direction"] == "reference_only"
+
+
+def test_full_bundle_nodes_before_edges_and_endpoints_exist():
+    g = seed_graph()
+    jamie = ingest_contact(g, "Jamie", relationship="friend")
+    bundle = to_hellgraph(g)
+    node_ids = {n["id"] for n in bundle["nodes"]}
+    assert "self" in node_ids and jamie in node_ids
+    # every edge endpoint is present as a node in the same bundle
+    for e in bundle["edges"]:
+        assert e["from"] in node_ids and e["to"] in node_ids
diff --git a/tests/test_pkg_merge.py b/tests/test_pkg_merge.py
new file mode 100644
index 0000000..6868828
--- /dev/null
+++ b/tests/test_pkg_merge.py
@@ -0,0 +1,129 @@
+"""CRDT Slice 2/3 — reconstruct + converge.
+
+Proves the OR-Set merge is commutative / associative / idempotent, that offline
+replicas converge, and that concurrent add-vs-retract resolves add-wins.
+"""
+from __future__ import annotations
+
+import random
+
+from prophet_mesh.pkg import PKG, Node, Edge, Provenance
+from prophet_mesh.pkg_ops import EmittingPKG, materialize, merge
+
+
+def _prov(src="workspace:contacts"):
+    return Provenance(source=src, method="ingested")
+
+
+def _node(nid, label):
+    return Node(id=nid, type="Person", label=label, provenance=_prov())
+
+
+def _edge(src, rel, dst):
+    return Edge(src=src, dst=dst, rel=rel, provenance=_prov())
+
+
+def _state(g: PKG):
+    """Comparable convergence fingerprint: node ids+labels and edge triples."""
+    nodes = frozenset((n.id, n.label) for n in g.nodes.values())
+    edges = frozenset((e.src, e.rel, e.dst) for e in g.edges)
+    return nodes, edges
+
+
+def _replica_a():
+    a = EmittingPKG.seeded("self", replica_id="device-a", locus="trusted_private")
+    a.add_node(_node("person:ada", "Ada"))
+    a.add_edge(_edge("self", "knows", "person:ada"))
+    return a
+
+
+def _replica_b():
+    b = EmittingPKG.seeded("self", replica_id="device-b", locus="trusted_private")
+    b.add_node(_node("person:bob", "Bob"))
+    b.add_edge(_edge("self", "knows", "person:bob"))
+    return b
+
+
+# ── Slice 2: reconstruct ────────────────────────────────────────────────────
+def test_materialize_matches_live_graph():
+    a = _replica_a()
+    assert _state(materialize(a.log.ops)) == _state(a.g)
+
+
+def test_retract_removes_element_but_keeps_the_rest():
+    a = _replica_a()
+    a.retract_node("person:ada")
+    g = materialize(a.log.ops)
+    ids = {n.id for n in g.nodes.values()}
+    assert "person:ada" not in ids and "self" in ids
+
+
+# ── Slice 3: converge ───────────────────────────────────────────────────────
+def test_merge_is_commutative():
+    a, b = _replica_a(), _replica_b()
+    assert _state(materialize(merge(a.log, b.log))) == _state(materialize(merge(b.log, a.log)))
+
+
+def test_merge_is_idempotent():
+    a = _replica_a()
+    once = merge(a.log)
+    twice = merge(a.log, a.log)
+    assert {e["eventId"] for e in once} == {e["eventId"] for e in twice}
+    assert _state(materialize(twice)) == _state(materialize(a.log.ops))
+
+
+def test_merge_is_associative():
+    a, b = _replica_a(), _replica_b()
+    c = EmittingPKG.seeded("self", replica_id="device-c", locus="local")
+    c.add_node(_node("person:cyd", "Cyd"))
+    left = merge(merge(a.log, b.log), c.log)
+    right = merge(a.log, merge(b.log, c.log))
+    assert _state(materialize(left)) == _state(materialize(right))
+
+
+def test_two_offline_replicas_converge():
+    a, b = _replica_a(), _replica_b()  # each edited offline, never saw the other
+    converged = materialize(merge(a.log, b.log))
+    nodes, edges = _state(converged)
+    ids = {nid for nid, _ in nodes}
+    assert ids == {"self", "person:ada", "person:bob"}
+    assert ("self", "knows", "person:ada") in edges
+    assert ("self", "knows", "person:bob") in edges
+
+
+def test_order_independent_under_random_shuffle():
+    a, b = _replica_a(), _replica_b()
+    union = merge(a.log, b.log)
+    baseline = _state(materialize(union))
+    rng = random.Random(1729)
+    for _ in range(20):
+        shuffled = union[:]
+        rng.shuffle(shuffled)
+        assert _state(materialize(shuffled)) == baseline
+
+
+def test_add_wins_over_concurrent_retract():
+    # A adds Ada; B observes that add, then retracts it. Concurrently A re-adds Ada
+    # (a NEW tag B never saw) with an updated label. Merge → Ada present, updated.
+    a = EmittingPKG.seeded("self", replica_id="device-a", locus="local")
+    add1 = a.add_node(_node("person:ada", "Ada"))
+
+    b = EmittingPKG.seeded("self", replica_id="device-b", locus="local")
+    b.apply_remote(add1)                 # B observed Ada@tag1
+    b.retract_node("person:ada")         # removes only tag1
+
+    a.add_node(_node("person:ada", "Ada (updated)"))  # concurrent re-add, tag2
+
+    converged = materialize(merge(a.log, b.log))
+    ada = [n for n in converged.nodes.values() if n.id == "person:ada"]
+    assert len(ada) == 1                 # add-wins: survives the concurrent retract
+    assert ada[0].label == "Ada (updated)"
+
+
+def test_retract_wins_when_all_observed_tags_removed():
+    # If the retract observed every add-tag, the element is genuinely gone.
+    a = EmittingPKG.seeded("self", replica_id="device-a", locus="local")
+    a.add_node(_node("person:ada", "Ada"))
+    a.retract_node("person:ada")         # observed the only tag
+    ids = {n.id for n in materialize(a.log.ops).nodes.values()}
+    assert "person:ada" not in ids
diff --git a/tests/test_pkg_ops.py b/tests/test_pkg_ops.py
new file mode 100644
index 0000000..9c17278
--- /dev/null
+++ b/tests/test_pkg_ops.py
@@ -0,0 +1,74 @@
+"""CRDT Slice 1 tests — op-emitter conformance, hash chain, and fold round-trip."""
+from __future__ import annotations
+
+import re
+
+from prophet_mesh.pkg import PKG, Node, Edge, Provenance
+from prophet_mesh.pkg_ops import EmittingPKG, fold, verify_chain, LOCI, SPEC_VERSION
+
+_EVENT_ID = re.compile(r"^urn:srcos:event:")
+_REQUIRED = {"eventId", "eventType", "specVersion", "occurredAt", "actor", "objectId", "payload"}
+
+
+def _build() -> EmittingPKG:
+    # seeded() emits the Self anchor as the genesis op → the log is complete.
+    epkg = EmittingPKG.seeded("self", replica_id="device-a", locus="trusted_private")
+    epkg.add_node(Node(
+        id="person:ada", type="Person", label="Ada",
+        provenance=Provenance(source="workspace:contacts", method="ingested"),
+    ))
+    epkg.add_edge(Edge(
+        src="self", dst="person:ada", rel="knows",
+        provenance=Provenance(source="workspace:contacts", method="ingested"),
+    ))
+    return epkg
+
+
+def test_every_mutation_emits_a_conformant_envelope():
+    epkg = _build()
+    assert len(epkg.log.ops) == 3  # genesis Self + Person node + edge
+    for env in epkg.log.ops:
+        # EventEnvelope required fields (additionalProperties:false — only allowed keys)
+        assert _REQUIRED <= set(env.keys())
+        assert set(env.keys()) <= _REQUIRED | {"integrity"}
+        assert _EVENT_ID.match(env["eventId"])
+        assert env["specVersion"] == SPEC_VERSION
+        assert env["actor"]["subjectId"].startswith("urn:srcos:replica:")
+        assert env["eventType"] in ("GraphNodeAsserted", "GraphEdgeAsserted")
+
+
+def test_crdt_metadata_records_locus_and_causal_order():
+    epkg = _build()
+    self_op, node_op, edge_op = epkg.log.ops
+    for env in (self_op, node_op, edge_op):
+        crdt = env["payload"]["crdt"]
+        assert crdt["locus"] in LOCI and crdt["locus"] == "trusted_private"
+        assert crdt["tag"] == env["eventId"]          # OR-Set add-tag identity
+        assert crdt["trustLevel"] == "trusted-workspace-source"
+    # lamport increases; edge causally follows the node (parents = node's hash)
+    assert self_op["payload"]["crdt"]["lamport"] == 1
+    assert node_op["payload"]["crdt"]["lamport"] == 2
+    assert edge_op["payload"]["crdt"]["lamport"] == 3
+    assert edge_op["payload"]["crdt"]["parents"] == [node_op["integrity"]["eventHash"]]
+
+
+def test_hash_chain_verifies():
+    epkg = _build()
+    assert verify_chain(epkg.log.ops) is True
+
+
+def test_tamper_breaks_the_chain():
+    epkg = _build()
+    epkg.log.ops[0]["payload"]["node"]["label"] = "Mallory"  # mutate content, not hash
+    assert verify_chain(epkg.log.ops) is False
+
+
+def test_fold_reconstructs_graph_state_from_log_alone():
+    epkg = _build()
+    rebuilt: PKG = fold(epkg.log.ops)
+    # nodes + edges recovered from the op-log match the live graph
+    assert set(rebuilt.nodes) == set(epkg.g.nodes)
+    assert rebuilt.nodes["person:ada"].label == "Ada"
+    live_edges = {(e.src, e.rel, e.dst) for e in epkg.g.edges}
+    fold_edges = {(e.src, e.rel, e.dst) for e in rebuilt.edges}
+    assert live_edges == fold_edges
diff --git a/tests/test_pkg_replica.py b/tests/test_pkg_replica.py
new file mode 100644
index 0000000..dad9ff7
--- /dev/null
+++ b/tests/test_pkg_replica.py
@@ -0,0 +1,130 @@
+"""CRDT Slice 5 — live replicas + HellGraph ingester.
+
+Proves two live replicas converge via anti-entropy, the ingester replays a PKG
+through a HellGraphStore-shaped sink (idempotently), and end-to-end only the
+gated-canonical graph reaches the store.
+"""
+from __future__ import annotations
+
+from prophet_mesh.pkg import PKG, Node, Edge, Provenance
+from prophet_mesh.pkg_replica import (
+    Replica, sync, replay_to_hellgraph, persist_canonical, HellGraphSink,
+)
+from prophet_mesh.pkg_gate import passing
+
+PASS_REF = "urn:srcos:reasoning-receipt:pass-001"
+
+
+def _prov(src="workspace:contacts"):
+    return Provenance(source=src, method="ingested")
+
+
+def _node(nid, label, src="workspace:contacts"):
+    return Node(id=nid, type="Person", label=label, provenance=_prov(src))
+
+
+def _edge(src, rel, dst):
+    return Edge(src=src, dst=dst, rel=rel, provenance=_prov())
+
+
+def _state(g: PKG):
+    return (frozenset((n.id, n.label) for n in g.nodes.values()),
+            frozenset((e.src, e.rel, e.dst) for e in g.edges))
+
+
+class FakeHellGraphStore:
+    """In-memory double of the TS HellGraphStore façade. Content-addressed:
+    nodes keyed by id, edges by (from, label, to) — so replay is idempotent."""
+
+    def __init__(self):
+        self.nodes: dict[str, dict] = {}
+        self.edges: dict[tuple, dict] = {}
+
+    def addNode(self, id, labels, properties, createdAt=None):
+        self.nodes[id] = {"labels": labels, "properties": properties, "createdAt": createdAt}
+
+    def addEdge(self, label, frm, to, properties):
+        self.edges[(frm, label, to)] = properties
+
+
+# ── live convergence ─────────────────────────────────────────────────────────
+def test_two_live_replicas_converge_via_sync():
+    a = Replica("device-a", locus="trusted_private")
+    b = Replica("device-b", locus="trusted_private")
+    a.add_node(_node("person:ada", "Ada"))
+    a.add_edge(_edge("self", "knows", "person:ada"))
+    b.add_node(_node("person:bob", "Bob"))
+    b.add_edge(_edge("self", "knows", "person:bob"))
+
+    n_ab, n_ba = sync(a, b)
+    assert n_ab > 0 and n_ba > 0
+    assert a._seen == b._seen                       # anti-entropy complete
+    assert _state(a.materialized()) == _state(b.materialized())
+    ids = {n.id for n in a.materialized().nodes.values()}
+    assert ids == {"self", "person:ada", "person:bob"}
+
+
+def test_sync_is_idempotent_second_pass_transfers_nothing():
+    a = Replica("device-a", locus="trusted_private")
+    b = Replica("device-b", locus="trusted_private")
+    a.add_node(_node("person:ada", "Ada"))
+    sync(a, b)
+    assert sync(a, b) == (0, 0)                      # nothing left to exchange
+
+
+def test_local_edit_after_sync_propagates_back():
+    a = Replica("device-a", locus="trusted_private")
+    b = Replica("device-b", locus="trusted_private")
+    sync(a, b)
+    b.add_node(_node("person:cyd", "Cyd"))          # B edits after first sync
+    sync(a, b)
+    assert "person:cyd" in {n.id for n in a.materialized().nodes.values()}
+
+
+# ── the ingester (gap #4) ────────────────────────────────────────────────────
+def test_replay_persists_to_hellgraph_sink():
+    a = Replica("device-a", locus="trusted_private")
+    a.add_node(_node("person:ada", "Ada"))
+    a.add_edge(_edge("self", "knows", "person:ada"))
+    store = FakeHellGraphStore()
+    assert isinstance(store, HellGraphSink)          # structural conformance
+    counts = replay_to_hellgraph(a.materialized(), store)
+    assert "person:ada" in store.nodes and "self" in store.nodes
+    assert ("self", "knows", "person:ada") in store.edges
+    assert counts["nodes"] == len(store.nodes)
+
+
+def test_replay_is_idempotent():
+    a = Replica("device-a", locus="trusted_private")
+    a.add_node(_node("person:ada", "Ada"))
+    a.add_edge(_edge("self", "knows", "person:ada"))
+    store = FakeHellGraphStore()
+    replay_to_hellgraph(a.materialized(), store)
+    n1, e1 = len(store.nodes), len(store.edges)
+    replay_to_hellgraph(a.materialized(), store)     # replay again
+    assert (len(store.nodes), len(store.edges)) == (n1, e1)
+
+
+# ── end-to-end: only the canonical graph is persisted ────────────────────────
+def test_only_gated_canonical_graph_reaches_the_store():
+    a = Replica("device-a", locus="trusted_private")
+    a.add_node(_node("person:ada", "Ada"))
+
+    cloud = Replica("cloud-1", locus="burst_cloud", seed=False)
+    cloud.add_node(_node("claim:cpi", "CPI", src="cloud:analysis"), attestation_ref=PASS_REF)
+    cloud.add_node(_node("claim:gdp", "GDP", src="cloud:analysis"))   # unattested
+
+    sync(a, cloud)                                   # everything converges into both logs
+    # both unattested + attested claims are present in the converged (pre-gate) view
+    conv_ids = {n.id for n in a.materialized().nodes.values()}
+    assert {"claim:cpi", "claim:gdp"} <= conv_ids
+
+    store = FakeHellGraphStore()
+    res, counts = persist_canonical(a, store, resolve=passing(PASS_REF))
+
+    persisted = set(store.nodes)
+    assert "person:ada" in persisted                 # trusted → persisted
+    assert "claim:cpi" in persisted                  # attested → persisted
+    assert "claim:gdp" not in persisted              # unattested → quarantined, NOT persisted
+    assert res.quarantined >= 1
+    assert res.receipt["type"] == "SyncCycleReceipt"
diff --git a/tests/test_pkg_workspace.py b/tests/test_pkg_workspace.py
new file mode 100644
index 0000000..e11d6e2
--- /dev/null
+++ b/tests/test_pkg_workspace.py
@@ -0,0 +1,104 @@
+"""Tests for the prophet-workspace → PKG adapters (real record shapes)."""
+from __future__ import annotations
+
+from prophet_mesh.pkg import load_spec, seed_graph, validate
+from prophet_mesh.pkg_workspace import (
+    adapt_contact,
+    adapt_event,
+    adapt_message,
+    ingest_workspace,
+)
+
+SPEC = load_spec()
+
+CONTACT_MOM = {
+    "schemaVersion": "v0.1", "contactId": "c-mom", "accountRef": "acct-1",
+    "contactClass": "person", "displayName": "Mom", "givenName": "Ada",
+    "labels": ["family"], "sourceAdapter": "google_contacts",
+    "externalId": "people/c123", "updatedAt": "2026-06-28T14:22:00Z",
+    "socialProfiles": [{"platform": "linkedin", "handle": "ada", "url": "https://linkedin.com/in/ada"}],
+}
+CONTACT_JAMIE = {
+    "schemaVersion": "v0.1", "contactId": "c-jamie", "accountRef": "acct-1",
+    "contactClass": "person", "displayName": "Jamie", "organizationRef": "Acme Music",
+    "labels": ["bandmate"], "sourceAdapter": "carddav", "updatedAt": "2026-06-01T00:00:00Z",
+}
+EVENT_PRACTICE = {
+    "schemaVersion": "v0.1", "eventId": "e-practice", "calendarRef": "cal-1",
+    "title": "Band practice", "status": "confirmed", "organizerRef": "me@x.com",
+    "start": {"dateTime": "2026-07-15T18:00:00"}, "end": {"dateTime": "2026-07-15T20:00:00"},
+    "sourceAdapter": "google_calendar", "updatedAt": "2026-07-10T00:00:00Z",
+    "attendees": [
+        {"email": "me@x.com", "isSelf": True, "isOrganizer": True},
+        {"email": "jamie@x.com", "name": "Jamie", "contactRef": "c-jamie", "responseStatus": "accepted"},
+    ],
+}
+MAIL_FROM_JAMIE = {
+    "schemaVersion": "v0.1", "messageId": "m-1", "accountRef": "acct-mail-1", "threadId": "t-1",
+    "from": {"email": "jamie@x.com", "name": "Jamie", "contactRef": "c-jamie"},
+    "to": [{"email": "me@x.com", "isSelf": True}], "subject": "setlist",
+    "receivedAt": "2026-06-28T09:45:00Z", "routingState": "imbox", "inboxSlot": "primary",
+}
+ARTIFACT_SETLIST = {
+    "schemaVersion": "v0.1", "artifactId": "a-setlist", "workroomId": "wr-band",
+    "artifactType": "document", "title": "Setlist", "format": "docx", "status": "approved",
+    "createdAt": "2026-06-20T00:00:00Z",
+}
+
+
+def test_contact_family_hint_becomes_relatedTo():
+    g = seed_graph()
+    nid = adapt_contact(g, CONTACT_MOM)
+    rel = {(e.src, e.dst): e.rel for e in g.edges}[("self", nid)]
+    assert rel == "relatedTo"
+    # sourceAdapter flows into provenance; social profile is a reference-only link
+    n = g.nodes[nid]
+    assert n.provenance.source == "workspace:contacts:google_contacts"
+    assert n.provenance.captured_at == "2026-06-28T14:22:00Z"
+    assert n.external[0].target_kg == "social_network"
+    assert validate(g, SPEC) == []
+
+
+def test_contact_with_org_makes_worksAt():
+    g = seed_graph()
+    nid = adapt_contact(g, CONTACT_JAMIE)
+    rels = {(e.src, e.dst): e.rel for e in g.edges}
+    assert rels[("self", nid)] == "knows"
+    assert rels[(nid, "org:Acme Music")] == "worksAt"
+    assert g.nodes["org:Acme Music"].type == "Organization"
+    assert validate(g, SPEC) == []
+
+
+def test_event_attendee_with_contactRef_participates():
+    g = seed_graph()
+    adapt_event(g, EVENT_PRACTICE)
+    rels = {(e.src, e.dst): e.rel for e in g.edges}
+    assert rels[("self", "event:e-practice")] == "participatedIn"
+    assert rels[("person:c-jamie", "event:e-practice")] == "participatedIn"  # attendee, not self
+    assert validate(g, SPEC) == []
+
+
+def test_mail_from_contact_is_communicatedWith():
+    g = seed_graph()
+    pid = adapt_message(g, MAIL_FROM_JAMIE)
+    assert pid == "person:c-jamie"
+    e = g.edges[-1]
+    assert (e.src, e.dst, e.rel) == ("self", "person:c-jamie", "communicatedWith")
+    assert e.provenance.captured_at == "2026-06-28T09:45:00Z"  # receivedAt
+    assert validate(g, SPEC) == []
+
+
+def test_full_workspace_export_ingests_and_validates():
+    g = seed_graph()
+    ingest_workspace(
+        g,
+        contacts=[CONTACT_MOM, CONTACT_JAMIE],
+        events=[EVENT_PRACTICE],
+        messages=[MAIL_FROM_JAMIE],
+        artifacts=[ARTIFACT_SETLIST],
+    )
+    assert validate(g, SPEC) == []
+    types = sorted({n.type for n in g.nodes.values()})
+    assert types == ["Document", "Event", "Organization", "Person", "Self"]
+    # mail resolved onto the SAME Jamie node created from contacts (no dup)
+    assert sum(1 for n in g.nodes.values() if n.label == "Jamie") == 1