From a57d1f367139ea9d6f74e38684279ad8d4cbf6d8 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Mon, 13 Apr 2026 16:14:16 -0400 Subject: [PATCH 1/4] docs: add content, support, and ops memory loop design --- docs/content-memory-and-recall.md | 128 ++++++++++++++++++++++++++++++ 1 file changed, 128 insertions(+) create mode 100644 docs/content-memory-and-recall.md 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. From 82f37e0060fbe2f4a614a9a61835958858152fc1 Mon Sep 17 00:00:00 2001 From: mdheller <21163552+mdheller@users.noreply.github.com> Date: Tue, 5 May 2026 00:44:27 -0400 Subject: [PATCH 2/4] Add typed artifact API module --- services/memoryd/app/artifact_api.py | 611 +++++++++++++++++++++++++++ 1 file changed, 611 insertions(+) create mode 100644 services/memoryd/app/artifact_api.py 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), + ) 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 3/4] 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 4/4] =?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()