diff --git a/.github/workflows/memory-distribution-grant.yml b/.github/workflows/memory-distribution-grant.yml new file mode 100644 index 0000000..0e4e3a8 --- /dev/null +++ b/.github/workflows/memory-distribution-grant.yml @@ -0,0 +1,37 @@ +name: memory-distribution-grant + +on: + pull_request: + paths: + - "schemas/memory-distribution-grant.schema.json" + - "examples/memory-distribution/**" + - "scripts/validate_memory_distribution_grant.py" + - "docs/architecture/memory-distribution-grant.md" + - ".github/workflows/memory-distribution-grant.yml" + - "Makefile" + push: + branches: + - main + paths: + - "schemas/memory-distribution-grant.schema.json" + - "examples/memory-distribution/**" + - "scripts/validate_memory_distribution_grant.py" + - "docs/architecture/memory-distribution-grant.md" + - ".github/workflows/memory-distribution-grant.yml" + - "Makefile" + workflow_dispatch: + +jobs: + validate-memory-distribution-grant: + runs-on: ubuntu-latest + permissions: + contents: read + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + - name: Install validation dependency + run: python -m pip install jsonschema + - name: Validate Memory Distribution Grant examples + run: make validate-memory-distribution-grant diff --git a/.sourceos/manifest.json b/.sourceos/manifest.json index 2845cdc..ff8f1b3 100644 --- a/.sourceos/manifest.json +++ b/.sourceos/manifest.json @@ -6,7 +6,8 @@ "MemoryObject", "MemoryProvenance", "MemoryLifecycleState", - "MemoryReview" + "MemoryReview", + "MemoryDistributionGrant" ], "syncEngines": [ { @@ -35,7 +36,11 @@ "memory.promoted", "memory.quarantined", "memory.revoked", - "memory.expired" + "memory.expired", + "memory.distribution_granted", + "memory.distribution_denied", + "memory.materialized_on_target", + "memory.revocation_propagated" ] } ], @@ -54,7 +59,11 @@ "memory.promoted", "memory.quarantined", "memory.revoked", - "memory.expired" + "memory.expired", + "memory.distribution_granted", + "memory.distribution_denied", + "memory.materialized_on_target", + "memory.revocation_propagated" ], "dangerousSurfaces": [ "memory.promote", @@ -62,7 +71,8 @@ "memory.cross_profile_sync", "memory.rewrite", "memory.revoke", - "memory.agent_proposed" + "memory.agent_proposed", + "memory.materialize_on_target" ], "authorityRepos": [ "SourceOS-Linux/sourceos-spec", diff --git a/Makefile b/Makefile index d7d026f..fb37f9d 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-prophet-mesh-scope-mirror 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-memory-distribution-grant 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 @@ -33,7 +33,10 @@ 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-prophet-mesh-scope-mirror +validate-memory-distribution-grant: + $(PYTHON) scripts/validate_memory_distribution_grant.py + +validate: validate-upstreams validate-python validate-deploy-assets validate-wallguard-memory-compartment-gate validate-memory-distribution-grant validate-prophet-mesh-scope-mirror local-preflight: bash deploy/local/scripts/preflight-podman-m2.sh diff --git a/adapters/openclaw-memory-mesh/src/edgeMemoryStore.ts b/adapters/openclaw-memory-mesh/src/edgeMemoryStore.ts index 7e65320..ae5af8f 100644 --- a/adapters/openclaw-memory-mesh/src/edgeMemoryStore.ts +++ b/adapters/openclaw-memory-mesh/src/edgeMemoryStore.ts @@ -15,6 +15,11 @@ export type EdgeMemoryEntry = { created_at: string; updated_at: string; sync_state: "local_only" | "pending_write" | "synced" | "conflict"; + // Distribution-grant fields (see docs/architecture/memory-distribution-grant.md). + // Revocation is enforced at read: a revoked entry is never returned by search(), + // regardless of whether a full sync has run since the tombstone landed. + revoked?: boolean; + revoked_at?: string | null; }; type EdgeMemoryStoreOptions = { @@ -150,8 +155,9 @@ export class EdgeMemoryStore { ]; for (const entry of entries.slice().sort((a, b) => b.updated_at.localeCompare(a.updated_at))) { lines.push(`## ${entry.memory_class}`); - lines.push(`- content: ${entry.content}`); + lines.push(`- content: ${entry.revoked ? "(revoked — tombstoned)" : entry.content}`); lines.push(`- sync_state: ${entry.sync_state}`); + lines.push(`- revoked: ${entry.revoked ? `true (${entry.revoked_at ?? "unknown"})` : "false"}`); lines.push(`- mesh_event_id: ${entry.mesh_event_id ?? "none"}`); lines.push(`- updated_at: ${entry.updated_at}`); lines.push(""); @@ -167,6 +173,12 @@ export class EdgeMemoryStore { const orderedScopes = buildScopeOrder(scopeOrder); const hits: MemoryHit[] = []; for (const entry of entries) { + // Read-enforced revocation: a tombstoned entry must never surface, even if a + // full sync has not yet reconciled it away. This is the edge honoring of the + // MemoryDistributionGrant revocation cursor (enforceAtRead). + if (entry.revoked) { + continue; + } const scope = scopeNameForEnvelope(envelope, entry.envelope); if (scope === "none") { continue; @@ -229,4 +241,35 @@ export class EdgeMemoryStore { await this.writeEntries(entries); return entry; } + + /** + * Apply a revocation tombstone at the edge as a first-class event, rather than + * waiting for the next full sync to reconcile the entry away. Matches by mesh + * event id or local id. Once tombstoned, the entry is retained (for audit and + * offline "last accepted revision" display) but is filtered out of search(). + * Returns the number of entries tombstoned. + */ + async revoke(ref: string): Promise { + if (!this.enabled) { + return 0; + } + const entries = await this.readEntries(); + const timestamp = new Date().toISOString(); + let tombstoned = 0; + for (const entry of entries) { + if (entry.revoked) { + continue; + } + if (entry.local_id === ref || entry.mesh_event_id === ref) { + entry.revoked = true; + entry.revoked_at = timestamp; + entry.updated_at = timestamp; + tombstoned += 1; + } + } + if (tombstoned > 0) { + await this.writeEntries(entries); + } + return tombstoned; + } } diff --git a/docs/architecture/memory-distribution-grant.md b/docs/architecture/memory-distribution-grant.md new file mode 100644 index 0000000..0f7875a --- /dev/null +++ b/docs/architecture/memory-distribution-grant.md @@ -0,0 +1,117 @@ +# Memory Distribution Grant + +## Status + +Contract v0.1. Fixture-backed and CI-validated. Runtime honoring begins in the OpenClaw +edge adapter (read-enforced revocation); broader runtime wiring is follow-on work. + +## Why this exists + +Memory Mesh already governs two decisions: + +1. **Scope approval** — may a memory become active in a governed scope? Owned by the + governed-learning lifecycle (`schemas/governed-learning-lifecycle-record.schema.json`): + `observed → proposed → scoped → approved → promoted → revoked`. +2. **Compartment admission** — may a memory cross from one wall/compartment to another + (including into a derived embedding/recall lane)? Owned by WallGuard + (`schemas/wallguard-memory-compartment-gate.schema.json`), which gates + `read_memory`, `write_memory`, `semantic_recall`, `embedding_write`, + `memory_promotion`, and `clean_room_release`, and can `quarantine`. + +Neither answers a third, independent question: **may _this specific endpoint_ — a device, +a profile, a surface — receive and materialize a memory that is already approved in scope +and admitted across the wall, and until when?** + +A memory can be globally approved and still be too sensitive for a shared kiosk, a +semi-trusted phone, or a profile that should never hold matter-privileged content. Scope +approval is not a distribution right. This contract makes that distinction first-class. + +## The three axes (and why we do not collapse them) + +| Axis | Question | Owner | +|------|----------|-------| +| Scope approval | Active in a governed scope? | governed-learning-lifecycle | +| Compartment admission | May it cross this wall / enter the derived lane? | WallGuard | +| **Endpoint distribution** | **May _this_ device/profile/surface hold it, until when?** | **MemoryDistributionGrant** | + +The grant **chains** the first two rather than re-deciding them. `upstreamAuthority` +carries a `lifecycleRecordRef` (+ `scopeApprovalStatus`) and a `wallDecisionRef` +(+ `wallDecisionOutcome`). The grant contributes exactly one new decision — the endpoint +one — plus the machinery to keep that decision honest over time. This is deliberate: we do +not want a second compartment engine, and we do not want distribution logic leaking into +WallGuard. If WallGuard denied or quarantined, or the scope was revoked, the grant cannot +admit — the validator enforces this. + +## What the grant adds + +- **Endpoint sensitivity ceiling.** `target.endpointSensitivityCeiling` bounds what an + endpoint may hold. An `admit` whose memory sensitivity exceeds the ceiling is rejected + (`over_sensitive_for_target`). This is the "globally approved, still too sensitive here" + case, encoded. +- **Standing-privilege expiry.** `grantValidity.expiresAt` forces re-authorization. A fresh + scope approval does not silently renew a distribution grant. +- **Content digest.** `memoryRef.contentDigest` (sha256) + `memoryVersion` give the target + tamper-evidence and dedup, and bind the grant to an exact revision. +- **Read-enforced revocation cursor.** `revocationCursor` is a monotonic `watermark` plus + `tombstoneRefs`, with `enforceAtRead: true`. Revocation propagates as a first-class event + (`propagationMode: event_first`), **but correctness does not depend on the event landing**: + the target filters against the cursor at read time, so a revoked item cannot be served + even if the push has not yet arrived. `poll_fallback` covers degraded connectivity. +- **Explicit offline posture.** A disconnected target never claims current global truth. It + either serves the last accepted revision flagged stale + (`serve_last_accepted_with_staleness`, within a `stalenessBudgetSeconds`) or denies until + fresh (`deny_until_fresh`). `sensitive-redacted` memory is pinned to `deny_until_fresh`. +- **Evidence-fabric receipt.** `distributionReceiptRef` points into the shared + Run/Event/Receipt spine rather than a bespoke sync-receipt blob, so distribution is + auditable the same way every other governed surface is. + +## Materialization rule + +`materialization.verifyGrantBeforeMaterialize` is always true: a target verifies the grant +(decision `admit`, unexpired, cursor not revoked, digest match) **before** materializing. +Only an admitted grant may be materialized, and a materialized grant must carry a receipt. +While an item is quarantined, `quarantineDerivedLaneBlocked` records that the target will +not admit it into a derived graph/index lane another endpoint can query — the lane itself is +gated by WallGuard, not re-implemented here. + +## Graph materialization (HellGraph) honors revocation + +The derived-lane concern is concrete in this estate: memory is materialized into the managed +**HellGraph** graph (the `workspace_ingestion` CSKG → HellGraph path). A revocation that only +tombstoned the canonical row would leave the revoked knowledge queryable as graph edges. So +revocation propagates into the graph: + +- `memoryd` `POST /v1/revoke` tombstones the memory (read-enforced across the in-memory, + SQLite, and Postgres backends — a revoked row is excluded from every subsequent recall, + including externally-indexed vector hits), **and** calls HellGraph `POST /v1/retract` to + supersede the edges derived from that memory. +- **Provenance convention (matters for correctness):** HellGraph edges are keyed by their + originating `workspace-source:{surface}/{slug}` refs (the `workspace_ingestion` CSKG path), + *not* by memory id. So the retract keys on the real provenance: `memoryd` reads the memory's + stored `metadata.provenance_refs` and passes those to retract, merged with any caller-supplied + refs and a canonical `memory://{memory_id}` self-ref. The PKG ingest writeback records those + `workspace-source` refs into the summary memory's metadata precisely so revoking it supersedes + the edges it derived. A memoryd memory with no derived edges retracts nothing (no-op). +- The retract binding graceful-degrades: with `HELLGRAPH_URL` unset it is inert and the local + tombstone still holds, so the service (and tests) run without a live graph. +- A `memory.revocation_propagated` audit event records the graph source refs and retract result. + +This closes the loop: recall, the vector index, and the graph materialization all stop serving +a revoked memory, without waiting on a full sync. + +## Fixtures + +- `grant.admitted.example.json` — trusted device, internal memory within ceiling. +- `grant.over-sensitive-denied.example.json` — sensitive-redacted memory denied to a + semi-trusted shared surface. +- `grant.revoked-propagated.example.json` — scope revoked; cursor advanced with tombstone; + deny + not materialized. +- `grant.offline-stale.example.json` — served last accepted revision, flagged stale, within + budget. + +## Related + +- `schemas/wallguard-memory-compartment-gate.schema.json` — compartment admission. +- `schemas/governed-learning-lifecycle-record.schema.json` — scope approval + revocation record. +- `contracts/prophet-mesh/prophet-mesh-memory-scope.v0.1.json` — scope-presence mirror. +- Issue #18 (governed memory lifecycle), issue #36 (WallGuard runtime memory gate). diff --git a/docs/architecture/memorymesh-architecture.md b/docs/architecture/memorymesh-architecture.md index 7f96b04..e47566a 100644 --- a/docs/architecture/memorymesh-architecture.md +++ b/docs/architecture/memorymesh-architecture.md @@ -84,7 +84,7 @@ We are not building around a managed memory product. We can still choose pragmat ### Functional capabilities -The system provides six core capabilities. +The system provides seven core capabilities. #### A. Memory write @@ -111,19 +111,32 @@ The system: - optionally expands to backend workers; - merges, scores, deduplicates, and returns bounded hits. -#### C. Workload configuration +#### C. Memory revocation + +A caller revokes a memory by id. The system: + +- tombstones the row so it is excluded from every subsequent recall (read-enforced across the + in-memory, SQLite, and Postgres backends, and filtered out of externally-indexed vector hits); +- propagates the revocation into the HellGraph materialization by retracting the derived edges + (`memory://{memory_id}` plus any declared upstream provenance refs), graceful-degrading when no + graph is configured; +- emits a `memory.revocation_propagated` audit event. + +See [memory-distribution-grant.md](./memory-distribution-grant.md). + +#### D. Workload configuration A caller or sidecar asks for the compiled config for a workload. This config is derived from attached resources such as `MemoryAttachment`, `GlobalRecallPolicy`, `MemoryPeer`, `ExportPolicy`, and `ConflictPolicy`. -#### D. Config watch +#### E. Config watch Long-lived processes can subscribe to a config watch stream and react to policy changes without restarts. -#### E. Event inspection +#### F. Event inspection Operators can inspect recent events for local debugging and audit. -#### F. Adapter mediation +#### G. Adapter mediation Thin adapters translate native upstream conventions into the canonical envelope and the `memoryd` contract. diff --git a/docs/architecture/openclaw-edge-memory.md b/docs/architecture/openclaw-edge-memory.md index 445a1b6..8406432 100644 --- a/docs/architecture/openclaw-edge-memory.md +++ b/docs/architecture/openclaw-edge-memory.md @@ -20,7 +20,11 @@ The current branch now includes: - adapter configuration for edge-memory enablement and local recall thresholds; - a file-backed edge-memory mirror with generated `MEMORY.md` projection; - adapter-side recall merge logic across edge and mesh hits; -- `memoryd` recall ordering that respects compiled `local_first` and `recall_scope_order` more concretely. +- `memoryd` recall ordering that respects compiled `local_first` and `recall_scope_order` more concretely; +- read-enforced revocation: `EdgeMemoryStore.revoke()` tombstones an entry as a first-class + event and `search()` filters revoked entries at read time, so a revoked memory cannot be + served from the edge even before the next full sync (see + [memory-distribution-grant.md](./memory-distribution-grant.md)). ## Current scope model diff --git a/examples/memory-distribution/grant.admitted.example.json b/examples/memory-distribution/grant.admitted.example.json new file mode 100644 index 0000000..4e48240 --- /dev/null +++ b/examples/memory-distribution/grant.admitted.example.json @@ -0,0 +1,61 @@ +{ + "schemaVersion": "memory-mesh.memory-distribution-grant.v0.1", + "recordType": "MemoryDistributionGrant", + "grantId": "urn:srcos:memory-distribution-grant:laptop-primary-agentplane-completion", + "memoryRef": { + "memoryId": "urn:srcos:memory:agentplane-completion-contract-v2", + "memoryVersion": 2, + "sensitivity": "internal", + "sourceScope": "repo:SocioProphet/agentplane", + "contentDigest": { + "algo": "sha256", + "value": "9f2b7c1d4e6a8b0c2d4f6a8c0e2b4d6f8a0c2e4b6d8f0a2c4e6b8d0f2a4c6e8b" + } + }, + "upstreamAuthority": { + "lifecycleRecordRef": "urn:srcos:memory-lifecycle:accepted-agentplane-completion-contract-v2", + "scopeApprovalStatus": "active", + "wallDecisionRef": "urn:srcos:wallguard:decision:same-wall-firm-approved-allow", + "wallDecisionOutcome": "allow" + }, + "target": { + "targetClass": "device", + "targetRef": "urn:srcos:device:laptop-primary-m2", + "endpointSensitivityCeiling": "restricted", + "endpointTrust": "trusted" + }, + "distributionDecision": "admit", + "reasonCode": "grant_admitted", + "grantValidity": { + "issuedAt": "2026-07-17T00:30:00Z", + "expiresAt": "2026-07-24T00:30:00Z" + }, + "materialization": { + "verifyGrantBeforeMaterialize": true, + "materialized": true, + "materializedAt": "2026-07-17T00:30:02Z", + "quarantineDerivedLaneBlocked": false + }, + "revocationCursor": { + "watermark": 41, + "enforceAtRead": true, + "propagationMode": "event_first", + "tombstoneRefs": [], + "lastPropagatedAt": "2026-07-17T00:30:01Z" + }, + "offlinePosture": { + "mode": "serve_last_accepted_with_staleness", + "stalenessBudgetSeconds": 900, + "lastAcceptedRevision": 2, + "servedStale": false + }, + "distributionReceiptRef": "urn:srcos:receipt:distribution:laptop-primary-agentplane-completion", + "evidenceRefs": [ + "urn:srcos:artifact:agentplane-completion-contract-v2", + "urn:srcos:wallguard:decision:same-wall-firm-approved-allow" + ], + "policyDecisionRefs": [ + "sourceos/core/distribution-allow", + "memory-mesh/distribution/endpoint-ceiling-check" + ] +} diff --git a/examples/memory-distribution/grant.offline-stale.example.json b/examples/memory-distribution/grant.offline-stale.example.json new file mode 100644 index 0000000..3ee48f4 --- /dev/null +++ b/examples/memory-distribution/grant.offline-stale.example.json @@ -0,0 +1,61 @@ +{ + "schemaVersion": "memory-mesh.memory-distribution-grant.v0.1", + "recordType": "MemoryDistributionGrant", + "grantId": "urn:srcos:memory-distribution-grant:laptop-primary-offline-recall", + "memoryRef": { + "memoryId": "urn:srcos:memory:agentplane-completion-contract-v2", + "memoryVersion": 2, + "sensitivity": "internal", + "sourceScope": "repo:SocioProphet/agentplane", + "contentDigest": { + "algo": "sha256", + "value": "9f2b7c1d4e6a8b0c2d4f6a8c0e2b4d6f8a0c2e4b6d8f0a2c4e6b8d0f2a4c6e8b" + } + }, + "upstreamAuthority": { + "lifecycleRecordRef": "urn:srcos:memory-lifecycle:accepted-agentplane-completion-contract-v2", + "scopeApprovalStatus": "active", + "wallDecisionRef": "urn:srcos:wallguard:decision:same-wall-firm-approved-allow", + "wallDecisionOutcome": "allow" + }, + "target": { + "targetClass": "device", + "targetRef": "urn:srcos:device:laptop-primary-m2", + "endpointSensitivityCeiling": "restricted", + "endpointTrust": "trusted" + }, + "distributionDecision": "admit", + "reasonCode": "grant_admitted", + "grantValidity": { + "issuedAt": "2026-07-17T00:30:00Z", + "expiresAt": "2026-07-24T00:30:00Z" + }, + "materialization": { + "verifyGrantBeforeMaterialize": true, + "materialized": true, + "materializedAt": "2026-07-17T00:30:02Z", + "quarantineDerivedLaneBlocked": false + }, + "revocationCursor": { + "watermark": 41, + "enforceAtRead": true, + "propagationMode": "poll_fallback", + "tombstoneRefs": [], + "lastPropagatedAt": "2026-07-17T00:14:00Z" + }, + "offlinePosture": { + "mode": "serve_last_accepted_with_staleness", + "stalenessBudgetSeconds": 900, + "lastAcceptedRevision": 2, + "servedStale": true + }, + "distributionReceiptRef": "urn:srcos:receipt:distribution:laptop-primary-offline-recall-stale", + "evidenceRefs": [ + "urn:srcos:artifact:agentplane-completion-contract-v2", + "urn:srcos:policy:offline-staleness-budget" + ], + "policyDecisionRefs": [ + "memory-mesh/distribution/offline-posture", + "sourceos/core/distribution-allow" + ] +} diff --git a/examples/memory-distribution/grant.over-sensitive-denied.example.json b/examples/memory-distribution/grant.over-sensitive-denied.example.json new file mode 100644 index 0000000..de75501 --- /dev/null +++ b/examples/memory-distribution/grant.over-sensitive-denied.example.json @@ -0,0 +1,61 @@ +{ + "schemaVersion": "memory-mesh.memory-distribution-grant.v0.1", + "recordType": "MemoryDistributionGrant", + "grantId": "urn:srcos:memory-distribution-grant:shared-kiosk-client-matter-note", + "memoryRef": { + "memoryId": "urn:srcos:memory:client-matter-privileged-note", + "memoryVersion": 5, + "sensitivity": "sensitive-redacted", + "sourceScope": "matter:acme-v-globex", + "contentDigest": { + "algo": "sha256", + "value": "1a3c5e7092b4d6f8103254768a9cbedf012345678abcdef0fedcba9876543210" + } + }, + "upstreamAuthority": { + "lifecycleRecordRef": "urn:srcos:memory-lifecycle:accepted-client-matter-privileged-note", + "scopeApprovalStatus": "active", + "wallDecisionRef": "urn:srcos:wallguard:decision:matter-scoped-clean-room-allow", + "wallDecisionOutcome": "allow" + }, + "target": { + "targetClass": "surface", + "targetRef": "urn:srcos:surface:shared-kiosk-lobby", + "endpointSensitivityCeiling": "internal", + "endpointTrust": "semi_trusted" + }, + "distributionDecision": "deny", + "reasonCode": "over_sensitive_for_target", + "grantValidity": { + "issuedAt": "2026-07-17T00:31:00Z", + "expiresAt": "2026-07-17T01:31:00Z" + }, + "materialization": { + "verifyGrantBeforeMaterialize": true, + "materialized": false, + "materializedAt": null, + "quarantineDerivedLaneBlocked": true + }, + "revocationCursor": { + "watermark": 63, + "enforceAtRead": true, + "propagationMode": "event_first", + "tombstoneRefs": [], + "lastPropagatedAt": "2026-07-17T00:31:00Z" + }, + "offlinePosture": { + "mode": "deny_until_fresh", + "stalenessBudgetSeconds": 0, + "lastAcceptedRevision": 0, + "servedStale": false + }, + "distributionReceiptRef": "urn:srcos:receipt:distribution:shared-kiosk-client-matter-note-denied", + "evidenceRefs": [ + "urn:srcos:wallguard:decision:matter-scoped-clean-room-allow", + "urn:srcos:policy:endpoint-sensitivity-ceiling" + ], + "policyDecisionRefs": [ + "memory-mesh/distribution/endpoint-ceiling-check", + "sourceos/core/distribution-deny" + ] +} diff --git a/examples/memory-distribution/grant.revoked-propagated.example.json b/examples/memory-distribution/grant.revoked-propagated.example.json new file mode 100644 index 0000000..4abcb20 --- /dev/null +++ b/examples/memory-distribution/grant.revoked-propagated.example.json @@ -0,0 +1,63 @@ +{ + "schemaVersion": "memory-mesh.memory-distribution-grant.v0.1", + "recordType": "MemoryDistributionGrant", + "grantId": "urn:srcos:memory-distribution-grant:laptop-primary-stop-gate-remediation", + "memoryRef": { + "memoryId": "urn:srcos:memory:agentplane-stop-gate-remediation", + "memoryVersion": 1, + "sensitivity": "internal", + "sourceScope": "repo:SocioProphet/agentplane", + "contentDigest": { + "algo": "sha256", + "value": "abcdef0123456789abcdef0123456789abcdef0123456789abcdef0123456789" + } + }, + "upstreamAuthority": { + "lifecycleRecordRef": "urn:srcos:memory-lifecycle:revoked-stop-gate-remediation", + "scopeApprovalStatus": "revoked", + "wallDecisionRef": "urn:srcos:wallguard:decision:same-wall-firm-approved-allow", + "wallDecisionOutcome": "allow" + }, + "target": { + "targetClass": "device", + "targetRef": "urn:srcos:device:laptop-primary-m2", + "endpointSensitivityCeiling": "restricted", + "endpointTrust": "trusted" + }, + "distributionDecision": "deny", + "reasonCode": "revoked", + "grantValidity": { + "issuedAt": "2026-05-10T20:00:00Z", + "expiresAt": "2026-05-17T20:00:00Z" + }, + "materialization": { + "verifyGrantBeforeMaterialize": true, + "materialized": false, + "materializedAt": null, + "quarantineDerivedLaneBlocked": true + }, + "revocationCursor": { + "watermark": 88, + "enforceAtRead": true, + "propagationMode": "event_first", + "tombstoneRefs": [ + "urn:srcos:tombstone:memory:agentplane-stop-gate-remediation" + ], + "lastPropagatedAt": "2026-05-10T20:12:03Z" + }, + "offlinePosture": { + "mode": "serve_last_accepted_with_staleness", + "stalenessBudgetSeconds": 900, + "lastAcceptedRevision": 1, + "servedStale": false + }, + "distributionReceiptRef": "urn:srcos:receipt:distribution:laptop-primary-stop-gate-remediation-revoked", + "evidenceRefs": [ + "urn:srcos:tombstone:memory:agentplane-stop-gate-remediation", + "urn:srcos:revocation:stop-gate-remediation-v1" + ], + "policyDecisionRefs": [ + "sourceos/core/revocation-allow", + "memory-mesh/distribution/revocation-cursor-enforce" + ] +} diff --git a/schemas/memory-distribution-grant.schema.json b/schemas/memory-distribution-grant.schema.json new file mode 100644 index 0000000..430c604 --- /dev/null +++ b/schemas/memory-distribution-grant.schema.json @@ -0,0 +1,140 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://schemas.socioprophet.org/memory-mesh/memory-distribution-grant.schema.json", + "title": "Memory Distribution Grant", + "description": "Per-endpoint distribution authority for an already-approved memory. Answers the question WallGuard and the governed-learning lifecycle do NOT: may THIS device/profile/surface receive and materialize this memory, and until when. Scope approval (governed-learning-lifecycle) and compartment admission (wallguard-memory-compartment-gate) are referenced, never re-decided here.", + "type": "object", + "additionalProperties": false, + "required": [ + "schemaVersion", + "recordType", + "grantId", + "memoryRef", + "upstreamAuthority", + "target", + "distributionDecision", + "reasonCode", + "grantValidity", + "materialization", + "revocationCursor", + "offlinePosture", + "evidenceRefs", + "policyDecisionRefs" + ], + "properties": { + "schemaVersion": {"const": "memory-mesh.memory-distribution-grant.v0.1"}, + "recordType": {"const": "MemoryDistributionGrant"}, + "grantId": {"type": "string", "minLength": 1}, + "memoryRef": { + "type": "object", + "additionalProperties": false, + "required": ["memoryId", "memoryVersion", "sensitivity", "sourceScope", "contentDigest"], + "properties": { + "memoryId": {"type": "string", "minLength": 1}, + "memoryVersion": {"type": "integer", "minimum": 1}, + "sensitivity": {"enum": ["public", "internal", "restricted", "sensitive-redacted"]}, + "sourceScope": {"type": "string", "minLength": 1}, + "contentDigest": { + "type": "object", + "additionalProperties": false, + "required": ["algo", "value"], + "properties": { + "algo": {"const": "sha256"}, + "value": {"type": "string", "pattern": "^[0-9a-f]{64}$"} + } + } + } + }, + "upstreamAuthority": { + "type": "object", + "additionalProperties": false, + "description": "Authority this grant chains from. The grant carries no compartment or scope-approval logic of its own; it points at the records that already made those decisions.", + "required": ["lifecycleRecordRef", "scopeApprovalStatus", "wallDecisionRef", "wallDecisionOutcome"], + "properties": { + "lifecycleRecordRef": {"type": "string", "minLength": 1}, + "scopeApprovalStatus": {"enum": ["active", "revoked", "superseded", "unknown"]}, + "wallDecisionRef": {"type": "string", "minLength": 1}, + "wallDecisionOutcome": {"enum": ["allow", "deny", "redact", "quarantine", "escalate"]} + } + }, + "target": { + "type": "object", + "additionalProperties": false, + "required": ["targetClass", "targetRef", "endpointSensitivityCeiling", "endpointTrust"], + "properties": { + "targetClass": {"enum": ["device", "profile", "surface"]}, + "targetRef": {"type": "string", "minLength": 1}, + "endpointSensitivityCeiling": {"enum": ["public", "internal", "restricted", "sensitive-redacted"]}, + "endpointTrust": {"enum": ["trusted", "semi_trusted", "untrusted"]} + } + }, + "distributionDecision": {"enum": ["admit", "deny", "require-review", "fail-closed"]}, + "reasonCode": { + "enum": [ + "grant_admitted", + "over_sensitive_for_target", + "scope_not_approved", + "wall_denied", + "grant_expired", + "endpoint_untrusted", + "policy_decision_missing", + "revoked" + ] + }, + "grantValidity": { + "type": "object", + "additionalProperties": false, + "required": ["issuedAt", "expiresAt"], + "description": "Distribution grants are standing privilege and therefore expire. A scope approval does not renew a distribution grant.", + "properties": { + "issuedAt": {"type": "string", "format": "date-time"}, + "expiresAt": {"type": "string", "format": "date-time"} + } + }, + "materialization": { + "type": "object", + "additionalProperties": false, + "required": ["verifyGrantBeforeMaterialize", "materialized", "materializedAt", "quarantineDerivedLaneBlocked"], + "properties": { + "verifyGrantBeforeMaterialize": {"const": true}, + "materialized": {"type": "boolean"}, + "materializedAt": {"type": ["string", "null"], "format": "date-time"}, + "quarantineDerivedLaneBlocked": { + "type": "boolean", + "description": "True asserts the target will not admit this item into a derived graph/index lane while quarantined. Enforcement of that lane belongs to WallGuard (embedding_write/semantic_recall); this flag records that the target honors it." + } + } + }, + "revocationCursor": { + "type": "object", + "additionalProperties": false, + "required": ["watermark", "enforceAtRead", "propagationMode", "tombstoneRefs", "lastPropagatedAt"], + "description": "Revocation is a first-class, read-enforced event. The target filters against the watermark at read time, so a revoked item cannot be served even if the push event has not yet landed.", + "properties": { + "watermark": {"type": "integer", "minimum": 0}, + "enforceAtRead": {"const": true}, + "propagationMode": {"enum": ["event_first", "poll_fallback"]}, + "tombstoneRefs": {"type": "array", "items": {"type": "string", "minLength": 1}}, + "lastPropagatedAt": {"type": ["string", "null"], "format": "date-time"} + } + }, + "offlinePosture": { + "type": "object", + "additionalProperties": false, + "required": ["mode", "stalenessBudgetSeconds", "lastAcceptedRevision", "servedStale"], + "description": "How the target behaves while it cannot confirm cursor freshness. It never claims current global truth; it either serves the last accepted revision flagged stale, or denies until fresh.", + "properties": { + "mode": {"enum": ["serve_last_accepted_with_staleness", "deny_until_fresh"]}, + "stalenessBudgetSeconds": {"type": "integer", "minimum": 0}, + "lastAcceptedRevision": {"type": "integer", "minimum": 0}, + "servedStale": {"type": "boolean"} + } + }, + "distributionReceiptRef": { + "type": ["string", "null"], + "description": "Pointer into the reasoning/evidence fabric Receipt for this distribution decision. Preferred over a bespoke sync-receipt blob so distribution rides the same Run/Event/Receipt spine as every other governed surface." + }, + "evidenceRefs": {"type": "array", "minItems": 1, "items": {"type": "string", "minLength": 1}}, + "policyDecisionRefs": {"type": "array", "minItems": 1, "items": {"type": "string", "minLength": 1}} + } +} diff --git a/scripts/validate_memory_distribution_grant.py b/scripts/validate_memory_distribution_grant.py new file mode 100644 index 0000000..cef0a9b --- /dev/null +++ b/scripts/validate_memory_distribution_grant.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +"""Validate Memory Distribution Grant examples. + +The distribution grant is the per-endpoint authority layer. It must never carry +compartment or scope-approval logic of its own -- it references the WallGuard +decision and the governed-learning lifecycle record that already made those +calls. These invariants enforce the separation and the fail-safe rules +(read-enforced revocation, endpoint sensitivity ceiling, expiry). +""" + +from __future__ import annotations + +import json +from datetime import datetime +from pathlib import Path + +from jsonschema import Draft202012Validator + +ROOT = Path(__file__).resolve().parents[1] +SCHEMA = ROOT / "schemas" / "memory-distribution-grant.schema.json" +EXAMPLE_DIR = ROOT / "examples" / "memory-distribution" + +SENSITIVITY_RANK = { + "public": 0, + "internal": 1, + "restricted": 2, + "sensitive-redacted": 3, +} + + +def load_json(path: Path) -> dict: + with path.open("r", encoding="utf-8") as handle: + return json.load(handle) + + +def parse_ts(value: str) -> datetime: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + + +def validate_schema(instance: dict, schema: dict, *, source_label: str) -> None: + validator = Draft202012Validator(schema) + errors = sorted(validator.iter_errors(instance), key=lambda error: list(error.path)) + if errors: + lines = [f"{source_label} failed validation:"] + for error in errors: + location = ".".join(str(part) for part in error.path) or "" + lines.append(f" - {location}: {error.message}") + raise ValueError("\n".join(lines)) + + +def validate_invariants(grant: dict, *, source_label: str) -> None: + decision = grant["distributionDecision"] + reason = grant["reasonCode"] + memory = grant["memoryRef"] + target = grant["target"] + upstream = grant["upstreamAuthority"] + materialization = grant["materialization"] + cursor = grant["revocationCursor"] + offline = grant["offlinePosture"] + validity = grant["grantValidity"] + + # Grant never re-decides compartment/scope: it must point at both authorities. + if not upstream["lifecycleRecordRef"] or not upstream["wallDecisionRef"]: + raise ValueError( + f"{source_label}: distribution grant must chain lifecycleRecordRef and wallDecisionRef, not re-decide them" + ) + + # Revocation must be enforced at read (schema pins const true, assert defensively). + if cursor["enforceAtRead"] is not True: + raise ValueError(f"{source_label}: revocationCursor.enforceAtRead must be true") + if materialization["verifyGrantBeforeMaterialize"] is not True: + raise ValueError(f"{source_label}: target must verify grant before materialize") + + # Expiry sanity. + if parse_ts(validity["expiresAt"]) <= parse_ts(validity["issuedAt"]): + raise ValueError(f"{source_label}: grant expiresAt must be after issuedAt") + + # A materialized item is an admitted, receipted item only. + if materialization["materialized"]: + if decision != "admit": + raise ValueError(f"{source_label}: only an admitted grant may be materialized") + if not grant.get("distributionReceiptRef"): + raise ValueError(f"{source_label}: a materialized grant requires a distributionReceiptRef") + if materialization["materializedAt"] is None: + raise ValueError(f"{source_label}: a materialized grant requires materializedAt") + else: + if materialization["materializedAt"] is not None: + raise ValueError(f"{source_label}: a non-materialized grant must not carry materializedAt") + + # Endpoint sensitivity ceiling: an admit may not exceed the endpoint's ceiling. + over_ceiling = SENSITIVITY_RANK[memory["sensitivity"]] > SENSITIVITY_RANK[target["endpointSensitivityCeiling"]] + if decision == "admit" and over_ceiling: + raise ValueError( + f"{source_label}: memory sensitivity exceeds endpoint ceiling; admit is not permitted" + ) + if reason == "over_sensitive_for_target": + if not over_ceiling: + raise ValueError( + f"{source_label}: over_sensitive_for_target requires memory sensitivity above endpoint ceiling" + ) + if decision not in ("deny", "fail-closed"): + raise ValueError(f"{source_label}: over_sensitive_for_target must deny or fail closed") + + # Revoked upstream scope must not be admitted, and must surface a tombstone. + if upstream["scopeApprovalStatus"] == "revoked": + if decision == "admit" or materialization["materialized"]: + raise ValueError(f"{source_label}: revoked scope must not be admitted or materialized") + if reason != "revoked": + raise ValueError(f"{source_label}: revoked scope must carry reasonCode 'revoked'") + if not cursor["tombstoneRefs"]: + raise ValueError(f"{source_label}: revoked distribution must carry a tombstone in the cursor") + + if reason == "revoked" and decision == "admit": + raise ValueError(f"{source_label}: reasonCode 'revoked' may not be admitted") + + # WallGuard denial cannot be admitted here. + if upstream["wallDecisionOutcome"] in ("deny", "quarantine") and decision == "admit": + raise ValueError(f"{source_label}: a wall deny/quarantine outcome cannot be admitted downstream") + + # Fail-safe offline posture: sensitive-redacted must not be served stale. + if memory["sensitivity"] == "sensitive-redacted" and offline["mode"] != "deny_until_fresh": + raise ValueError( + f"{source_label}: sensitive-redacted memory must use deny_until_fresh offline posture" + ) + if offline["servedStale"] and offline["mode"] != "serve_last_accepted_with_staleness": + raise ValueError(f"{source_label}: servedStale only valid under serve_last_accepted_with_staleness") + + +def main() -> int: + schema = load_json(SCHEMA) + Draft202012Validator.check_schema(schema) + + examples = sorted(EXAMPLE_DIR.glob("grant.*.example.json")) + if not examples: + raise SystemExit("No memory distribution grant examples found") + + checked = [] + for path in examples: + data = load_json(path) + validate_schema(data, schema, source_label=str(path)) + validate_invariants(data, source_label=str(path)) + checked.append(path.name) + + print(json.dumps({"ok": True, "checked": checked}, indent=2)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/services/memoryd/app/hellgraph_retract.py b/services/memoryd/app/hellgraph_retract.py new file mode 100644 index 0000000..484940a --- /dev/null +++ b/services/memoryd/app/hellgraph_retract.py @@ -0,0 +1,68 @@ +"""HellGraph retract binding for memoryd — propagates revocation into the graph lane. + +When a memory is revoked, its derived edges in the managed HellGraph must be superseded +too, otherwise the revoked knowledge stays queryable in the graph even though recall no +longer serves it. HellGraph is a downstream *materialization* of memory, never the +canonical store, so memoryd only needs the retract half of the contract here: + + POST {HELLGRAPH_URL}/v1/retract {"provenance_refs": [...]} → supersede derived edges + +Graceful-degrade like mem0_client / qdrant_index: if HELLGRAPH_URL is unset the binding is +disabled and returns an inert result, so memoryd runs (and tests pass) without a live graph. +The workspace_ingestion service owns the full CSKG ingest path; this binding deliberately +does not duplicate it. +""" +from __future__ import annotations + +import os +from typing import Any + +import httpx + + +class HellGraphRetractClient: + def __init__(self, base_url: str | None = None, api_key: str | None = None, timeout_seconds: float = 10.0) -> None: + self.base_url = (base_url if base_url is not None else os.getenv('HELLGRAPH_URL', '')).rstrip('/') + self.api_key = api_key if api_key is not None else os.getenv('HELLGRAPH_API_KEY', '') + 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 retract(self, source_refs: list[str]) -> dict[str, Any]: + """Supersede every graph edge derived from the given provenance/source refs.""" + if not self.enabled: + return {'retracted': False, 'reason': 'hellgraph disabled', 'source_refs': source_refs} + if not source_refs: + return {'retracted': False, 'reason': 'no source refs', 'source_refs': source_refs} + async with httpx.AsyncClient(timeout=self.timeout_seconds) as client: + response = await client.post( + f'{self.base_url}/v1/retract', + json={'provenance_refs': source_refs}, + headers=self.headers, + ) + response.raise_for_status() + return dict(response.json()) + + +def revocation_source_refs(memory_id: str, extra_refs: list[str] | None = None) -> list[str]: + """Collect the provenance refs a revoked memory contributed to the graph. + + Every memoryd memory contributes edges under the canonical self-ref + ``memory://{memory_id}``; callers that know additional upstream provenance refs + (e.g. a WorkspaceSource id the memory was derived from) pass them as ``extra_refs`` + so those edges are superseded too. + """ + refs: list[str] = [f'memory://{memory_id}'] + for ref in extra_refs or []: + if isinstance(ref, str) and ref and ref not in refs: + refs.append(ref) + return refs diff --git a/services/memoryd/app/main.py b/services/memoryd/app/main.py index 0854a4b..aa18117 100644 --- a/services/memoryd/app/main.py +++ b/services/memoryd/app/main.py @@ -12,6 +12,7 @@ from .embedding import HashingEmbedder from .mem0_client import Mem0RestClient +from .hellgraph_retract import HellGraphRetractClient, revocation_source_refs from .models import ( ApplyResourceResponse, DEFAULT_SCOPE_ORDER, @@ -19,6 +20,8 @@ MeshResource, RecallRequest, RecallResponse, + RevokeRequest, + RevokeResponse, WriteRequest, WriteResponse, dump_model, @@ -57,6 +60,13 @@ ) embedder = HashingEmbedder(dimension=VECTOR_SIZE, salt=EMBEDDING_SALT) +# Revocation propagates into the graph materialization (graceful-degrade if HELLGRAPH_URL unset). +hellgraph_retract = HellGraphRetractClient( + base_url=os.getenv('HELLGRAPH_URL'), + api_key=os.getenv('HELLGRAPH_API_KEY'), + timeout_seconds=float(os.getenv('HELLGRAPH_TIMEOUT_SECONDS', '10')), +) + def build_store(store_uri: str) -> StoreProtocol: if not store_uri or store_uri == 'memory://': @@ -364,6 +374,50 @@ async def write(request: WriteRequest, x_api_key: str | None = Header(default=No ) +@app.post('/v1/revoke', response_model=RevokeResponse) +async def revoke(request: RevokeRequest, x_api_key: str | None = Header(default=None)) -> RevokeResponse: + await require_api_key(x_api_key) + # Read the memory's stored provenance before tombstoning so we can retract the exact + # graph edges it derived (e.g. the workspace-source refs a PKG ingest wrote back), + # merged with any refs the caller supplies. + stored_metadata = await store.get_memory_metadata(request.memory_id) or {} + stored_refs = stored_metadata.get('provenance_refs') or stored_metadata.get('provenanceRefs') or [] + if isinstance(stored_refs, str): + stored_refs = [stored_refs] + + revoked = await store.revoke_memory(request.memory_id) + if not revoked: + raise HTTPException(status_code=404, detail='memory not found or already revoked') + + # Propagate revocation into the HellGraph materialization: supersede the edges + # derived from this memory so it is not queryable in the graph lane either. + extra_refs = list(request.provenance_refs) + [r for r in stored_refs if r not in request.provenance_refs] + source_refs = revocation_source_refs(request.memory_id, extra_refs) + graph_result: dict[str, Any] + try: + graph_result = await hellgraph_retract.retract(source_refs) + except Exception as exc: # pragma: no cover + graph_result = {'retracted': False, 'error': str(exc), 'source_refs': source_refs} + + event = await store.append_event( + 'memory.revocation_propagated', + { + 'memory_id': request.memory_id, + 'reason': request.reason, + 'revocation_ref': request.revocation_ref, + 'tombstone_ref': request.tombstone_ref, + 'graph_source_refs': source_refs, + 'graph_retract': graph_result, + }, + ) + return RevokeResponse( + memory_id=request.memory_id, + revoked=True, + event_id=event.event_id, + graph_retract=graph_result, + ) + + @app.get('/v1/events') async def list_events(limit: int = 50, x_api_key: str | None = Header(default=None)) -> dict: await require_api_key(x_api_key) diff --git a/services/memoryd/app/models.py b/services/memoryd/app/models.py index e87e9d6..67bcf2d 100644 --- a/services/memoryd/app/models.py +++ b/services/memoryd/app/models.py @@ -155,6 +155,26 @@ class WriteResponse(BaseModel): stored_locally: bool = True +class RevokeRequest(BaseModel): + # Read-enforced revocation (see docs/architecture/memory-distribution-grant.md). + # Tombstoning a memory here removes it from every subsequent recall on this node, + # so a revoked memory cannot be served even before a full sync reconciles it away. + memory_id: str = Field(min_length=1) + reason: str | None = None + revocation_ref: str | None = None + tombstone_ref: str | None = None + # Upstream provenance refs this memory contributed to the graph, so revocation can + # supersede the derived edges too. The canonical memory://{id} ref is always added. + provenance_refs: list[str] = Field(default_factory=list) + + +class RevokeResponse(BaseModel): + memory_id: str + revoked: bool + event_id: str + graph_retract: dict[str, Any] = Field(default_factory=dict) + + class EventRecord(BaseModel): event_id: str = Field(default_factory=lambda: uuid4().hex) event_type: str diff --git a/services/memoryd/app/postgres_store.py b/services/memoryd/app/postgres_store.py index a7c857f..89f069d 100644 --- a/services/memoryd/app/postgres_store.py +++ b/services/memoryd/app/postgres_store.py @@ -69,6 +69,8 @@ def _migrate(self) -> None: ''', f"CREATE INDEX IF NOT EXISTS idx_{self.schema}_memories_user_id ON {self.schema}.memories ((envelope->>'user_id'))", f"CREATE INDEX IF NOT EXISTS idx_{self.schema}_memories_workload_id ON {self.schema}.memories ((envelope->>'workload_id'))", + # Read-enforced revocation column (added defensively for pre-existing tables). + f'ALTER TABLE {self.schema}.memories ADD COLUMN IF NOT EXISTS revoked BOOLEAN NOT NULL DEFAULT FALSE', ] with self._connect() as conn: with conn.cursor() as cur: @@ -195,9 +197,44 @@ async def search_local_memories(self, request: RecallRequest) -> list[MemoryHit] lexical_hits = await asyncio.to_thread(self._search_lexical_sync, request) vector_hits = await self._vector_index.search(request) if self._vector_index is not None and request.query_vector else [] merged = dedupe_hits(lexical_hits + vector_hits) + # Read-enforced revocation for externally-indexed vector hits. + if vector_hits: + revoked = await asyncio.to_thread(self._revoked_ids_sync) + merged = [hit for hit in merged if hit.memory_id not in revoked] merged.sort(key=lambda hit: hit.score, reverse=True) return merged[: request.top_k] + def _revoked_ids_sync(self) -> set[str]: + with self._connect() as conn: + with conn.cursor() as cur: + cur.execute(f'SELECT memory_id FROM {self.schema}.memories WHERE revoked = TRUE') + rows = cur.fetchall() + return {row[0] for row in rows} + + async def revoke_memory(self, memory_id: str) -> bool: + return await asyncio.to_thread(self._revoke_memory_sync, memory_id) + + def _revoke_memory_sync(self, memory_id: str) -> bool: + with self._connect() as conn: + with conn.cursor() as cur: + cur.execute( + f'UPDATE {self.schema}.memories SET revoked = TRUE WHERE memory_id = %s AND revoked = FALSE', + (memory_id,), + ) + changed = cur.rowcount > 0 + conn.commit() + return changed + + async def get_memory_metadata(self, memory_id: str) -> dict | None: + return await asyncio.to_thread(self._get_memory_metadata_sync, memory_id) + + def _get_memory_metadata_sync(self, memory_id: str) -> dict | None: + with self._connect() as conn: + with conn.cursor() as cur: + cur.execute(f'SELECT metadata FROM {self.schema}.memories WHERE memory_id = %s', (memory_id,)) + row = cur.fetchone() + return dict(row[0] or {}) if row else None + def _search_lexical_sync(self, request: RecallRequest) -> list[MemoryHit]: limit = max(request.top_k * 20, 100) with self._connect() as conn: @@ -207,7 +244,8 @@ def _search_lexical_sync(self, request: RecallRequest) -> list[MemoryHit]: f''' SELECT memory_id, text_content, tags, metadata, event_id, envelope FROM {self.schema}.memories - WHERE (envelope->>'user_id') = %s + WHERE revoked = FALSE + AND (envelope->>'user_id') = %s AND (envelope->>'workload_id') = %s ORDER BY created_at DESC LIMIT %s @@ -219,7 +257,8 @@ def _search_lexical_sync(self, request: RecallRequest) -> list[MemoryHit]: f''' SELECT memory_id, text_content, tags, metadata, event_id, envelope FROM {self.schema}.memories - WHERE (envelope->>'user_id') = %s + WHERE revoked = FALSE + AND (envelope->>'user_id') = %s AND (envelope->>'workload_id') = %s AND (envelope->>'workspace_id') = %s ORDER BY created_at DESC diff --git a/services/memoryd/app/sqlite_store.py b/services/memoryd/app/sqlite_store.py index 0111b41..b1f39fa 100644 --- a/services/memoryd/app/sqlite_store.py +++ b/services/memoryd/app/sqlite_store.py @@ -65,6 +65,10 @@ def _migrate(self) -> None: CREATE INDEX IF NOT EXISTS idx_memories_created_at ON memories(created_at DESC); ''' ) + # Read-enforced revocation column (added defensively for pre-existing DBs). + columns = {row['name'] for row in conn.execute('PRAGMA table_info(memories)').fetchall()} + if 'revoked' not in columns: + conn.execute('ALTER TABLE memories ADD COLUMN revoked INTEGER NOT NULL DEFAULT 0') conn.commit() @staticmethod @@ -174,13 +178,41 @@ async def search_local_memories(self, request: RecallRequest) -> list[MemoryHit] lexical_hits = await asyncio.to_thread(self._search_lexical_sync, request) vector_hits = await self._vector_index.search(request) if self._vector_index is not None and request.query_vector else [] merged = dedupe_hits(lexical_hits + vector_hits) + # Read-enforced revocation for vector hits: the lexical query already excludes + # revoked rows, but vector hits come from an external index that may still hold + # the tombstoned vector, so drop anything revoked before returning. + if vector_hits: + revoked = await asyncio.to_thread(self._revoked_ids_sync) + merged = [hit for hit in merged if hit.memory_id not in revoked] merged.sort(key=lambda hit: hit.score, reverse=True) return merged[: request.top_k] + def _revoked_ids_sync(self) -> set[str]: + with self._connect() as conn: + rows = conn.execute('SELECT memory_id FROM memories WHERE revoked = 1').fetchall() + return {row['memory_id'] for row in rows} + + async def revoke_memory(self, memory_id: str) -> bool: + return await asyncio.to_thread(self._revoke_memory_sync, memory_id) + + def _revoke_memory_sync(self, memory_id: str) -> bool: + with self._connect() as conn: + cur = conn.execute('UPDATE memories SET revoked = 1 WHERE memory_id = ? AND revoked = 0', (memory_id,)) + conn.commit() + return cur.rowcount > 0 + + async def get_memory_metadata(self, memory_id: str) -> dict | None: + return await asyncio.to_thread(self._get_memory_metadata_sync, memory_id) + + def _get_memory_metadata_sync(self, memory_id: str) -> dict | None: + with self._connect() as conn: + row = conn.execute('SELECT metadata_json FROM memories WHERE memory_id = ?', (memory_id,)).fetchone() + return dict(json.loads(row['metadata_json'])) if row else None + def _search_lexical_sync(self, request: RecallRequest) -> list[MemoryHit]: with self._connect() as conn: rows = conn.execute( - 'SELECT memory_id, text_content, tags_json, metadata_json, event_id, envelope_json FROM memories ORDER BY created_at DESC LIMIT ?', + 'SELECT memory_id, text_content, tags_json, metadata_json, event_id, envelope_json FROM memories WHERE revoked = 0 ORDER BY created_at DESC LIMIT ?', (max(request.top_k * 20, 100),), ).fetchall() query_tokens = tokenize(request.query) diff --git a/services/memoryd/app/store.py b/services/memoryd/app/store.py index 9df3f0f..84f43c7 100644 --- a/services/memoryd/app/store.py +++ b/services/memoryd/app/store.py @@ -33,6 +33,8 @@ async def list_events(self, limit: int = 50) -> list[EventRecord]: ... async def compile_workload_config(self, workload_id: str) -> CompiledWorkloadConfig: ... async def add_local_memory(self, request: WriteRequest, event_id: str) -> str: ... async def search_local_memories(self, request: RecallRequest) -> list[MemoryHit]: ... + async def revoke_memory(self, memory_id: str) -> bool: ... + async def get_memory_metadata(self, memory_id: str) -> dict | None: ... async def health(self) -> dict: ... @@ -41,6 +43,7 @@ def __init__(self) -> None: self._resources: dict[str, MeshResource] = {} self._events: list[EventRecord] = [] self._memories: dict[str, dict] = {} + self._revoked: set[str] = set() self._resource_index: dict[str, list[str]] = defaultdict(list) async def init(self) -> None: @@ -91,6 +94,9 @@ async def search_local_memories(self, request: RecallRequest) -> list[MemoryHit] query_tokens = tokenize(request.query) hits: list[MemoryHit] = [] for record in self._memories.values(): + # Read-enforced revocation: a tombstoned memory never surfaces in recall. + if record['memory_id'] in self._revoked: + continue env = record['envelope'] scope_bonus, scope_name = scope_bonus_for_request(request, env) if scope_bonus < 0: @@ -114,12 +120,25 @@ async def search_local_memories(self, request: RecallRequest) -> list[MemoryHit] hits.sort(key=lambda hit: hit.score, reverse=True) return hits[: request.top_k] + async def revoke_memory(self, memory_id: str) -> bool: + # Tombstone the memory so recall never serves it again. Idempotent: + # returns True only the first time a known memory is revoked. + if memory_id not in self._memories or memory_id in self._revoked: + return False + self._revoked.add(memory_id) + return True + + async def get_memory_metadata(self, memory_id: str) -> dict | None: + record = self._memories.get(memory_id) + return dict(record['metadata']) if record else None + async def health(self) -> dict: return { 'backend': 'memory', 'resource_count': len(self._resources), 'event_count': len(self._events), 'memory_count': len(self._memories), + 'revoked_count': len(self._revoked), } diff --git a/services/memoryd/tests/test_distribution_revocation.py b/services/memoryd/tests/test_distribution_revocation.py new file mode 100644 index 0000000..d500b95 --- /dev/null +++ b/services/memoryd/tests/test_distribution_revocation.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import tempfile +import unittest + +from services.memoryd.app.hellgraph_retract import HellGraphRetractClient, revocation_source_refs +from services.memoryd.app.models import RecallRequest, ScopeEnvelope, WriteRequest +from services.memoryd.app.sqlite_store import SQLiteStore +from services.memoryd.app.store import InMemoryStore + + +def make_envelope() -> ScopeEnvelope: + return ScopeEnvelope( + user_id='lord', + agent_id='openclaw-main', + run_id='run-1', + workload_id='openclaw-gateway', + source_interface='openclaw', + ) + + +class RevocationTests(unittest.IsolatedAsyncioTestCase): + async def test_inmemory_revoked_memory_is_not_recalled(self) -> None: + store = InMemoryStore() + req = WriteRequest(envelope=make_envelope(), content='The launch code is hunter2') + event = await store.append_event('memory.write', {'content': req.content}) + memory_id = await store.add_local_memory(req, event.event_id) + + before = await store.search_local_memories( + RecallRequest(envelope=make_envelope(), query='What is the launch code?', top_k=5) + ) + self.assertTrue(before) + + self.assertTrue(await store.revoke_memory(memory_id)) + self.assertFalse(await store.revoke_memory(memory_id)) # idempotent + self.assertFalse(await store.revoke_memory('does-not-exist')) + + after = await store.search_local_memories( + RecallRequest(envelope=make_envelope(), query='What is the launch code?', top_k=5) + ) + self.assertEqual(after, []) + health = await store.health() + self.assertEqual(health['revoked_count'], 1) + + async def test_sqlite_revoked_memory_is_not_recalled(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + store = SQLiteStore(db_path=f'{tmp}/memorymesh.db') + await store.init() + req = WriteRequest(envelope=make_envelope(), content='The vault combination is memorymesh') + event = await store.append_event('memory.write', {'content': req.content}) + memory_id = await store.add_local_memory(req, event.event_id) + + before = await store.search_local_memories( + RecallRequest(envelope=make_envelope(), query='What is the vault combination?', top_k=5) + ) + self.assertTrue(before) + + self.assertTrue(await store.revoke_memory(memory_id)) + self.assertFalse(await store.revoke_memory(memory_id)) # idempotent + + after = await store.search_local_memories( + RecallRequest(envelope=make_envelope(), query='What is the vault combination?', top_k=5) + ) + self.assertEqual(after, []) + + # Revocation survives reopen (durable tombstone, not just an in-process flag). + reopened = SQLiteStore(db_path=f'{tmp}/memorymesh.db') + await reopened.init() + after_reopen = await reopened.search_local_memories( + RecallRequest(envelope=make_envelope(), query='What is the vault combination?', top_k=5) + ) + self.assertEqual(after_reopen, []) + + async def test_revoke_endpoint_retracts_stored_workspace_source_edges(self) -> None: + # The linkage that matters: a PKG-ingest summary memory carries workspace-source + # provenance in metadata. Revoking it must retract those exact graph edges, without + # the caller having to restate them. + from services.memoryd.app import main as svc + + class _CapturingRetract: + enabled = True + + def __init__(self): + self.calls: list[list[str]] = [] + + async def retract(self, source_refs): + self.calls.append(source_refs) + return {'retracted': True, 'superseded': len(source_refs), 'source_refs': source_refs} + + try: + from starlette.testclient import TestClient + except Exception: # pragma: no cover + self.skipTest('starlette TestClient unavailable') + + original_store = svc.store + original_retract = svc.hellgraph_retract + svc.store = InMemoryStore() + capture = _CapturingRetract() + svc.hellgraph_retract = capture + try: + await svc.store.init() + env = {'user_id': 'lord', 'agent_id': 'a1', 'run_id': 'r1', + 'workload_id': 'personal-knowledge-graph', 'source_interface': 'pkg'} + with TestClient(svc.app) as client: + write = client.post('/v1/write', json={ + 'envelope': env, + 'content': 'personal_context_graph.updated', + 'metadata': {'provenance_refs': ['workspace-source:contacts/jamie', + 'workspace-source:mail/jamie-thread']}, + }) + memory_id = write.json()['memory_id'] + resp = client.post('/v1/revoke', json={'memory_id': memory_id}) + self.assertEqual(resp.status_code, 200) + body = resp.json() + self.assertTrue(body['graph_retract']['retracted']) + # The retract call carried the canonical self-ref plus the stored workspace-source refs. + self.assertEqual(len(capture.calls), 1) + sent = capture.calls[0] + self.assertIn(f'memory://{memory_id}', sent) + self.assertIn('workspace-source:contacts/jamie', sent) + self.assertIn('workspace-source:mail/jamie-thread', sent) + finally: + svc.store = original_store + svc.hellgraph_retract = original_retract + + async def test_graph_retract_source_refs_and_graceful_degrade(self) -> None: + refs = revocation_source_refs('abc123', ['workspace-source:xyz']) + self.assertEqual(refs[0], 'memory://abc123') + self.assertIn('workspace-source:xyz', refs) + + # With HELLGRAPH_URL unset the binding is inert and never raises. + client = HellGraphRetractClient(base_url='', api_key='') + self.assertFalse(client.enabled) + result = await client.retract(refs) + self.assertFalse(result['retracted']) + + +if __name__ == '__main__': + unittest.main() diff --git a/services/workspace_ingestion/app/main.py b/services/workspace_ingestion/app/main.py index 0d84355..0561a59 100644 --- a/services/workspace_ingestion/app/main.py +++ b/services/workspace_ingestion/app/main.py @@ -71,7 +71,11 @@ async def ingest_workspace(request: IngestWorkspaceRequest, x_api_key: str | Non 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}, + # provenance_refs is the linkage memoryd revocation reads: revoking this summary + # memory must retract the workspace-source edges it derived, so carry them in + # metadata (not only in the opaque content blob). + metadata={"self_ref": self_ref, "workload_id": request.envelope.workload_id, + "provenance_refs": source_refs}, ) recall_ref = summary.get("memory_id") if isinstance(summary, dict) else None diff --git a/services/workspace_ingestion/tests/test_workspace_ingestion.py b/services/workspace_ingestion/tests/test_workspace_ingestion.py index 1e7a87d..b9bd04c 100644 --- a/services/workspace_ingestion/tests/test_workspace_ingestion.py +++ b/services/workspace_ingestion/tests/test_workspace_ingestion.py @@ -100,7 +100,11 @@ async def retract(self, source_refs): class _FakeMemoryd: enabled = True + def __init__(self): + self.last_kwargs: dict | None = None + async def write_summary(self, **kwargs): + self.last_kwargs = kwargs return {"stored": True, "memory_id": "mem-123"} @@ -127,6 +131,16 @@ async def test_ingest_persists_and_writes_back(self): self.assertEqual(types, ["Document", "Event", "Organization", "Person", "Self"]) self.assertIsNotNone(svc.hellgraph.ingested) + # Linkage: the writeback summary must carry the workspace-source provenance refs in + # metadata, so a later memoryd revoke retracts the graph edges this ingest derived. + meta = svc.memoryd.last_kwargs["metadata"] + self.assertIn("provenance_refs", meta) + self.assertTrue(meta["provenance_refs"]) + self.assertTrue(all(r.startswith("workspace-source:") for r in meta["provenance_refs"])) + # And they match the edges' provenance, so retract(request.source_refs) hits them. + edge_refs = {r for e in resp.bundle.edges for r in e.provenance_refs} + self.assertTrue(set(meta["provenance_refs"]) & edge_refs) + 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) diff --git a/specs/memory.mesh.v1.trpc.yaml b/specs/memory.mesh.v1.trpc.yaml index 4924e3e..17dad83 100644 --- a/specs/memory.mesh.v1.trpc.yaml +++ b/specs/memory.mesh.v1.trpc.yaml @@ -180,6 +180,19 @@ spec: memory_id: string? backend_memory_ids: [string] stored_locally: bool + - name: Revoke + req: + envelope: ScopeEnvelope + memory_id: string + reason: string? + revocation_ref: string? + tombstone_ref: string? + provenance_refs: [string]? + res: + memory_id: string + revoked: bool + event_id: string + graph_retract: object - name: Events.List req: envelope: ScopeEnvelope @@ -213,6 +226,12 @@ spec: - deny when compiled_workload_config.writeback_enabled is false - force persist_to_backend false when compiled_workload_config.allow_backend_persistence is false - emit a typed write.accepted or write.rejected event + revoke: + mandatory_enforcement: + - a revoked memory must be excluded from every subsequent recall (read-enforced, not sync-dependent) + - revocation must propagate to the graph materialization (HellGraph retract of derived edges) + - graph retract degrades gracefully when HELLGRAPH_URL is unset; the local tombstone still holds + - emit a typed memory.revocation_propagated event carrying the graph source refs resource_apply: mandatory_enforcement: - config recompile must occur for affected workload targets @@ -238,6 +257,7 @@ spec: - memory.recall.denied - memory.write.accepted - memory.write.rejected + - memory.revocation_propagated - memory.export.planned - memory.export.denied - memory.conflict.detected diff --git a/specs/memoryd.openapi.yaml b/specs/memoryd.openapi.yaml index eb3d682..b60adee 100644 --- a/specs/memoryd.openapi.yaml +++ b/specs/memoryd.openapi.yaml @@ -219,6 +219,42 @@ paths: application/json: schema: $ref: '#/components/schemas/HTTPValidationError' + /v1/revoke: + post: + summary: Revoke + description: >- + Tombstone a memory (read-enforced revocation) and propagate the revocation into + the HellGraph materialization by superseding the derived edges. See + docs/architecture/memory-distribution-grant.md. + operationId: revoke_v1_revoke_post + parameters: + - required: false + schema: + title: X-Api-Key + type: string + name: x-api-key + in: header + requestBody: + content: + application/json: + schema: + $ref: '#/components/schemas/RevokeRequest' + required: true + responses: + '200': + description: Successful Response + content: + application/json: + schema: + $ref: '#/components/schemas/RevokeResponse' + '404': + description: Memory not found or already revoked + '422': + description: Validation Error + content: + application/json: + schema: + $ref: '#/components/schemas/HTTPValidationError' /v1/events: get: summary: List Events @@ -588,3 +624,47 @@ components: title: Stored Locally type: boolean default: true + RevokeRequest: + title: RevokeRequest + required: + - memory_id + type: object + properties: + memory_id: + title: Memory Id + minLength: 1 + type: string + reason: + title: Reason + type: string + revocation_ref: + title: Revocation Ref + type: string + tombstone_ref: + title: Tombstone Ref + type: string + provenance_refs: + title: Provenance Refs + type: array + items: + type: string + RevokeResponse: + title: RevokeResponse + required: + - memory_id + - revoked + - event_id + type: object + properties: + memory_id: + title: Memory Id + type: string + revoked: + title: Revoked + type: boolean + event_id: + title: Event Id + type: string + graph_retract: + title: Graph Retract + type: object