workspace_ingestion — PKG runtime (WorkspaceSource → CSKG → managed HellGraph)

.github/workflows/prophet-mesh-scope-mirror.yml

@@ -0,0 +1,24 @@
+name: prophet-mesh-scope-mirror
+
+on:
+  pull_request:
+  push:
+    branches: [main]
+    paths:
+      - "contracts/prophet-mesh/**"
+      - "scripts/validate-prophet-mesh-scope-mirror.mjs"
+      - ".github/workflows/prophet-mesh-scope-mirror.yml"
+
+permissions:
+  contents: read
+
+jobs:
+  validate:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-node@v4
+        with:
+          node-version: "20"
+      - name: Validate prophet-mesh memory-scope mirror contract
+        run: node scripts/validate-prophet-mesh-scope-mirror.mjs

Makefile

@@ -1,14 +1,15 @@
 PYTHON ?= python
 
-.PHONY: validate-upstreams validate-python validate-deploy-assets validate-agent-learning-proposal validate-scenario-learning-binding validate-governed-learning-lifecycle validate-workspace-recall-promotion validate-channel-provenance-write-gate validate-wallguard-memory-compartment-gate validate local-preflight local-up local-smoke local-debug local-down
+.PHONY: validate-upstreams validate-python validate-deploy-assets validate-agent-learning-proposal validate-scenario-learning-binding validate-governed-learning-lifecycle validate-workspace-recall-promotion validate-channel-provenance-write-gate validate-wallguard-memory-compartment-gate validate-prophet-mesh-scope-mirror validate local-preflight local-up local-smoke local-debug local-down
 
 validate-upstreams:
 	$(PYTHON) scripts/validate_upstreams.py third_party/upstreams.lock.yaml
 	$(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
@@ -32,7 +33,7 @@ validate-channel-provenance-write-gate:
 validate-wallguard-memory-compartment-gate:
 	$(PYTHON) scripts/validate_wallguard_memory_compartment_gate.py
 
-validate: validate-upstreams validate-python validate-deploy-assets validate-wallguard-memory-compartment-gate
+validate: validate-upstreams validate-python validate-deploy-assets validate-wallguard-memory-compartment-gate validate-prophet-mesh-scope-mirror
 
 local-preflight:
 	bash deploy/local/scripts/preflight-podman-m2.sh

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).

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),
+            },
+        }

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)