From 0c5f14a0be3603653d103841ab073280cc96a8c1 Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Wed, 10 Jun 2026 22:14:40 -0400 Subject: [PATCH 1/2] feat(lane-g): wire validate-prophet-mesh-scope-mirror into make validate and CI Adds validate-prophet-mesh-scope-mirror to the main validate target so the prophet-mesh memory-scope mirror contract is gated on every validate run, and adds a CI workflow that triggers on changes to the contract or validator script. --- .../workflows/prophet-mesh-scope-mirror.yml | 24 +++++++++++++++++++ Makefile | 4 ++-- 2 files changed, 26 insertions(+), 2 deletions(-) create mode 100644 .github/workflows/prophet-mesh-scope-mirror.yml diff --git a/.github/workflows/prophet-mesh-scope-mirror.yml b/.github/workflows/prophet-mesh-scope-mirror.yml new file mode 100644 index 0000000..080666e --- /dev/null +++ b/.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 diff --git a/Makefile b/Makefile index 6ffa5f3..bbe4e41 100644 --- a/Makefile +++ b/Makefile @@ -1,6 +1,6 @@ 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 @@ -32,7 +32,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 From 4f41ad7c9c2eafbd7bb9f5102f8e3e15d98b487e Mon Sep 17 00:00:00 2001 From: Michael Heller <21163552+mdheller@users.noreply.github.com> Date: Wed, 1 Jul 2026 15:51:53 -0400 Subject: [PATCH 2/2] =?UTF-8?q?feat(workspace=5Fingestion):=20PKG=20runtim?= =?UTF-8?q?e=20=E2=80=94=20WorkspaceSource=20=E2=86=92=20CSKG=20=E2=86=92?= =?UTF-8?q?=20managed=20HellGraph?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Step 4 of the person-graph substrate plan: the runtime that realises the PersonalContextGraph contract (prophet-workspace) on the managed HellGraph (prophet-platform). memory-mesh owns the runtime. New service services/workspace_ingestion (conforms to the finance_saa pattern): - mapping.py: pure WorkspaceSource → CSKGNode/CSKGEdge. contact→Person (family vs social relation, worksAt Organization, social→reference-only projection), calendar→Event+participatedIn (self+attendees), mail→communicatedWith, office-artifact→Document+authored. Every element provenance-bound to its workspace-source id; people ingested first so mail/calendar edges dedup. - 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 unset (house client pattern). - memory_mesh_client.py: writeback summary to memoryd after ingest. - main.py: POST /v1/ingest/workspace, POST /v1/retract (step 6 retention), /healthz. 7/7 tests pass (mapping invariants + service ingest/dry-run/retract with fakes); wired into Makefile validate-python. Runs without live HellGraph/memoryd. --- Makefile | 3 +- services/workspace_ingestion/README.md | 41 ++++ services/workspace_ingestion/app/__init__.py | 0 .../app/hellgraph_client.py | 122 ++++++++++ services/workspace_ingestion/app/main.py | 89 +++++++ services/workspace_ingestion/app/mapping.py | 228 ++++++++++++++++++ .../app/memory_mesh_client.py | 48 ++++ services/workspace_ingestion/app/models.py | 80 ++++++ services/workspace_ingestion/requirements.txt | 5 + .../workspace_ingestion/tests/__init__.py | 0 .../tests/test_workspace_ingestion.py | 144 +++++++++++ 11 files changed, 759 insertions(+), 1 deletion(-) create mode 100644 services/workspace_ingestion/README.md create mode 100644 services/workspace_ingestion/app/__init__.py create mode 100644 services/workspace_ingestion/app/hellgraph_client.py create mode 100644 services/workspace_ingestion/app/main.py create mode 100644 services/workspace_ingestion/app/mapping.py create mode 100644 services/workspace_ingestion/app/memory_mesh_client.py create mode 100644 services/workspace_ingestion/app/models.py create mode 100644 services/workspace_ingestion/requirements.txt create mode 100644 services/workspace_ingestion/tests/__init__.py create mode 100644 services/workspace_ingestion/tests/test_workspace_ingestion.py 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/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/ + 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()