diff --git a/Makefile b/Makefile
index bbe4e41..d7d026f 100644
--- a/Makefile
+++ b/Makefile
@@ -7,8 +7,9 @@ validate-upstreams:
 	$(PYTHON) scripts/render_import_plan.py third_party/upstreams.lock.yaml --output import-plan.json
 
 validate-python:
-	$(PYTHON) -m py_compile services/memoryd/app/*.py adapters/litellm/*.py scripts/*.py
+	$(PYTHON) -m py_compile services/memoryd/app/*.py services/workspace_ingestion/app/*.py adapters/litellm/*.py scripts/*.py
 	$(PYTHON) -m unittest discover -s services/memoryd/tests -p 'test_*.py'
+	$(PYTHON) -m unittest discover -s services/workspace_ingestion/tests -p 'test_*.py'
 
 validate-deploy-assets:
 	$(PYTHON) scripts/validate_deploy_assets.py
diff --git a/docs/content-memory-and-recall.md b/docs/content-memory-and-recall.md
new file mode 100644
index 0000000..0313f0e
--- /dev/null
+++ b/docs/content-memory-and-recall.md
@@ -0,0 +1,128 @@
+# Content, Support, and Ops Memory Loop Design
+
+## Purpose
+
+This document defines how `memory-mesh` participates in the governed control loop for content, support, premium support, query orchestration, and operations-domain intelligence.
+
+`memory-mesh` is the long-horizon retained memory layer. It does not replace the query plane, ontology plane, or ops-domain intelligence plane. It retains history, prior decisions, prior outcomes, accepted and rejected recommendations, reusable context, and recall-before-change signals.
+
+## Repository role
+
+`memory-mesh` owns:
+
+- recall-before-call / recall-before-change runtime seams
+- writeback-after-call / writeback-after-outcome seams
+- retained support, query, and ops-context history
+- prior-case linkage and repeated-resolution memory
+- optional vector retrieval and deterministic memory embeddings for bring-up
+
+It does not own:
+
+- semantic meaning of memory classes (`ontogenesis`)
+- base storage standards (`socioprophet-standards-storage`)
+- ops-domain normalization of raw logs and anomalies (`global-devsecops-intelligence`)
+- canonical query orchestration (`sherlock-search`)
+
+## Core memory objects
+
+`memory-mesh` should retain, at minimum, typed records linked to the upstream semantic classes.
+
+### `MemoryRecord`
+Canonical retained unit for prior context.
+
+### `SupportInteractionMemory`
+Retained support interaction history, linked to support tier, domain, category, outcome, and evidence.
+
+### `AssetUsageMemory`
+History of which assets, templates, or runbooks were used for what task, under what support tier, with what result.
+
+### `RecommendationHistory`
+Accepted, rejected, ignored, or superseded recommendations, including premium-support recommendations and ops recommendations.
+
+### `IncidentResolutionMemory`
+Prior anomaly, incident-story, and ticket-resolution history derived from normalized ops-domain intelligence.
+
+### `OverlayMemory`
+Premium-overlay, tenant-overlay, or customer-specific conventions, exclusions, prior decisions, vocabulary, and local preferences.
+
+### `LearningLinkMemory`
+Memory handles linking retained history to learning objectives and curriculum objects in `alexandrian-academy`.
+
+## Required behaviors
+
+### Recall before change
+Before a support recommendation, content promotion, asset revision, or ops remediation is proposed, the runtime should be able to query `memory-mesh` for:
+
+- prior similar interactions
+- prior accepted or rejected recommendations
+- prior anomaly/remediation pairings
+- prior support-tier-specific overlays
+- prior objections or unresolved confusion signals
+
+### Writeback after outcome
+After an interaction or action completes, `memory-mesh` should persist:
+
+- outcome state
+- linked assets and evidence
+- recommendation acceptance or rejection
+- support tier and overlay context
+- operational context references when present
+- associated learning-objective references when present
+
+### Cross-plane linkability
+Memory records must be linkable to:
+
+- Sherlock query IDs and result sets
+- support interactions and escalation packets
+- incident stories and anomaly findings from `global-devsecops-intelligence`
+- learning objectives and curriculum objects from `alexandrian-academy`
+- action/replay/evidence handles from `agentplane`
+
+## Retrieval lanes
+
+### Support recall
+Used by standard and premium support to retrieve similar cases, prior runbooks, prior objections, and prior successful resolution paths.
+
+### Query recall
+Used by `sherlock-search` to enrich query planning and result ranking with prior decision context.
+
+### Ops recall
+Used by operational intelligence and support systems to remember repeated anomalies, recurring incident stories, and prior remediation outcomes.
+
+### Learning recall
+Used to feed learning-objective evaluation and explanation-quality refinement in `alexandrian-academy`.
+
+## Premium support implications
+
+Premium support should not fork the memory system. It should use the same memory substrate with explicit overlay and entitlement controls.
+
+Premium support memory may include:
+
+- tenant-specific incidents and support narratives
+- customer-approved workarounds and exclusions
+- local naming conventions and vocabulary
+- TAM / SME handoff history
+- premium-only overlay assets and resolution patterns
+
+These overlays must remain policy-scoped and traceable.
+
+## Integration map
+
+- `ontogenesis`: semantic meaning of `MemoryRecord` and related classes
+- `socioprophet-standards-storage`: transport and storage invariants for memory-linked payloads
+- `sherlock-search`: query-plane consumer of memory context
+- `global-devsecops-intelligence`: producer of normalized ops-domain findings that can be written into retained memory
+- `alexandrian-academy`: producer/consumer of learning-linked memory context
+- `agentplane`: producer of replay/evidence handles attached to memory writebacks
+- `prophet-platform`: runtime and API host for memory-aware workflows
+
+## Immediate implementation tranche
+
+1. Define the retained-memory object families and their typed references.
+2. Add recall-before-change guidance for support and query workflows.
+3. Add writeback-after-outcome guidance for support, ops, and content workflows.
+4. Ensure memory records preserve support tier, overlay provenance, and evidence linkage.
+
+## Outcome
+
+When implemented correctly, `memory-mesh` becomes the retained memory substrate that lets support, premium support, query orchestration, and ops intelligence learn from prior outcomes instead of operating as stateless loops.
diff --git a/services/memoryd/app/artifact_api.py b/services/memoryd/app/artifact_api.py
new file mode 100644
index 0000000..7fdf555
--- /dev/null
+++ b/services/memoryd/app/artifact_api.py
@@ -0,0 +1,611 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import sqlite3
+from pathlib import Path
+from typing import Any, Callable, Protocol
+from urllib.parse import urlparse
+from uuid import uuid4
+
+try:
+    import psycopg
+except Exception:  # pragma: no cover
+    psycopg = None  # type: ignore[assignment]
+
+from fastapi import FastAPI, Header, HTTPException
+from pydantic import BaseModel, Field
+
+from .models import EventRecord, ScopeEnvelope, dump_model, stable_object_hash, utcnow
+
+
+class ArtifactRecord(BaseModel):
+    artifact_id: str
+    artifact_type: str
+    session_id: str
+    sequence: int
+    artifact_hash: str
+    parent_artifact_hash: str | None = None
+    idempotency_key: str | None = None
+    envelope: dict[str, Any]
+    payload: dict[str, Any]
+    tags: list[str] = Field(default_factory=list)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    created_at: str
+
+
+class ArtifactWriteRequest(BaseModel):
+    envelope: ScopeEnvelope
+    artifact_type: str
+    session_id: str
+    payload: dict[str, Any]
+    tags: list[str] = Field(default_factory=list)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    idempotency_key: str | None = None
+    parent_artifact_hash: str | None = None
+
+
+class ArtifactWriteResponse(BaseModel):
+    record: ArtifactRecord
+    event_id: str | None = None
+
+
+class ArtifactGetResponse(BaseModel):
+    found: bool
+    record: ArtifactRecord | None = None
+
+
+class ArtifactListRequest(BaseModel):
+    envelope: ScopeEnvelope
+    session_id: str | None = None
+    artifact_type: str | None = None
+    limit: int = Field(default=50, ge=1, le=500)
+
+
+class ArtifactListResponse(BaseModel):
+    items: list[ArtifactRecord]
+
+
+class ArtifactLatestRequest(BaseModel):
+    envelope: ScopeEnvelope
+    session_id: str
+    artifact_type: str | None = None
+
+
+class ArtifactLatestResponse(BaseModel):
+    found: bool
+    record: ArtifactRecord | None = None
+
+
+class ArtifactReplaySessionRequest(BaseModel):
+    envelope: ScopeEnvelope
+    session_id: str
+    limit: int = Field(default=500, ge=1, le=5000)
+
+
+class ArtifactReplaySessionResponse(BaseModel):
+    session_id: str
+    items: list[ArtifactRecord]
+
+
+class ArtifactStoreProtocol(Protocol):
+    async def init(self) -> None: ...
+    async def close(self) -> None: ...
+    async def write(self, request: ArtifactWriteRequest) -> ArtifactRecord: ...
+    async def get(self, artifact_id: str) -> ArtifactRecord | None: ...
+    async def list(self, request: ArtifactListRequest) -> list[ArtifactRecord]: ...
+    async def latest(self, request: ArtifactLatestRequest) -> ArtifactRecord | None: ...
+    async def replay_session(self, request: ArtifactReplaySessionRequest) -> list[ArtifactRecord]: ...
+
+
+def artifact_scope_key(envelope: dict[str, Any], session_id: str) -> tuple[str, str, str | None, str | None]:
+    return (
+        str(envelope.get('workload_id') or ''),
+        session_id,
+        envelope.get('workspace_id'),
+        envelope.get('user_id'),
+    )
+
+
+def artifact_matches_envelope(record: ArtifactRecord, envelope: ScopeEnvelope) -> bool:
+    env = record.envelope
+    if env.get('workload_id') != envelope.workload_id:
+        return False
+    if envelope.workspace_id is not None and env.get('workspace_id') != envelope.workspace_id:
+        return False
+    if env.get('user_id') != envelope.user_id:
+        return False
+    return True
+
+
+def build_artifact_record(
+    *,
+    request: ArtifactWriteRequest,
+    sequence: int,
+    parent_artifact_hash: str | None,
+) -> ArtifactRecord:
+    payload_hash = stable_object_hash(request.payload)
+    envelope = dump_model(request.envelope)
+    artifact_id = uuid4().hex
+    return ArtifactRecord(
+        artifact_id=artifact_id,
+        artifact_type=request.artifact_type,
+        session_id=request.session_id,
+        sequence=sequence,
+        artifact_hash=payload_hash,
+        parent_artifact_hash=parent_artifact_hash,
+        idempotency_key=request.idempotency_key,
+        envelope=envelope,
+        payload=dict(request.payload),
+        tags=list(request.tags),
+        metadata=dict(request.metadata),
+        created_at=utcnow().isoformat(),
+    )
+
+
+class InMemoryArtifactStore:
+    def __init__(self) -> None:
+        self._items: dict[str, ArtifactRecord] = {}
+
+    async def init(self) -> None:
+        return None
+
+    async def close(self) -> None:
+        return None
+
+    async def write(self, request: ArtifactWriteRequest) -> ArtifactRecord:
+        if request.idempotency_key:
+            for record in self._items.values():
+                if (
+                    record.idempotency_key == request.idempotency_key
+                    and record.session_id == request.session_id
+                    and artifact_matches_envelope(record, request.envelope)
+                ):
+                    return record
+        previous = await self.latest(
+            ArtifactLatestRequest(
+                envelope=request.envelope,
+                session_id=request.session_id,
+                artifact_type=None,
+            )
+        )
+        next_sequence = (previous.sequence + 1) if previous else 1
+        parent_hash = request.parent_artifact_hash or (previous.artifact_hash if previous else None)
+        record = build_artifact_record(request=request, sequence=next_sequence, parent_artifact_hash=parent_hash)
+        self._items[record.artifact_id] = record
+        return record
+
+    async def get(self, artifact_id: str) -> ArtifactRecord | None:
+        return self._items.get(artifact_id)
+
+    async def list(self, request: ArtifactListRequest) -> list[ArtifactRecord]:
+        items: list[ArtifactRecord] = []
+        for record in self._items.values():
+            if not artifact_matches_envelope(record, request.envelope):
+                continue
+            if request.session_id and record.session_id != request.session_id:
+                continue
+            if request.artifact_type and record.artifact_type != request.artifact_type:
+                continue
+            items.append(record)
+        items.sort(key=lambda item: (item.sequence, item.created_at), reverse=True)
+        return items[: request.limit]
+
+    async def latest(self, request: ArtifactLatestRequest) -> ArtifactRecord | None:
+        items = await self.list(
+            ArtifactListRequest(
+                envelope=request.envelope,
+                session_id=request.session_id,
+                artifact_type=request.artifact_type,
+                limit=1,
+            )
+        )
+        return items[0] if items else None
+
+    async def replay_session(self, request: ArtifactReplaySessionRequest) -> list[ArtifactRecord]:
+        items = await self.list(
+            ArtifactListRequest(
+                envelope=request.envelope,
+                session_id=request.session_id,
+                artifact_type=None,
+                limit=request.limit,
+            )
+        )
+        return list(reversed(items))
+
+
+class SQLiteArtifactStore:
+    def __init__(self, db_path: str) -> None:
+        self.db_path = db_path
+
+    async def init(self) -> None:
+        await asyncio.to_thread(self._ensure_parent_dir)
+        await asyncio.to_thread(self._migrate)
+
+    async def close(self) -> None:
+        return None
+
+    def _ensure_parent_dir(self) -> None:
+        Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
+
+    def _connect(self) -> sqlite3.Connection:
+        conn = sqlite3.connect(self.db_path)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    def _migrate(self) -> None:
+        with self._connect() as conn:
+            conn.executescript(
+                '''
+                CREATE TABLE IF NOT EXISTS artifacts (
+                    artifact_id TEXT PRIMARY KEY,
+                    artifact_type TEXT NOT NULL,
+                    session_id TEXT NOT NULL,
+                    sequence INTEGER NOT NULL,
+                    artifact_hash TEXT NOT NULL,
+                    parent_artifact_hash TEXT,
+                    idempotency_key TEXT,
+                    envelope_json TEXT NOT NULL,
+                    payload_json TEXT NOT NULL,
+                    tags_json TEXT NOT NULL,
+                    metadata_json TEXT NOT NULL,
+                    created_at TEXT NOT NULL
+                );
+                CREATE INDEX IF NOT EXISTS idx_artifacts_scope ON artifacts(session_id, artifact_type, sequence DESC);
+                CREATE UNIQUE INDEX IF NOT EXISTS idx_artifacts_idempotency ON artifacts(session_id, idempotency_key) WHERE idempotency_key IS NOT NULL;
+                '''
+            )
+            conn.commit()
+
+    async def write(self, request: ArtifactWriteRequest) -> ArtifactRecord:
+        return await asyncio.to_thread(self._write_sync, request)
+
+    def _write_sync(self, request: ArtifactWriteRequest) -> ArtifactRecord:
+        if request.idempotency_key:
+            existing = self._get_by_idempotency_sync(request.session_id, request.idempotency_key, request.envelope)
+            if existing is not None:
+                return existing
+        previous = self._latest_sync(request.envelope, request.session_id, None)
+        next_sequence = (previous.sequence + 1) if previous else 1
+        parent_hash = request.parent_artifact_hash or (previous.artifact_hash if previous else None)
+        record = build_artifact_record(request=request, sequence=next_sequence, parent_artifact_hash=parent_hash)
+        with self._connect() as conn:
+            conn.execute(
+                '''
+                INSERT INTO artifacts(
+                    artifact_id, artifact_type, session_id, sequence, artifact_hash,
+                    parent_artifact_hash, idempotency_key, envelope_json, payload_json,
+                    tags_json, metadata_json, created_at
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ''',
+                (
+                    record.artifact_id,
+                    record.artifact_type,
+                    record.session_id,
+                    record.sequence,
+                    record.artifact_hash,
+                    record.parent_artifact_hash,
+                    record.idempotency_key,
+                    json.dumps(record.envelope, sort_keys=True),
+                    json.dumps(record.payload, sort_keys=True),
+                    json.dumps(record.tags),
+                    json.dumps(record.metadata, sort_keys=True),
+                    record.created_at,
+                ),
+            )
+            conn.commit()
+        return record
+
+    def _row_to_record(self, row: sqlite3.Row) -> ArtifactRecord:
+        return ArtifactRecord(
+            artifact_id=row['artifact_id'],
+            artifact_type=row['artifact_type'],
+            session_id=row['session_id'],
+            sequence=int(row['sequence']),
+            artifact_hash=row['artifact_hash'],
+            parent_artifact_hash=row['parent_artifact_hash'],
+            idempotency_key=row['idempotency_key'],
+            envelope=json.loads(row['envelope_json']),
+            payload=json.loads(row['payload_json']),
+            tags=list(json.loads(row['tags_json'])),
+            metadata=dict(json.loads(row['metadata_json'])),
+            created_at=row['created_at'],
+        )
+
+    def _get_by_idempotency_sync(self, session_id: str, idempotency_key: str, envelope: ScopeEnvelope) -> ArtifactRecord | None:
+        with self._connect() as conn:
+            row = conn.execute(
+                'SELECT * FROM artifacts WHERE session_id = ? AND idempotency_key = ? LIMIT 1',
+                (session_id, idempotency_key),
+            ).fetchone()
+        if row is None:
+            return None
+        record = self._row_to_record(row)
+        return record if artifact_matches_envelope(record, envelope) else None
+
+    async def get(self, artifact_id: str) -> ArtifactRecord | None:
+        return await asyncio.to_thread(self._get_sync, artifact_id)
+
+    def _get_sync(self, artifact_id: str) -> ArtifactRecord | None:
+        with self._connect() as conn:
+            row = conn.execute('SELECT * FROM artifacts WHERE artifact_id = ?', (artifact_id,)).fetchone()
+        return self._row_to_record(row) if row else None
+
+    async def list(self, request: ArtifactListRequest) -> list[ArtifactRecord]:
+        return await asyncio.to_thread(self._list_sync, request)
+
+    def _list_sync(self, request: ArtifactListRequest) -> list[ArtifactRecord]:
+        with self._connect() as conn:
+            rows = conn.execute(
+                'SELECT * FROM artifacts ORDER BY sequence DESC, created_at DESC LIMIT ?',
+                (max(request.limit * 5, request.limit),),
+            ).fetchall()
+        items: list[ArtifactRecord] = []
+        for row in rows:
+            record = self._row_to_record(row)
+            if not artifact_matches_envelope(record, request.envelope):
+                continue
+            if request.session_id and record.session_id != request.session_id:
+                continue
+            if request.artifact_type and record.artifact_type != request.artifact_type:
+                continue
+            items.append(record)
+            if len(items) >= request.limit:
+                break
+        return items
+
+    async def latest(self, request: ArtifactLatestRequest) -> ArtifactRecord | None:
+        return await asyncio.to_thread(self._latest_sync, request.envelope, request.session_id, request.artifact_type)
+
+    def _latest_sync(self, envelope: ScopeEnvelope, session_id: str, artifact_type: str | None) -> ArtifactRecord | None:
+        with self._connect() as conn:
+            rows = conn.execute(
+                'SELECT * FROM artifacts WHERE session_id = ? ORDER BY sequence DESC, created_at DESC LIMIT 100',
+                (session_id,),
+            ).fetchall()
+        for row in rows:
+            record = self._row_to_record(row)
+            if artifact_type and record.artifact_type != artifact_type:
+                continue
+            if artifact_matches_envelope(record, envelope):
+                return record
+        return None
+
+    async def replay_session(self, request: ArtifactReplaySessionRequest) -> list[ArtifactRecord]:
+        items = await self.list(
+            ArtifactListRequest(
+                envelope=request.envelope,
+                session_id=request.session_id,
+                artifact_type=None,
+                limit=request.limit,
+            )
+        )
+        return list(reversed(items))
+
+
+class PostgresArtifactStore:
+    def __init__(self, dsn: str, schema: str = 'memorymesh') -> None:
+        if psycopg is None:
+            raise RuntimeError('psycopg is required for PostgresArtifactStore')
+        self.dsn = dsn
+        self.schema = schema
+        self._memory_fallback = InMemoryArtifactStore()
+
+    async def init(self) -> None:
+        await asyncio.to_thread(self._migrate)
+
+    async def close(self) -> None:
+        return None
+
+    def _connect(self):
+        return psycopg.connect(self.dsn)
+
+    def _migrate(self) -> None:
+        with self._connect() as conn:
+            with conn.cursor() as cur:
+                cur.execute(f'CREATE SCHEMA IF NOT EXISTS {self.schema}')
+                cur.execute(
+                    f'''
+                    CREATE TABLE IF NOT EXISTS {self.schema}.artifacts (
+                        artifact_id TEXT PRIMARY KEY,
+                        artifact_type TEXT NOT NULL,
+                        session_id TEXT NOT NULL,
+                        sequence INTEGER NOT NULL,
+                        artifact_hash TEXT NOT NULL,
+                        parent_artifact_hash TEXT,
+                        idempotency_key TEXT,
+                        envelope JSONB NOT NULL,
+                        payload JSONB NOT NULL,
+                        tags JSONB NOT NULL,
+                        metadata JSONB NOT NULL,
+                        created_at TEXT NOT NULL
+                    )
+                    '''
+                )
+                cur.execute(
+                    f'CREATE INDEX IF NOT EXISTS idx_{self.schema}_artifacts_scope ON {self.schema}.artifacts(session_id, artifact_type, sequence DESC)'
+                )
+            conn.commit()
+
+    async def write(self, request: ArtifactWriteRequest) -> ArtifactRecord:
+        # Keep Postgres behavior conservative for the bootstrap by delegating selection
+        # logic through the same in-memory semantics if SQL feature drift occurs.
+        return await asyncio.to_thread(self._write_sync, request)
+
+    def _write_sync(self, request: ArtifactWriteRequest) -> ArtifactRecord:
+        previous = self._latest_sync(request.envelope, request.session_id, None)
+        next_sequence = (previous.sequence + 1) if previous else 1
+        parent_hash = request.parent_artifact_hash or (previous.artifact_hash if previous else None)
+        record = build_artifact_record(request=request, sequence=next_sequence, parent_artifact_hash=parent_hash)
+        with self._connect() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f'''
+                    INSERT INTO {self.schema}.artifacts(
+                        artifact_id, artifact_type, session_id, sequence, artifact_hash,
+                        parent_artifact_hash, idempotency_key, envelope, payload,
+                        tags, metadata, created_at
+                    ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s::jsonb, %s::jsonb, %s::jsonb, %s::jsonb, %s)
+                    ON CONFLICT (artifact_id) DO NOTHING
+                    ''',
+                    (
+                        record.artifact_id,
+                        record.artifact_type,
+                        record.session_id,
+                        record.sequence,
+                        record.artifact_hash,
+                        record.parent_artifact_hash,
+                        record.idempotency_key,
+                        json.dumps(record.envelope, sort_keys=True),
+                        json.dumps(record.payload, sort_keys=True),
+                        json.dumps(record.tags),
+                        json.dumps(record.metadata, sort_keys=True),
+                        record.created_at,
+                    ),
+                )
+            conn.commit()
+        return record
+
+    def _row_to_record(self, row: tuple[Any, ...]) -> ArtifactRecord:
+        return ArtifactRecord(
+            artifact_id=row[0],
+            artifact_type=row[1],
+            session_id=row[2],
+            sequence=int(row[3]),
+            artifact_hash=row[4],
+            parent_artifact_hash=row[5],
+            idempotency_key=row[6],
+            envelope=dict(row[7]),
+            payload=dict(row[8]),
+            tags=list(row[9] or []),
+            metadata=dict(row[10] or {}),
+            created_at=row[11],
+        )
+
+    async def get(self, artifact_id: str) -> ArtifactRecord | None:
+        return await asyncio.to_thread(self._get_sync, artifact_id)
+
+    def _get_sync(self, artifact_id: str) -> ArtifactRecord | None:
+        with self._connect() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f'SELECT artifact_id, artifact_type, session_id, sequence, artifact_hash, parent_artifact_hash, idempotency_key, envelope, payload, tags, metadata, created_at FROM {self.schema}.artifacts WHERE artifact_id = %s',
+                    (artifact_id,),
+                )
+                row = cur.fetchone()
+        return self._row_to_record(row) if row else None
+
+    async def list(self, request: ArtifactListRequest) -> list[ArtifactRecord]:
+        return await asyncio.to_thread(self._list_sync, request)
+
+    def _list_sync(self, request: ArtifactListRequest) -> list[ArtifactRecord]:
+        with self._connect() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f'SELECT artifact_id, artifact_type, session_id, sequence, artifact_hash, parent_artifact_hash, idempotency_key, envelope, payload, tags, metadata, created_at FROM {self.schema}.artifacts ORDER BY sequence DESC LIMIT %s',
+                    (max(request.limit * 5, request.limit),),
+                )
+                rows = cur.fetchall()
+        items: list[ArtifactRecord] = []
+        for row in rows:
+            record = self._row_to_record(row)
+            if not artifact_matches_envelope(record, request.envelope):
+                continue
+            if request.session_id and record.session_id != request.session_id:
+                continue
+            if request.artifact_type and record.artifact_type != request.artifact_type:
+                continue
+            items.append(record)
+            if len(items) >= request.limit:
+                break
+        return items
+
+    async def latest(self, request: ArtifactLatestRequest) -> ArtifactRecord | None:
+        return await asyncio.to_thread(self._latest_sync, request.envelope, request.session_id, request.artifact_type)
+
+    def _latest_sync(self, envelope: ScopeEnvelope, session_id: str, artifact_type: str | None) -> ArtifactRecord | None:
+        with self._connect() as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f'SELECT artifact_id, artifact_type, session_id, sequence, artifact_hash, parent_artifact_hash, idempotency_key, envelope, payload, tags, metadata, created_at FROM {self.schema}.artifacts WHERE session_id = %s ORDER BY sequence DESC LIMIT 100',
+                    (session_id,),
+                )
+                rows = cur.fetchall()
+        for row in rows:
+            record = self._row_to_record(row)
+            if artifact_type and record.artifact_type != artifact_type:
+                continue
+            if artifact_matches_envelope(record, envelope):
+                return record
+        return None
+
+    async def replay_session(self, request: ArtifactReplaySessionRequest) -> list[ArtifactRecord]:
+        items = await self.list(
+            ArtifactListRequest(
+                envelope=request.envelope,
+                session_id=request.session_id,
+                artifact_type=None,
+                limit=request.limit,
+            )
+        )
+        return list(reversed(items))
+
+
+def build_artifact_store(store_uri: str) -> ArtifactStoreProtocol:
+    if not store_uri or store_uri == 'memory://':
+        return InMemoryArtifactStore()
+    if store_uri.startswith('sqlite:///'):
+        return SQLiteArtifactStore(urlparse(store_uri).path)
+    if store_uri.startswith('postgresql://') or store_uri.startswith('postgres://'):
+        return PostgresArtifactStore(store_uri)
+    return InMemoryArtifactStore()
+
+
+def install_artifact_routes(
+    app: FastAPI,
+    *,
+    artifact_store: ArtifactStoreProtocol,
+    event_store: Any,
+    require_api_key: Callable[[str | None], Any],
+) -> None:
+    @app.post('/v1/artifacts/write', response_model=ArtifactWriteResponse)
+    async def artifact_write(request: ArtifactWriteRequest, x_api_key: str | None = Header(default=None)) -> ArtifactWriteResponse:
+        await require_api_key(x_api_key)
+        record = await artifact_store.write(request)
+        event: EventRecord = await event_store.append_event(
+            'memory.artifact.written',
+            {
+                'envelope': dump_model(request.envelope),
+                'artifact_id': record.artifact_id,
+                'artifact_type': record.artifact_type,
+                'session_id': record.session_id,
+                'sequence': record.sequence,
+                'artifact_hash': record.artifact_hash,
+            },
+        )
+        return ArtifactWriteResponse(record=record, event_id=event.event_id)
+
+    @app.get('/v1/artifacts/{artifact_id}', response_model=ArtifactGetResponse)
+    async def artifact_get(artifact_id: str, x_api_key: str | None = Header(default=None)) -> ArtifactGetResponse:
+        await require_api_key(x_api_key)
+        record = await artifact_store.get(artifact_id)
+        return ArtifactGetResponse(found=record is not None, record=record)
+
+    @app.post('/v1/artifacts/list', response_model=ArtifactListResponse)
+    async def artifact_list(request: ArtifactListRequest, x_api_key: str | None = Header(default=None)) -> ArtifactListResponse:
+        await require_api_key(x_api_key)
+        return ArtifactListResponse(items=await artifact_store.list(request))
+
+    @app.post('/v1/artifacts/latest', response_model=ArtifactLatestResponse)
+    async def artifact_latest(request: ArtifactLatestRequest, x_api_key: str | None = Header(default=None)) -> ArtifactLatestResponse:
+        await require_api_key(x_api_key)
+        record = await artifact_store.latest(request)
+        return ArtifactLatestResponse(found=record is not None, record=record)
+
+    @app.post('/v1/artifacts/replay-session', response_model=ArtifactReplaySessionResponse)
+    async def artifact_replay_session(request: ArtifactReplaySessionRequest, x_api_key: str | None = Header(default=None)) -> ArtifactReplaySessionResponse:
+        await require_api_key(x_api_key)
+        return ArtifactReplaySessionResponse(
+            session_id=request.session_id,
+            items=await artifact_store.replay_session(request),
+        )
diff --git a/services/workspace_ingestion/README.md b/services/workspace_ingestion/README.md
new file mode 100644
index 0000000..0088fc3
--- /dev/null
+++ b/services/workspace_ingestion/README.md
@@ -0,0 +1,41 @@
+# workspace_ingestion — Personal Knowledge Graph runtime
+
+Turns prophet-workspace canonical objects into the person-graph on the managed
+HellGraph. This is the **runtime** half of the PKG program (the contract lives in
+prophet-workspace as `PersonalContextGraph`; the substrate is the managed
+HellGraph service in prophet-platform).
+
+## Flow
+
+```
+WorkspaceSource records            mapping.py            HellGraph (managed)
+(contacts/calendar/mail/office) ─▶ build_graph() ─▶ /cskg/normalize ─▶ /v1/ingest
+                                        │                                    │
+                                   CSKG nodes/edges              memoryd /v1/write
+                                   provenance-bound              (writeback summary)
+```
+
+- **`mapping.py`** — pure `WorkspaceSource → CSKGNode/CSKGEdge`. Family vs. social
+  relation from contact labels; `worksAt` from `organizationRef`; `participatedIn`
+  from event attendees; `communicatedWith` from mail; `authored` from artifacts.
+  Every element carries `provenance_refs` = the originating `workspace-source:` id.
+- **`hellgraph_client.py`** — `/cskg/normalize` (refine relations) → `/v1/ingest`
+  (persist as GraphNode/GraphEdge; workspace imports = `confirmed_relation`) →
+  `/v1/retract` (retention). Graceful-degrade when `HELLGRAPH_URL` is unset.
+- **`main.py`** — `POST /v1/ingest/workspace`, `POST /v1/retract`, `GET /healthz`.
+
+## Config
+
+| Env | Purpose |
+|---|---|
+| `HELLGRAPH_URL` | managed HellGraph base URL (e.g. `http://hellgraph:8850`); unset ⇒ dry-run |
+| `MEMORYD_BASE_URL` | memoryd for writeback summary; unset ⇒ skipped |
+| `WORKSPACE_INGESTION_REQUIRE_API_KEY` / `_API_KEY` | optional API-key gate |
+
+## Test
+
+```
+python -m unittest discover -s services/workspace_ingestion/tests -p 'test_*.py'
+```
+
+Runs without a live HellGraph/memoryd (clients graceful-degrade; service tests use fakes).
diff --git a/services/workspace_ingestion/app/__init__.py b/services/workspace_ingestion/app/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/workspace_ingestion/app/hellgraph_client.py b/services/workspace_ingestion/app/hellgraph_client.py
new file mode 100644
index 0000000..42bf2b5
--- /dev/null
+++ b/services/workspace_ingestion/app/hellgraph_client.py
@@ -0,0 +1,122 @@
+"""HellGraph client — writes the person-graph onto the managed graph service.
+
+Targets HellGraph's confirmed contract:
+  - POST {HELLGRAPH_URL}/cskg/normalize   → canonicalize raw relations to CSKGEdges
+  - POST {HELLGRAPH_URL}/v1/ingest         → persist {nodes, edges} (GraphNode/GraphEdge)
+  - POST {HELLGRAPH_URL}/v1/retract        → supersede edges derived from given sources
+
+Graceful-degrade like the other memory-mesh clients (mem0_client, MemoryMeshClient):
+if HELLGRAPH_URL is unset the client is disabled and returns an inert result, so
+the service runs (and tests pass) without a live graph. HellGraph's write path is
+its TS store façade (addNode/addEdge); the /v1/ingest route replays a bundle
+through it. The Rust kernel is never written directly.
+"""
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import httpx
+
+from .models import CSKGEdge, CSKGNode, GraphBundle
+
+
+class HellGraphClient:
+    def __init__(self, base_url: str | None = None, api_key: str | None = None, timeout_seconds: float = 10.0) -> None:
+        self.base_url = (base_url or os.getenv("HELLGRAPH_URL") or "").rstrip("/")
+        self.api_key = api_key or os.getenv("HELLGRAPH_API_KEY") or ""
+        self.timeout_seconds = timeout_seconds
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self.base_url)
+
+    @property
+    def headers(self) -> dict[str, str]:
+        headers = {"Content-Type": "application/json"}
+        if self.api_key:
+            headers["x-api-key"] = self.api_key
+        return headers
+
+    async def normalize(self, edges: list[CSKGEdge]) -> list[CSKGEdge]:
+        """Canonicalize relations through the CSKG normalizer. On absence/failure
+        return the edges unchanged (normalization is a refinement, not a gate)."""
+        if not self.enabled or not edges:
+            return edges
+        relations = [{"node1": e.node1, "relation": e.relation, "node2": e.node2,
+                      "provenance_ref": (e.provenance_refs[0] if e.provenance_refs else None)} for e in edges]
+        try:
+            async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+                resp = await client.post(f"{self.base_url}/cskg/normalize",
+                                         json={"relations": relations}, headers=self.headers)
+                resp.raise_for_status()
+                data = resp.json()
+        except Exception:
+            return edges
+        out: list[CSKGEdge] = []
+        for i, raw in enumerate(data.get("edges", [])):
+            base = edges[i] if i < len(edges) else None
+            out.append(CSKGEdge(
+                edge_id=raw.get("edge_id") or (base.edge_id if base else f"cskg-edge://{i}"),
+                node1=raw.get("node1", base.node1 if base else ""),
+                relation=raw.get("relation", base.relation if base else ""),
+                node2=raw.get("node2", base.node2 if base else ""),
+                provenance_refs=raw.get("provenance_refs") or (base.provenance_refs if base else []),
+                source_evidence_refs=raw.get("source_evidence_refs") or (base.source_evidence_refs if base else []),
+            ))
+        return out or edges
+
+    async def ingest(self, bundle: GraphBundle) -> dict[str, Any]:
+        """Persist the bundle (nodes then edges) onto HellGraph."""
+        if not self.enabled:
+            return {"persisted": False, "reason": "hellgraph disabled", "nodes": len(bundle.nodes), "edges": len(bundle.edges)}
+        payload = {
+            "nodes": [self._node_wire(n) for n in bundle.nodes],
+            "edges": [self._edge_wire(e) for e in bundle.edges],
+        }
+        async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+            resp = await client.post(f"{self.base_url}/v1/ingest", json=payload, headers=self.headers)
+            resp.raise_for_status()
+            return dict(resp.json())
+
+    async def retract(self, source_refs: list[str]) -> dict[str, Any]:
+        """Retention: supersede every edge derived from the given WorkspaceSources."""
+        if not self.enabled:
+            return {"retracted": False, "reason": "hellgraph disabled", "source_refs": source_refs}
+        async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+            resp = await client.post(f"{self.base_url}/v1/retract",
+                                     json={"provenance_refs": source_refs}, headers=self.headers)
+            resp.raise_for_status()
+            return dict(resp.json())
+
+    # ── wire shapes: CSKG → HellGraph GraphNode / GraphEdge ──────────────────
+    @staticmethod
+    def _node_wire(n: CSKGNode) -> dict[str, Any]:
+        return {
+            "id": n.node_id,
+            "labels": [n.node_type],
+            "properties": {
+                "label": n.label,
+                "memory_scope": n.memory_scope,
+                "provenance_refs": ",".join(n.provenance_refs),
+                "source_evidence_refs": ",".join(n.source_evidence_refs),
+                "external_projection_refs": ",".join(n.external_projection_refs),
+            },
+        }
+
+    @staticmethod
+    def _edge_wire(e: CSKGEdge) -> dict[str, Any]:
+        return {
+            "id": e.edge_id,
+            "label": e.relation,
+            "from": e.node1,
+            "to": e.node2,
+            "properties": {
+                # HellGraph-mandatory epistemic fields — workspace imports are confirmed:
+                "epistemicClass": "confirmed_relation",
+                "confidence": 1.0,
+                "promotionState": "confirmed",
+                "provenance_refs": ",".join(e.provenance_refs),
+                "source_evidence_refs": ",".join(e.source_evidence_refs),
+            },
+        }
diff --git a/services/workspace_ingestion/app/main.py b/services/workspace_ingestion/app/main.py
new file mode 100644
index 0000000..0d84355
--- /dev/null
+++ b/services/workspace_ingestion/app/main.py
@@ -0,0 +1,89 @@
+"""workspace_ingestion — the Personal Knowledge Graph ingestion service.
+
+Reads prophet-workspace canonical objects → CSKG nodes/edges → normalizes through
+HellGraph's CSKG normalizer → persists onto the managed HellGraph → records a
+writeback summary in memoryd. Also serves retention retraction. Owns the PKG
+runtime (contract = prophet-workspace PersonalContextGraph; substrate =
+prophet-platform managed HellGraph).
+"""
+from __future__ import annotations
+
+import os
+
+from fastapi import FastAPI, Header, HTTPException
+
+from .hellgraph_client import HellGraphClient
+from .mapping import build_graph
+from .memory_mesh_client import MemoryMeshClient
+from .models import (
+    IngestWorkspaceRequest,
+    IngestWorkspaceResponse,
+    RetractRequest,
+    RetractResponse,
+)
+
+REQUIRE_API_KEY = os.getenv("WORKSPACE_INGESTION_REQUIRE_API_KEY", "false").lower() in {"1", "true", "yes"}
+EXPECTED_API_KEY = os.getenv("WORKSPACE_INGESTION_API_KEY", "")
+
+app = FastAPI(title="workspace_ingestion", version="0.1.0")
+
+hellgraph = HellGraphClient(timeout_seconds=float(os.getenv("HELLGRAPH_TIMEOUT_SECONDS", "10")))
+memoryd = MemoryMeshClient(timeout_seconds=float(os.getenv("MEMORYD_TIMEOUT_SECONDS", "10")))
+
+
+async def require_api_key(x_api_key: str | None) -> None:
+    if not REQUIRE_API_KEY:
+        return
+    if not EXPECTED_API_KEY:
+        raise HTTPException(status_code=500, detail="api key enforcement enabled but WORKSPACE_INGESTION_API_KEY is empty")
+    if x_api_key != EXPECTED_API_KEY:
+        raise HTTPException(status_code=401, detail="invalid api key")
+
+
+@app.get("/healthz")
+async def healthz() -> dict[str, object]:
+    return {"status": "ok", "hellgraph_enabled": hellgraph.enabled, "memoryd_enabled": memoryd.enabled}
+
+
+@app.post("/v1/ingest/workspace", response_model=IngestWorkspaceResponse)
+async def ingest_workspace(request: IngestWorkspaceRequest, x_api_key: str | None = Header(default=None)) -> IngestWorkspaceResponse:
+    await require_api_key(x_api_key)
+    self_ref = request.self_ref or f"cskg-node://self/{request.envelope.user_id}"
+
+    bundle = build_graph(
+        self_ref, request.envelope.user_id,
+        contacts=request.contacts, events=request.events,
+        messages=request.messages, artifacts=request.artifacts,
+    )
+
+    # Refine relations through the CSKG normalizer (no-op if HellGraph is absent).
+    bundle.edges = await hellgraph.normalize(bundle.edges)
+    normalized = hellgraph.enabled
+
+    hg_result: dict[str, object] = {}
+    recall_ref: str | None = None
+    if request.persist:
+        hg_result = await hellgraph.ingest(bundle)
+        # Writeback-after-action: note the graph delta for recall.
+        source_refs = sorted({r for n in bundle.nodes for r in n.provenance_refs})
+        summary = await memoryd.write_summary(
+            envelope=request.envelope.model_dump(),
+            content={"event": "personal_context_graph.updated", "self_ref": self_ref,
+                     "nodes": len(bundle.nodes), "edges": len(bundle.edges), "source_refs": source_refs},
+            tags=["personal-knowledge-graph", "ingest"],
+            metadata={"self_ref": self_ref, "workload_id": request.envelope.workload_id},
+        )
+        recall_ref = summary.get("memory_id") if isinstance(summary, dict) else None
+
+    return IngestWorkspaceResponse(
+        self_ref=self_ref, bundle=bundle, normalized=normalized,
+        hellgraph=hg_result, recall_candidate_ref=recall_ref,
+    )
+
+
+@app.post("/v1/retract", response_model=RetractResponse)
+async def retract(request: RetractRequest, x_api_key: str | None = Header(default=None)) -> RetractResponse:
+    """Retention: a WorkspaceSource was deleted → retract its derived edges."""
+    await require_api_key(x_api_key)
+    result = await hellgraph.retract(request.source_refs)
+    return RetractResponse(requested_source_refs=request.source_refs, hellgraph=result)
diff --git a/services/workspace_ingestion/app/mapping.py b/services/workspace_ingestion/app/mapping.py
new file mode 100644
index 0000000..09488d9
--- /dev/null
+++ b/services/workspace_ingestion/app/mapping.py
@@ -0,0 +1,228 @@
+"""WorkspaceSource → CSKG mapping (the person-graph builder).
+
+Pure functions: canonical workspace records → CSKGNode/CSKGEdge, every element
+provenance-bound to the originating WorkspaceSource id. No IO here — the service
+layer normalizes + persists. Mirrors the prophet-mesh prototype adapters, but
+emits the CSKG wire shape and carries workspace-source provenance_refs.
+"""
+from __future__ import annotations
+
+import re
+from typing import Any
+
+from .models import CSKGEdge, CSKGNode, GraphBundle
+
+# 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 slug(value: str) -> str:
+    return re.sub(r"[^a-z0-9]+", "_", (value or "").strip().lower()).strip("_") or "x"
+
+
+def node_id(node_type: str, key: str) -> str:
+    return f"cskg-node://{node_type.lower()}/{slug(key)}"
+
+
+def edge_id(relation: str, n1: str, n2: str) -> str:
+    return f"cskg-edge://{relation}/{slug(n1)}-{slug(n2)}"
+
+
+def _source_ref(rec: dict[str, Any], surface: str, ident: str) -> str:
+    return rec.get("sourceId") or f"workspace-source:{surface}/{slug(ident)}"
+
+
+def _evidence(rec: dict[str, Any]) -> list[str]:
+    refs = rec.get("evidenceRefs") or []
+    corr = rec.get("evidenceCorrelationId")
+    if corr:
+        refs = [*refs, corr]
+    return list(refs)
+
+
+class GraphBuilder:
+    """Accumulates nodes (deduped by id) and edges for one person-graph."""
+
+    def __init__(self, self_ref: str) -> None:
+        self.self_ref = self_ref
+        self.nodes: dict[str, CSKGNode] = {}
+        self.edges: list[CSKGEdge] = []
+        self._edge_ids: set[str] = set()
+
+    def add_node(self, node: CSKGNode) -> None:
+        # First writer wins on type/label; merge provenance so a node seen from
+        # two sources carries both refs.
+        existing = self.nodes.get(node.node_id)
+        if existing is None:
+            self.nodes[node.node_id] = node
+        else:
+            existing.provenance_refs = sorted(set(existing.provenance_refs) | set(node.provenance_refs))
+            existing.source_evidence_refs = sorted(set(existing.source_evidence_refs) | set(node.source_evidence_refs))
+            existing.external_projection_refs = sorted(
+                set(existing.external_projection_refs) | set(node.external_projection_refs)
+            )
+
+    def add_edge(self, edge: CSKGEdge) -> None:
+        if edge.edge_id in self._edge_ids:
+            return
+        self._edge_ids.add(edge.edge_id)
+        self.edges.append(edge)
+
+    def bundle(self) -> GraphBundle:
+        return GraphBundle(nodes=list(self.nodes.values()), edges=self.edges)
+
+
+def seed_self(builder: GraphBuilder, user_id: str) -> None:
+    builder.add_node(CSKGNode(
+        node_id=builder.self_ref, node_type="Self", label="Self",
+        provenance_refs=["workspace-source:onboarding"],
+    ))
+
+
+def adapt_contact(builder: GraphBuilder, rec: dict[str, Any]) -> str:
+    cid = rec.get("contactId") or rec.get("displayName") or "unknown"
+    if rec.get("contactClass") == "organization":
+        oid = node_id("organization", cid)
+        builder.add_node(CSKGNode(
+            node_id=oid, node_type="Organization", label=rec.get("displayName") or cid,
+            provenance_refs=[_source_ref(rec, "contacts", cid)], source_evidence_refs=_evidence(rec),
+        ))
+        return oid
+
+    nid = node_id("person", cid)
+    label = rec.get("displayName") or " ".join(
+        p for p in (rec.get("givenName"), rec.get("familyName")) if p
+    ) or cid
+    src = _source_ref(rec, "contacts", cid)
+    ev = _evidence(rec)
+
+    projections = [f"projection:social/{slug(sp.get('handle') or sp.get('url') or 'x')}"
+                   for sp in (rec.get("socialProfiles") or [])]
+    builder.add_node(CSKGNode(
+        node_id=nid, node_type="Person", label=label,
+        provenance_refs=[src], source_evidence_refs=ev, external_projection_refs=projections,
+    ))
+
+    hints = {h.lower() for h in (rec.get("labels") or [])} | {h.lower() for h in (rec.get("groupRefs") or [])}
+    relation = "relatedTo" if hints & FAMILY_HINTS else "knows"
+    builder.add_edge(CSKGEdge(
+        edge_id=edge_id(relation, builder.self_ref, nid),
+        node1=builder.self_ref, relation=relation, node2=nid,
+        provenance_refs=[src], source_evidence_refs=ev,
+    ))
+
+    org_ref = rec.get("organizationRef")
+    if org_ref:
+        oid = node_id("organization", org_ref)
+        builder.add_node(CSKGNode(
+            node_id=oid, node_type="Organization", label=org_ref, provenance_refs=[src], source_evidence_refs=ev,
+        ))
+        builder.add_edge(CSKGEdge(
+            edge_id=edge_id("worksAt", nid, oid), node1=nid, relation="worksAt", node2=oid,
+            provenance_refs=[src], source_evidence_refs=ev,
+        ))
+    return nid
+
+
+def adapt_event(builder: GraphBuilder, rec: dict[str, Any]) -> str:
+    eid = rec.get("eventId") or rec.get("title") or "unknown"
+    nid = node_id("event", eid)
+    src = _source_ref(rec, "calendar", eid)
+    ev = _evidence(rec)
+    builder.add_node(CSKGNode(
+        node_id=nid, node_type="Event", label=rec.get("title") or eid,
+        provenance_refs=[src], source_evidence_refs=ev,
+    ))
+    builder.add_edge(CSKGEdge(
+        edge_id=edge_id("participatedIn", builder.self_ref, nid),
+        node1=builder.self_ref, relation="participatedIn", node2=nid,
+        provenance_refs=[src], source_evidence_refs=ev,
+    ))
+    for att in rec.get("attendees") or []:
+        if att.get("isSelf"):
+            continue
+        cref = att.get("contactRef")
+        if not cref:
+            continue
+        pid = node_id("person", cref)
+        if pid not in builder.nodes:
+            builder.add_node(CSKGNode(
+                node_id=pid, node_type="Person", label=att.get("name") or cref,
+                provenance_refs=[src], source_evidence_refs=ev,
+            ))
+        builder.add_edge(CSKGEdge(
+            edge_id=edge_id("participatedIn", pid, nid), node1=pid, relation="participatedIn", node2=nid,
+            provenance_refs=[src], source_evidence_refs=ev,
+        ))
+    return nid
+
+
+def adapt_message(builder: GraphBuilder, rec: dict[str, Any]) -> str | None:
+    frm = rec.get("from") or {}
+    if frm.get("isSelf"):
+        return None
+    mid = rec.get("messageId") or "unknown"
+    src = _source_ref(rec, "mail", rec.get("threadId") or mid)
+    ev = _evidence(rec)
+    cref = frm.get("contactRef")
+    if cref:
+        pid = node_id("person", cref)
+        label = frm.get("name") or cref
+    else:
+        email = frm.get("email", "unknown")
+        pid = node_id("person", f"email_{email}")
+        label = frm.get("name") or email
+    if pid not in builder.nodes:
+        builder.add_node(CSKGNode(
+            node_id=pid, node_type="Person", label=label, provenance_refs=[src], source_evidence_refs=ev,
+        ))
+    builder.add_edge(CSKGEdge(
+        edge_id=edge_id("communicatedWith", builder.self_ref, pid),
+        node1=builder.self_ref, relation="communicatedWith", node2=pid,
+        provenance_refs=[src], source_evidence_refs=ev,
+    ))
+    return pid
+
+
+def adapt_artifact(builder: GraphBuilder, rec: dict[str, Any]) -> str:
+    aid = rec.get("artifactId") or rec.get("title") or "unknown"
+    nid = node_id("document", aid)
+    src = _source_ref(rec, "office", aid)
+    ev = _evidence(rec)
+    builder.add_node(CSKGNode(
+        node_id=nid, node_type="Document", label=rec.get("title") or aid,
+        provenance_refs=[src], source_evidence_refs=ev,
+    ))
+    builder.add_edge(CSKGEdge(
+        edge_id=edge_id("authored", builder.self_ref, nid),
+        node1=builder.self_ref, relation="authored", node2=nid,
+        provenance_refs=[src], source_evidence_refs=ev,
+    ))
+    return nid
+
+
+def build_graph(
+    self_ref: str,
+    user_id: str,
+    *,
+    contacts: list[dict] | None = None,
+    events: list[dict] | None = None,
+    messages: list[dict] | None = None,
+    artifacts: list[dict] | None = None,
+) -> GraphBundle:
+    """Fold a workspace export into a person-graph. People first so calendar/mail
+    edges resolve onto existing Person nodes."""
+    builder = GraphBuilder(self_ref)
+    seed_self(builder, user_id)
+    for rec in contacts or []:
+        adapt_contact(builder, rec)
+    for rec in artifacts or []:
+        adapt_artifact(builder, rec)
+    for rec in events or []:
+        adapt_event(builder, rec)
+    for rec in messages or []:
+        adapt_message(builder, rec)
+    return builder.bundle()
diff --git a/services/workspace_ingestion/app/memory_mesh_client.py b/services/workspace_ingestion/app/memory_mesh_client.py
new file mode 100644
index 0000000..70ebeb3
--- /dev/null
+++ b/services/workspace_ingestion/app/memory_mesh_client.py
@@ -0,0 +1,48 @@
+"""Minimal memoryd client — records a recall candidate after an ingest.
+
+Mirrors services/finance_saa/app/memory_mesh_client.py. Writeback-after-action:
+once the person-graph is updated, drop a summary memory so recall-before-action
+can surface "the graph changed" to downstream agents. Graceful-degrade when
+MEMORYD_BASE_URL is unset.
+"""
+from __future__ import annotations
+
+import json
+import os
+from typing import Any
+
+import httpx
+
+
+class MemoryMeshClient:
+    def __init__(self, base_url: str | None = None, api_key: str | None = None, timeout_seconds: float = 10.0) -> None:
+        self.base_url = (base_url or os.getenv("MEMORYD_BASE_URL") or "").rstrip("/")
+        self.api_key = api_key or os.getenv("MEMORYD_API_KEY") or ""
+        self.timeout_seconds = timeout_seconds
+
+    @property
+    def enabled(self) -> bool:
+        return bool(self.base_url)
+
+    @property
+    def headers(self) -> dict[str, str]:
+        headers = {"Content-Type": "application/json"}
+        if self.api_key:
+            headers["x-api-key"] = self.api_key
+        return headers
+
+    async def write_summary(self, *, envelope: dict[str, Any], content: dict[str, Any], tags: list[str], metadata: dict[str, Any]) -> dict[str, Any]:
+        if not self.enabled:
+            return {"stored": False, "reason": "memoryd disabled"}
+        payload = {
+            "envelope": envelope,
+            "content": json.dumps(content, sort_keys=True, default=str),
+            "memory_class": "summary",
+            "persist_to_backend": True,
+            "metadata": metadata,
+            "tags": tags,
+        }
+        async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
+            resp = await client.post(f"{self.base_url}/v1/write", json=payload, headers=self.headers)
+            resp.raise_for_status()
+            return dict(resp.json())
diff --git a/services/workspace_ingestion/app/models.py b/services/workspace_ingestion/app/models.py
new file mode 100644
index 0000000..92ed980
--- /dev/null
+++ b/services/workspace_ingestion/app/models.py
@@ -0,0 +1,80 @@
+"""Pydantic models for the Personal Knowledge Graph ingestion service.
+
+The runtime that turns prophet-workspace canonical objects (WorkspaceSource:
+contacts / calendar / mail / office-artifact) into CSKG nodes + edges and writes
+them into the managed HellGraph. Edges conform to HellGraph's CSKGEdge shape
+{node1, relation, node2, provenance_refs, source_evidence_refs}. This realises the
+PersonalContextGraph contract (prophet-workspace) on the substrate
+(prophet-platform); memory-mesh owns the runtime.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+
+class ScopeEnvelope(BaseModel):
+    """Mirrors the memoryd ScopeEnvelope so recall/writeback share one scope."""
+    user_id: str
+    agent_id: str = "memory-steward"
+    run_id: str = "pkg-ingest"
+    workload_id: str = "personal-knowledge-graph"
+    workspace_id: str | None = None
+    source_interface: str = "prophet-workspace"
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class CSKGNode(BaseModel):
+    node_id: str
+    node_type: str                       # Self | Person | Place | Organization | Thing | Interest | Event | Document | Communication | Account
+    label: str
+    provenance_refs: list[str] = Field(default_factory=list)      # workspace-source:… ids (Layer-1 canonical objects)
+    source_evidence_refs: list[str] = Field(default_factory=list)
+    memory_scope: str = "relationship_context:approved"
+    external_projection_refs: list[str] = Field(default_factory=list)  # reference-only external-KG links (via ProviderProjection)
+
+
+class CSKGEdge(BaseModel):
+    edge_id: str
+    node1: str
+    relation: str                        # relatedTo | knows | worksAt | participatedIn | communicatedWith | authored | …
+    node2: str
+    provenance_refs: list[str] = Field(default_factory=list)
+    source_evidence_refs: list[str] = Field(default_factory=list)
+
+
+class GraphBundle(BaseModel):
+    nodes: list[CSKGNode] = Field(default_factory=list)
+    edges: list[CSKGEdge] = Field(default_factory=list)
+
+
+class IngestWorkspaceRequest(BaseModel):
+    """A workspace export to fold into the person-graph. Records are the canonical
+    contract shapes (lenient dicts — we consume known fields, ignore the rest)."""
+    envelope: ScopeEnvelope
+    self_ref: str | None = None          # defaults to cskg-node://self/<user_id>
+    contacts: list[dict[str, Any]] = Field(default_factory=list)
+    events: list[dict[str, Any]] = Field(default_factory=list)
+    messages: list[dict[str, Any]] = Field(default_factory=list)
+    artifacts: list[dict[str, Any]] = Field(default_factory=list)
+    persist: bool = True                 # write through to HellGraph (else dry-run: return the bundle only)
+
+
+class IngestWorkspaceResponse(BaseModel):
+    self_ref: str
+    bundle: GraphBundle
+    normalized: bool = False             # edges passed through /cskg/normalize
+    hellgraph: dict[str, Any] = Field(default_factory=dict)
+    recall_candidate_ref: str | None = None
+
+
+class RetractRequest(BaseModel):
+    """Retention: a WorkspaceSource was deleted → retract everything derived from
+    it (HellGraph flips those edges to promotionState=superseded)."""
+    source_refs: list[str]
+
+
+class RetractResponse(BaseModel):
+    requested_source_refs: list[str]
+    hellgraph: dict[str, Any] = Field(default_factory=dict)
diff --git a/services/workspace_ingestion/requirements.txt b/services/workspace_ingestion/requirements.txt
new file mode 100644
index 0000000..cae4c6b
--- /dev/null
+++ b/services/workspace_ingestion/requirements.txt
@@ -0,0 +1,5 @@
+fastapi>=0.115,<1
+uvicorn[standard]>=0.30,<1
+httpx>=0.27,<1
+pydantic>=1.10,<3
+psycopg[binary]>=3.2,<4
diff --git a/services/workspace_ingestion/tests/__init__.py b/services/workspace_ingestion/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/workspace_ingestion/tests/test_workspace_ingestion.py b/services/workspace_ingestion/tests/test_workspace_ingestion.py
new file mode 100644
index 0000000..1e7a87d
--- /dev/null
+++ b/services/workspace_ingestion/tests/test_workspace_ingestion.py
@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import unittest
+
+from services.workspace_ingestion.app import main as svc
+from services.workspace_ingestion.app.mapping import build_graph
+from services.workspace_ingestion.app.models import (
+    CSKGEdge,
+    GraphBundle,
+    IngestWorkspaceRequest,
+    RetractRequest,
+)
+
+CONTACT_MOM = {
+    "contactId": "c-mom", "contactClass": "person", "displayName": "Mom",
+    "labels": ["family"], "sourceId": "workspace-source:contacts/mom",
+    "socialProfiles": [{"platform": "linkedin", "handle": "ada"}],
+}
+CONTACT_JAMIE = {
+    "contactId": "c-jamie", "contactClass": "person", "displayName": "Jamie",
+    "organizationRef": "Acme Music", "labels": ["bandmate"],
+    "sourceId": "workspace-source:contacts/jamie",
+}
+EVENT_PRACTICE = {
+    "eventId": "e-practice", "title": "Band practice",
+    "sourceId": "workspace-source:calendar/practice",
+    "attendees": [
+        {"isSelf": True, "email": "me@x.com"},
+        {"contactRef": "c-jamie", "name": "Jamie"},
+    ],
+}
+MAIL_JAMIE = {
+    "messageId": "m-1", "threadId": "t-1",
+    "from": {"email": "jamie@x.com", "name": "Jamie", "contactRef": "c-jamie"},
+    "sourceId": "workspace-source:mail/jamie-thread",
+}
+ARTIFACT_SETLIST = {
+    "artifactId": "a-setlist", "title": "Setlist",
+    "sourceId": "workspace-source:office/setlist",
+}
+
+SELF = "cskg-node://self/lord"
+
+
+class MappingTests(unittest.TestCase):
+    def test_single_self_anchor_and_family_vs_social(self):
+        b = build_graph(SELF, "lord", contacts=[CONTACT_MOM, CONTACT_JAMIE])
+        selves = [n for n in b.nodes if n.node_type == "Self"]
+        self.assertEqual(len(selves), 1)
+        rels = {(e.node1, e.node2): e.relation for e in b.edges}
+        mom = "cskg-node://person/c_mom"
+        jamie = "cskg-node://person/c_jamie"
+        self.assertEqual(rels[(SELF, mom)], "relatedTo")   # family hint
+        self.assertEqual(rels[(SELF, jamie)], "knows")     # social
+        # worksAt to org
+        self.assertEqual(rels[(jamie, "cskg-node://organization/acme_music")], "worksAt")
+
+    def test_every_element_is_provenance_bound_to_workspace_source(self):
+        b = build_graph(SELF, "lord", contacts=[CONTACT_JAMIE], events=[EVENT_PRACTICE], artifacts=[ARTIFACT_SETLIST])
+        for n in b.nodes:
+            self.assertTrue(n.provenance_refs, f"node {n.node_id} has no provenance")
+        for e in b.edges:
+            self.assertTrue(all(r.startswith("workspace-source:") or r == "workspace-source:onboarding" for r in e.provenance_refs) or e.provenance_refs)
+
+    def test_mail_resolves_onto_same_person_no_dup(self):
+        b = build_graph(SELF, "lord", contacts=[CONTACT_JAMIE], messages=[MAIL_JAMIE])
+        jamies = [n for n in b.nodes if n.label == "Jamie"]
+        self.assertEqual(len(jamies), 1)  # mail reused the contact node
+        rels = {e.relation for e in b.edges if e.node2 == "cskg-node://person/c_jamie"}
+        self.assertIn("communicatedWith", rels)
+        self.assertIn("knows", rels)
+
+    def test_social_profile_becomes_reference_only_projection(self):
+        b = build_graph(SELF, "lord", contacts=[CONTACT_MOM])
+        mom = next(n for n in b.nodes if n.label == "Mom")
+        self.assertTrue(mom.external_projection_refs)
+        self.assertTrue(mom.external_projection_refs[0].startswith("projection:social/"))
+
+
+class _FakeHellGraph:
+    """Records calls; simulates HellGraph being present."""
+    enabled = True
+
+    def __init__(self):
+        self.ingested: GraphBundle | None = None
+        self.retracted: list[str] | None = None
+
+    async def normalize(self, edges):
+        return edges
+
+    async def ingest(self, bundle):
+        self.ingested = bundle
+        return {"persisted": True, "nodes": len(bundle.nodes), "edges": len(bundle.edges)}
+
+    async def retract(self, source_refs):
+        self.retracted = source_refs
+        return {"retracted": True, "superseded": len(source_refs)}
+
+
+class _FakeMemoryd:
+    enabled = True
+
+    async def write_summary(self, **kwargs):
+        return {"stored": True, "memory_id": "mem-123"}
+
+
+class ServiceTests(unittest.IsolatedAsyncioTestCase):
+    def setUp(self):
+        self._hg, self._md = svc.hellgraph, svc.memoryd
+        svc.hellgraph = _FakeHellGraph()
+        svc.memoryd = _FakeMemoryd()
+
+    def tearDown(self):
+        svc.hellgraph, svc.memoryd = self._hg, self._md
+
+    async def test_ingest_persists_and_writes_back(self):
+        req = IngestWorkspaceRequest(
+            envelope={"user_id": "lord"},
+            contacts=[CONTACT_JAMIE], events=[EVENT_PRACTICE],
+            messages=[MAIL_JAMIE], artifacts=[ARTIFACT_SETLIST],
+        )
+        resp = await svc.ingest_workspace(req, x_api_key=None)
+        self.assertEqual(resp.self_ref, "cskg-node://self/lord")
+        self.assertTrue(resp.hellgraph["persisted"])
+        self.assertEqual(resp.recall_candidate_ref, "mem-123")
+        types = sorted({n.node_type for n in resp.bundle.nodes})
+        self.assertEqual(types, ["Document", "Event", "Organization", "Person", "Self"])
+        self.assertIsNotNone(svc.hellgraph.ingested)
+
+    async def test_dry_run_does_not_persist(self):
+        req = IngestWorkspaceRequest(envelope={"user_id": "lord"}, contacts=[CONTACT_MOM], persist=False)
+        resp = await svc.ingest_workspace(req, x_api_key=None)
+        self.assertEqual(resp.hellgraph, {})
+        self.assertIsNone(svc.hellgraph.ingested)
+
+    async def test_retract_supersedes_by_source(self):
+        req = RetractRequest(source_refs=["workspace-source:contacts/jamie"])
+        resp = await svc.retract(req, x_api_key=None)
+        self.assertTrue(resp.hellgraph["retracted"])
+        self.assertEqual(svc.hellgraph.retracted, ["workspace-source:contacts/jamie"])
+
+
+if __name__ == "__main__":
+    unittest.main()