From c435cca892141cf64f1104e1b6800ba5d56df0b6 Mon Sep 17 00:00:00 2001 From: wolverinaton Date: Wed, 12 Aug 2026 21:52:58 -0300 Subject: [PATCH] feat: add governed graph observations --- .planning/GRAPH-OBSERVATIONS-V1.md | 63 ++ DOCS-MAP.md | 11 +- ROADMAP.md | 22 +- .../fixtures/graph_observations_v1.json | 55 ++ docs/generated/release-truth.json | 2 +- docs/generated/release-truth.md | 2 +- docs/public-v1.md | 46 +- memorymaster/capture/repository.py | 33 ++ memorymaster/dreaming/worker.py | 72 +++ .../evaluation/graph_observation_evaluator.py | 138 +++++ memorymaster/govern/llm_steward.py | 22 +- memorymaster/knowledge/context_bundle.py | 56 +- .../knowledge/graph_observation_engine.py | 293 +++++++++ .../knowledge/graph_observation_recall.py | 130 ++++ .../knowledge/graph_observation_repository.py | 556 ++++++++++++++++++ memorymaster/knowledge/graph_observations.py | 323 ++++++++++ memorymaster/public/demo.py | 138 ++++- memorymaster/public/v1.py | 56 +- .../migrations/0020_graph_observations.py | 102 ++++ memorymaster/surfaces/cli.py | 11 + memorymaster/surfaces/cli_handlers_public.py | 5 + memorymaster/surfaces/dashboard.py | 12 +- .../surfaces/graph_observations_dashboard.py | 109 ++++ memorymaster/surfaces/mcp_server.py | 6 + tests/test_graph_observations.py | 490 +++++++++++++++ tests/test_public_cli.py | 12 +- tests/test_public_demo.py | 9 +- tests/test_public_mcp.py | 7 +- 28 files changed, 2701 insertions(+), 80 deletions(-) create mode 100644 .planning/GRAPH-OBSERVATIONS-V1.md create mode 100644 benchmarks/fixtures/graph_observations_v1.json create mode 100644 memorymaster/evaluation/graph_observation_evaluator.py create mode 100644 memorymaster/knowledge/graph_observation_engine.py create mode 100644 memorymaster/knowledge/graph_observation_recall.py create mode 100644 memorymaster/knowledge/graph_observation_repository.py create mode 100644 memorymaster/knowledge/graph_observations.py create mode 100644 memorymaster/stores/migrations/0020_graph_observations.py create mode 100644 memorymaster/surfaces/graph_observations_dashboard.py create mode 100644 tests/test_graph_observations.py diff --git a/.planning/GRAPH-OBSERVATIONS-V1.md b/.planning/GRAPH-OBSERVATIONS-V1.md new file mode 100644 index 00000000..12768d1b --- /dev/null +++ b/.planning/GRAPH-OBSERVATIONS-V1.md @@ -0,0 +1,63 @@ + +# Graph Observations V1 +# Covers: deterministic discovery, governed synthesis, lifecycle, opt-in recall, and rollout evidence. +# Key terms: exact signatures, union-find, support fingerprint, candidate observation, steward gate. +# Read when: implementing, reviewing, testing, or rolling back PPR-7. +# Status: implemented and locally verified; separate PR and CI evidence in progress. + + +## Fixed boundary + +The feature derives optional observations only from current, confirmed, +non-sensitive claims with active evidence and graph-edge supports in one exact +tenant and scope. Components are deterministic; an LLM may summarize an +eligible component but cannot choose membership, cite outside it, promote its +own output, or feed observations back into extraction. + +## Work ledger + +| Area | Required outcome | Status | +|---|---|---| +| Migration 0020 | Add SQLite observation, support, and leased-job tables; PostgreSQL fails closed. | Implemented; migration tests pass | +| Discovery | Canonical signatures, hub suppression, union-find, bounds, deterministic fingerprints. | Implemented; 48-case evaluator is 100%/100% | +| Synthesis | Strict structured output, three-call cycle cap, provider/global budget, replay-safe `no_signal`. | Implemented; focused failure/cap tests pass | +| Lifecycle | Candidate-only creation, deterministic steward gate, immediate archive/stale on support change. | Implemented; promotion/retirement tests pass | +| Recall | Default byte-equivalent output; separately packed opt-in observations bounded to five. | Implemented; equivalence and opt-in tests pass | +| Surfaces | Python, CLI, MCP, dashboard, and disposable demo parity. | Implemented; focused surface tests pass | +| Evaluation | Versioned 40+ case corpus plus structural, citation, leakage, replay, and retrieval gates. | Local gates pass; PR CI pending | + +## Acceptance evidence + +- Structural component precision and recall are each at least 95% on the + versioned corpus; observation precision is at least 90% and root-cause + precision at least 85%. +- Citation/support correctness is 100%, with zero cross-tenant, cross-scope, + sensitive, retired, stale, candidate, or observation-generated support. +- Ordinary recall is byte-equivalent when observations are disabled; graph + top-five improves without more than 0.01 overall R@5/MRR regression. +- `improve()` only queues work and remains below 500 ms p95; an hourly scope + cycle performs at most three observation synthesis calls. +- Focused tests, the full non-ML suite, retrieval gates, Ruff, collection, + migration/restore, clean-wheel, supply-chain, and GitNexus checks pass before +a PPR-7 pull request is proposed. + +## Local verification evidence + +- Offline corpus: 48 cases, structural precision 1.00 and recall 1.00. +- Full non-ML suite: 4,463 passed, 72 skipped, 97 deselected, one expected + failure; Ruff and 4,633-test collection pass. +- Focused restore/retrieval gates: 17 passed; release/supply-chain contract + tests: 86 passed; regenerated release truth verifies. +- Clean wheel builds, passes Twine, installs into a fresh venv, initializes + migration 0020, and completes the observation promotion/recall/staleness demo. +- Feature generation remains disabled by default; the configured Dreaming + extraction/consolidation pair remains Gemini plus GLM. + +## Rollout and rollback + +Roll out through disposable SQLite and fake/local providers, feature-off wheel +installation, offline shadow evaluation, verified authoritative backup/restore, +then opt-in candidate generation. The 24-hour observation is post-implementation +operational evidence, not a coding prerequisite. Rollback disables generation +and recall, archives candidates, marks confirmed generated observations stale, +and preserves the additive tables and audit history. diff --git a/DOCS-MAP.md b/DOCS-MAP.md index 026a0592..74b190a3 100644 --- a/DOCS-MAP.md +++ b/DOCS-MAP.md @@ -1,16 +1,17 @@ - + # DOCS-MAP - memorymaster # Covers: trust verdicts and replacements for every canonical documentation surface. # Key terms: CURRENT, SUPERSEDED, ABANDONED, GENERATED, roadmap, paper radar, ADR. # Read when: locating authoritative project documentation before reading doc bodies. -# Updated: 2026-08-12 after repair5 verifier replay and PR #189 creation. -# Rule: PR #189 is open; merge, release, deploy, and PPR-7 remain blocked pending separate review and integration. +# Updated: 2026-08-12 after PR #189 merge, governed runtime deploy, and PPR-7 authorization. +# Rule: PPR-7 is isolated and opt-in; public release and automatic recall remain separately gated. | File | Verdict | Last change | Reason | |---|---|---|---| | CHANGELOG.md | CURRENT | 2026-08-04 | Public release history; v4.6.0 records governed universal capture, measured quality changes, security evidence, and known follow-ups. | -| ROADMAP.md | CURRENT | 2026-08-12 | Sole authoritative roadmap; elapsed P5 evidence, pinned Gemini+GLM replay, repair5 verifier pass, and PR #189 open. | +| ROADMAP.md | CURRENT | 2026-08-12 | Sole authoritative roadmap; P5 is merged and locally deployed, while isolated PPR-7 implementation is active. | +| .planning/GRAPH-OBSERVATIONS-V1.md | CURRENT | 2026-08-12 | Bounded PPR-7 implementation ledger for deterministic supported components, governed synthesis, lifecycle, opt-in recall, and verification. | | .planning/PAPER-RADAR-REVIEW-2026-08-08.md | CURRENT | 2026-08-08 | Primary-paper ledger covers 57-paper triage, 18 deep reviews, exact MemoryMaster gaps, and ordered PPR-1 through PPR-6 decisions subordinate to ROADMAP.md. | | .planning/PAPER-RESEARCH-IMPLEMENTATION-2026-08-08.md | CURRENT | 2026-08-08 | Executable status ledger for PPR-1 through PPR-6; records acceptance criteria and evidence without competing with ROADMAP.md. | | .planning/HERMES-SCOPE-SKILLS-INTEGRATION-2026-08-07.md | CURRENT | 2026-08-12 | Executable Tencent-derived ledger; repair5 passed and PR #189 is open without merge authority. | @@ -24,7 +25,7 @@ | .planning/VNEXT-GOVERNED-CAPTURE-SPEC.md | CURRENT | 2026-07-27 | Bounded implementation specification that explicitly implements, and does not compete with, ROADMAP.md. | | .planning/VNEXT-BASELINE-2026-07-27.md | CURRENT | 2026-07-27 | Reproducible pre-change retrieval, test, latency, package, capture, graph, and scheduler baseline at d33a268. | | docs/adr/0015-governed-universal-capture-lineage.md | CURRENT | 2026-07-27 | Accepted data-flow decision fixing producer-to-source-to-evidence-to-claim-to-supported-graph lineage and retirement semantics. | -| docs/public-v1.md | CURRENT | 2026-08-08 | Stable facade, capture boundary, additive approved-skill recall, retirement semantics, dashboard inbox, and demo. | +| docs/public-v1.md | CURRENT | 2026-08-12 | Stable facade, capture boundary, opt-in approved skills and observations, retirement semantics, dashboard panels, and demo. | | .planning/audits/2026-07-27-vnext-governed-capture/audit-delta.md | CURRENT | 2026-08-04 | SQLite activation, LifeAgent retirement, OAuth capture quality, public v4.6.0 release, rollback evidence, and the final seven-day observation gate. | | docs/archive/IMPROVEMENT_PLAN.md | ABANDONED | 2026-06-20 | The doc serves as a generated audit and roadmap from March 2026 but is not referenced by any current docs; it contains specific version claims and 'P0' bugs that likely represent a historical snapshot rather than a living plan. | | docs/archive/v315-experiments/E02-results.md | ABANDONED | 2026-06-20 | This is a negative result experiment from a past version (v315) where the code was explicitly reverted and not retained. | diff --git a/ROADMAP.md b/ROADMAP.md index 40c0a909..3a6a112a 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -1,10 +1,10 @@ - + # MemoryMaster roadmap # Covers: post-v4.6 sequence, Tencent-derived work, paper research, and deferrals. # Key terms: Hermes, governed skills, paper radar, temporal projection, sustainability. # Read when: choosing release scope, accepting a feature, or checking deferrals. # Authority: sole roadmap; planning ledgers implement it and never replace it. -# Safety: SQLite authority and steward promotion remain fixed; PR #189 is open, never authorizing merge, release, deployment, or PPR-7. +# Safety: SQLite authority and steward promotion remain fixed; PPR-7 stays additive, opt-in, and unreleased pending its own evidence. ## Shipped in v4.6.0 @@ -59,8 +59,15 @@ its governed-claims authority: - The pinned Gemini Flash Lite plus GLM 5.2 scheduled replay passes: exit 0, eight extraction/consolidation/application decisions, zero errors, and one stale crash run recovered. Repair5 independently replays the bounded gate - against the elapsed baseline and passes; [PR #189](https://github.com/wolverin0/memorymaster/pull/189) - is open, while merge, release, deployment, and PPR-7 remain prohibited. + against the elapsed baseline and passes. [PR #189](https://github.com/wolverin0/memorymaster/pull/189) + merged as `d34b80c` and its clean wheel is active in the governed local + runtime with healthy MCP HTTP probes; the scheduled Dreaming pair remains + Gemini Flash Lite plus GLM 5.2. +- PPR-7 governed graph observations are implemented and locally verified on an + isolated feature branch. Implementation is additive, candidate-first, and + opt-in; its separate PR/CI, feature-off deployment, and rollout evidence are + still required before activation. No public release or automatic + ordinary-recall inclusion is implied. - The invalid earlier window remains incident evidence only because it included a VM OOM/gateway interruption and did not contain P5. - Keep v4.6.0 operational while the post-release Obsidian opt-in and OpenCode @@ -78,7 +85,12 @@ its governed-claims authority: progressive approved-skill reuse are implemented, verified, and active. Windows snapshot/readiness gates, consoleless P5 runtime replacement, VM package rollback preparation, and live functional probes passed. The repair5 - bounded verifier now passes; create a PR only, with no merge or follow-on PPR-7 work. + bounded verifier passed, PR #189 merged, and the governed local runtime was + upgraded with a clean wheel and healthy post-restart probe. +- Review and integrate PPR-7 according to `.planning/GRAPH-OBSERVATIONS-V1.md`; + exact support signatures, deterministic discovery, candidate-only synthesis, + observation-specific steward validation, and explicit recall inclusion are + implemented with local evidence, while PR/CI and feature-off rollout remain. - Improve personal/local backup guidance beyond the already verified disposable backup/restore and migration procedure. - Keep semantic recall optional and disabled unless a local user deliberately diff --git a/benchmarks/fixtures/graph_observations_v1.json b/benchmarks/fixtures/graph_observations_v1.json new file mode 100644 index 00000000..91b93862 --- /dev/null +++ b/benchmarks/fixtures/graph_observations_v1.json @@ -0,0 +1,55 @@ +{ + "corpus_version": "graph-observations-v1", + "algorithm_version": "graph-observations-union-find-v1", + "ontology_version": "personal-v1", + "cases": [ + {"id":"dep-01","category":"dependency_chain","template":"eligible","expected":[[1,2,3]]}, + {"id":"dep-02","category":"dependency_chain","template":"eligible","expected":[[1,2,3]]}, + {"id":"dep-03","category":"dependency_chain","template":"merge","expected":[[1,2,3,4]]}, + {"id":"dep-04","category":"dependency_chain","template":"split","expected":[[1,2,3],[4,5,6]]}, + {"id":"root-01","category":"root_cause","template":"eligible","expected":[[1,2,3]]}, + {"id":"root-02","category":"root_cause","template":"merge","expected":[[1,2,3,4]]}, + {"id":"root-03","category":"root_cause","template":"eligible","expected":[[1,2,3]]}, + {"id":"root-04","category":"root_cause","template":"split","expected":[[1,2,3],[4,5,6]]}, + {"id":"pattern-01","category":"recurring_pattern","template":"eligible","expected":[[1,2,3]]}, + {"id":"pattern-02","category":"recurring_pattern","template":"merge","expected":[[1,2,3,4]]}, + {"id":"pattern-03","category":"recurring_pattern","template":"eligible","expected":[[1,2,3]]}, + {"id":"pattern-04","category":"recurring_pattern","template":"split","expected":[[1,2,3],[4,5,6]]}, + {"id":"unrelated-01","category":"unrelated_similarity","template":"unrelated","expected":[]}, + {"id":"unrelated-02","category":"unrelated_similarity","template":"unrelated","expected":[]}, + {"id":"unrelated-03","category":"unrelated_similarity","template":"one_signature","expected":[]}, + {"id":"unrelated-04","category":"unrelated_similarity","template":"insufficient_claims","expected":[]}, + {"id":"hub-01","category":"hub","template":"hub","expected":[]}, + {"id":"hub-02","category":"hub","template":"hub","expected":[]}, + {"id":"hub-03","category":"hub","template":"hub","expected":[]}, + {"id":"bounds-01","category":"bounds","template":"oversized","expected":[]}, + {"id":"bounds-02","category":"bounds","template":"oversized","expected":[]}, + {"id":"conflict-01","category":"conflict","template":"excluded","state":"conflicted","expected":[]}, + {"id":"conflict-02","category":"conflict","template":"excluded","state":"candidate","expected":[]}, + {"id":"conflict-03","category":"conflict","template":"excluded","state":"stale","expected":[]}, + {"id":"retired-01","category":"retired_evidence","template":"excluded","state":"retired","expected":[]}, + {"id":"retired-02","category":"retired_evidence","template":"excluded","state":"retired","expected":[]}, + {"id":"retired-03","category":"retired_evidence","template":"excluded","state":"retired","expected":[]}, + {"id":"scope-01","category":"scope_boundary","template":"cross_scope","expected":[[1,2,3]]}, + {"id":"scope-02","category":"scope_boundary","template":"cross_scope","expected":[[1,2,3]]}, + {"id":"scope-03","category":"scope_boundary","template":"excluded","state":"wrong_scope","expected":[]}, + {"id":"tenant-01","category":"tenant_boundary","template":"cross_tenant","expected":[[1,2,3]]}, + {"id":"tenant-02","category":"tenant_boundary","template":"cross_tenant","expected":[[1,2,3]]}, + {"id":"tenant-03","category":"tenant_boundary","template":"excluded","state":"wrong_tenant","expected":[]}, + {"id":"secret-01","category":"sensitive_data","template":"excluded","state":"sensitive_claim","expected":[]}, + {"id":"secret-02","category":"sensitive_data","template":"excluded","state":"sensitive_evidence","expected":[]}, + {"id":"secret-03","category":"sensitive_data","template":"excluded","state":"sensitive_source","expected":[]}, + {"id":"candidate-01","category":"lifecycle","template":"excluded","state":"candidate","expected":[]}, + {"id":"stale-01","category":"lifecycle","template":"excluded","state":"stale","expected":[]}, + {"id":"observation-01","category":"feedback_loop","template":"excluded","state":"observation","expected":[]}, + {"id":"skill-01","category":"feedback_loop","template":"excluded","state":"skill","expected":[]}, + {"id":"summary-01","category":"feedback_loop","template":"excluded","state":"summary","expected":[]}, + {"id":"observer-01","category":"feedback_loop","template":"excluded","state":"observer_agent","expected":[]}, + {"id":"minimum-01","category":"eligibility","template":"insufficient_claims","expected":[]}, + {"id":"minimum-02","category":"eligibility","template":"insufficient_evidence","expected":[]}, + {"id":"minimum-03","category":"eligibility","template":"one_signature","expected":[]}, + {"id":"symmetric-01","category":"canonicalization","template":"symmetric","expected":[[1,2,3]]}, + {"id":"replay-01","category":"determinism","template":"eligible","expected":[[1,2,3]]}, + {"id":"replay-02","category":"determinism","template":"merge","expected":[[1,2,3,4]]} + ] +} diff --git a/docs/generated/release-truth.json b/docs/generated/release-truth.json index c04e98c6..e8ccf091 100644 --- a/docs/generated/release-truth.json +++ b/docs/generated/release-truth.json @@ -134,7 +134,7 @@ "console_entrypoints": 8, "mcp_tools": 50, "ops_cli_commands": 5, - "pytest_test_functions": 3767 + "pytest_test_functions": 3781 }, "feature_profile_matrix": { "capture_hook": [ diff --git a/docs/generated/release-truth.md b/docs/generated/release-truth.md index 39ed4f15..3a418e6f 100644 --- a/docs/generated/release-truth.md +++ b/docs/generated/release-truth.md @@ -7,7 +7,7 @@ Do not edit this file by hand. Run `python scripts/generate_release_truth.py`. - Main CLI commands: **118** - Operations CLI commands: **5** - Console entrypoints: **8** -- Pytest source test functions: **3767** +- Pytest source test functions: **3781** ## MCP tools diff --git a/docs/public-v1.md b/docs/public-v1.md index e713acb5..42e266d4 100644 --- a/docs/public-v1.md +++ b/docs/public-v1.md @@ -1,10 +1,10 @@ + # Public v1: remember, recall, forget, improve # Covers: stable Python, CLI, and MCP contracts for governed personal memory. -# Key terms: memorymaster.public.v1, capture envelope, trusted recall, approved skills, logical retirement. +# Key terms: trusted recall, approved skills, derived observations, logical retirement. # Read when: integrating a producer, capturing a file, or building a friendly client. -# Defaults: project workspace scope, confirmed-only recall, preview-only retirement. -# Limits: 2 MiB text, 25 MiB document, 100-item batch; directories/archives unsupported. -# Updated: 2026-08-08; approved-skill projection is additive and off by default. +# Defaults: confirmed claims only; skills and observations off; retirement preview only. + MemoryMaster’s friendly facade does not bypass claim governance. Capture stores source and evidence synchronously, then queues extraction. Extracted claims are @@ -28,6 +28,8 @@ context = recall( token_budget=4000, include_skills=True, skill_limit=3, + include_observations=True, + observation_limit=2, ) preview = forget(source_item_id=receipt.source_item["id"]) @@ -37,11 +39,18 @@ queued = improve(scope="project:atlas", max_items=200) The response contract is versioned as `memorymaster.public.v1`. `remember` returns source, evidence, job IDs, replay/deduplication state, and warnings. `recall` returns rendered context plus claim IDs, citations, lifecycle state, -score explanations, and a `skills` tuple. `include_skills=True` adds complete +score explanations, plus separate `skills` and `observations` tuples. +`include_skills=True` adds complete confirmed skills authorized for the requested scopes as an explicit text section while sharing the same token budget. It excludes raw skill JSON from ordinary claim context. The option defaults off; candidate recall still requires `trust_mode="exploratory"`, and candidate skills are never projected. +`include_observations=True` adds a separately packed `DERIVED OBSERVATIONS` +section. Trusted mode revalidates exact support and returns confirmed +observations only; exploratory mode may label candidate or stale observations. +Observations never enter the ordinary claim list, and `observation_limit` is +bounded to five. The section reserves at most 25% or 800 tokens of the same +recall budget. ## CLI and MCP @@ -49,14 +58,15 @@ requires `trust_mode="exploratory"`, and candidate skills are never projected. memorymaster --workspace . remember --text "A governed observation." memorymaster --workspace . remember --file .\notes\decision.md memorymaster --workspace . remember --url https://example.com/reference -memorymaster --workspace . recall "governed observation" +memorymaster --workspace . recall "governed observation" --include-observations memorymaster --workspace . forget --claim-id 42 memorymaster --workspace . forget --source-item-id 7 --apply memorymaster --workspace . improve --scope project:example ``` -MCP exposes the same four operation names and response fields. Existing -advanced commands and tools remain available. +MCP exposes the same four operation names and response fields, including +`recall(include_observations=true, observation_limit=2)`. Existing advanced +commands and tools remain available. ## Capture boundary @@ -87,11 +97,17 @@ erasure workflow when the requirement is removal of sensitive payloads. ## Background processing and visibility -`improve` queues due claim extraction, steward review, and confirmed-claim -graph work. It never confirms or rewrites a claim in the caller’s request. -The existing Dreaming schedule drains the bounded queue under the shared -provider budgets. Capture status, evidence lineage, claims, citations, graph -supports, and preview/apply source retirement are visible in the dashboard’s -Capture Inbox. +`improve` queues due claim extraction, steward review, confirmed-claim graph +work, and observation discovery. It never runs synthesis, confirms, or rewrites +a claim in the caller's request. Its queue receipt reports observation discover +and synthesis counts separately. The existing Dreaming schedule drains the +bounded queue under the shared provider budgets. -Run `memorymaster --json demo` for a deterministic temporary-database example. +Capture status, evidence lineage, claims, citations, graph supports, and source +retirement are visible in the Capture Inbox. The Derived Observations panel +shows lifecycle status, type, evidence window, exact supporting +claims/evidence/relationships, diagnostics, and lifecycle history. + +Run `memorymaster --json demo` for a deterministic temporary-database example +that promotes a cited three-blocker observation and then proves automatic +staleness after support retirement. diff --git a/memorymaster/capture/repository.py b/memorymaster/capture/repository.py index c26ae94e..37b05d24 100644 --- a/memorymaster/capture/repository.py +++ b/memorymaster/capture/repository.py @@ -377,7 +377,20 @@ def retire_source(self, source_item_id: int, *, reason: str) -> bool: if not reason: raise ValueError("retirement reason is required") stamp = _iso(_now()) + observation_scopes: list[tuple[str, str | None]] = [] with self._connection() as conn: + if not self.postgres: + observation_scopes = [ + (str(row["scope"]), row["tenant_id"]) + for row in conn.execute( + """SELECT DISTINCT go.scope, go.tenant_id + FROM graph_observation_supports gos + JOIN graph_observations go + ON go.observation_claim_id=gos.observation_claim_id + WHERE gos.source_item_id=?""", + (source_item_id,), + ).fetchall() + ] cur = self._execute( conn, f"""UPDATE source_items SET retired_at={self.placeholder}, @@ -394,6 +407,22 @@ def retire_source(self, source_item_id: int, *, reason: str) -> bool: (stamp, source_item_id), ) self._commit(conn) + if changed and observation_scopes: + from memorymaster.knowledge.graph_observation_engine import ( + invalidate_changed_observations, + ) + from memorymaster.knowledge.graph_observation_repository import ( + GraphObservationRepository, + ) + + observations = GraphObservationRepository(self.store) + for scope, tenant_id in observation_scopes: + invalidate_changed_observations( + self.store, + observations, + scope=scope, + tenant_id=tenant_id, + ) return changed def status_counts(self) -> dict[str, int]: @@ -531,6 +560,10 @@ def due_confirmed_graph_claims(self, *, scope: str, limit: int) -> list[dict[str JOIN evidence_items e ON e.id=cel.evidence_item_id JOIN source_items s ON s.id=e.source_item_id WHERE c.id>{self.placeholder} AND c.status='confirmed' + AND COALESCE(c.claim_type, '') NOT IN + ('observation','skill','summary') + AND COALESCE(c.source_agent, '')<> + 'memorymaster-graph-observer' AND c.scope={self.placeholder} AND s.retired_at IS NULL GROUP BY c.id, c.updated_at ORDER BY c.id LIMIT {self.placeholder}""", diff --git a/memorymaster/dreaming/worker.py b/memorymaster/dreaming/worker.py index cbbedc05..3ee8863b 100644 --- a/memorymaster/dreaming/worker.py +++ b/memorymaster/dreaming/worker.py @@ -49,6 +49,13 @@ def _env_int(name: str, default: int) -> int: return default +def _env_bool(name: str, default: bool = False) -> bool: + raw = os.environ.get(name) + if raw is None: + return default + return raw.strip().lower() in {"1", "true", "yes", "on"} + + @dataclass(frozen=True, slots=True) class DreamConfig: idle_minutes: int = 30 @@ -63,6 +70,7 @@ class DreamConfig: lease_ttl_seconds: int = 900 retain_days: int = 7 max_capture_bytes: int = 256 * 1024 * 1024 + enable_graph_observations: bool = False @classmethod def from_env(cls) -> "DreamConfig": @@ -81,6 +89,9 @@ def from_env(cls) -> "DreamConfig": lease_ttl_seconds=_env_int("MEMORYMASTER_DREAM_LEASE_TTL_SECONDS", 900), retain_days=_env_int("MEMORYMASTER_DREAM_CAPTURE_RETAIN_DAYS", 7), max_capture_bytes=_env_int("MEMORYMASTER_DREAM_CAPTURE_MAX_BYTES", 256 * 1024 * 1024), + enable_graph_observations=_env_bool( + "MEMORYMASTER_GRAPH_OBSERVATIONS", False + ), ) @@ -121,6 +132,12 @@ def run(self, *, apply_candidates: bool, scope: str | None = None, max_sessions: if apply_candidates: pending = self.ledger.consolidated(max_sessions=limit, scope=scope) self._apply(run_id, pending, summary) + if self.config.enable_graph_observations: + summary["graph_observations"] = self._run_graph_observations( + owner=owner, + scope=scope, + synthesize=apply_candidates, + ) self.ledger.prune(retain_days=self.config.retain_days, max_bytes=self.config.max_capture_bytes, now=self.now()) self.ledger.finish_run(run_id, "ok" if not summary["errors"] else "partial", summary, now=self.now()) return summary @@ -131,6 +148,61 @@ def run(self, *, apply_candidates: bool, scope: str | None = None, max_sessions: finally: self.ledger.release_lease("dream-worker", owner) + def _observation_scope_pairs(self, scope: str | None) -> list[tuple[str, str | None]]: + params: tuple[Any, ...] = () if scope is None else (scope,) + clause = "" if scope is None else "AND scope=?" + with self.service.store.connect() as conn: + rows = conn.execute( + f"""SELECT DISTINCT scope, tenant_id FROM claims + WHERE status='confirmed' + AND COALESCE(claim_type, '') NOT IN + ('observation','skill','summary') + {clause} ORDER BY scope, tenant_id""", + params, + ).fetchall() + return [(str(row["scope"]), row["tenant_id"]) for row in rows] + + def _observation_llm(self, system: str, prompt: str) -> str: + from memorymaster.core.llm_provider import call_llm, use_call_scoped_env + + with use_call_scoped_env( + { + "MEMORYMASTER_LLM_PROVIDER": "opencode", + "MEMORYMASTER_LLM_MODEL": self.consolidator.model, + } + ): + return call_llm(system, prompt) + + def _run_graph_observations( + self, *, owner: str, scope: str | None, synthesize: bool + ) -> dict[str, int]: + from memorymaster.knowledge.graph_observation_engine import ( + GraphObservationEngine, + ) + from memorymaster.knowledge.ontology import load_ontology + + engine = GraphObservationEngine(self.service.store, llm_call=self._observation_llm) + totals = {"discovery_queued": 0, "synthesis_queued": 0, "emitted": 0, "failed": 0} + cycle_hour = self.now().astimezone(timezone.utc).strftime("%Y-%m-%dT%H") + scope_pairs = self._observation_scope_pairs(scope) + for target_scope, tenant_id in scope_pairs: + _job, created = engine.repo.queue_discovery( + tenant_id=tenant_id, + scope=target_scope, + ontology_version=load_ontology().version, + cycle_hour=cycle_hour, + ) + totals["discovery_queued"] += int(created) + for target_scope in sorted({item[0] for item in scope_pairs}): + discovered = engine.process_discovery(owner=owner, scope=target_scope) + totals["synthesis_queued"] += discovered.synthesis_queued + totals["failed"] += discovered.failed + if synthesize: + synthesized = engine.process_synthesis(owner=owner, scope=target_scope) + totals["emitted"] += synthesized.emitted + totals["failed"] += synthesized.failed + return totals + def _extract(self, run_id: str, scope: str | None, limit: int, summary: dict[str, Any]) -> list[dict[str, Any]]: rows = self.ledger.eligible(idle_minutes=self.config.idle_minutes, max_sessions=limit, scope=scope, now=self.now()) extracted: list[dict[str, Any]] = [] diff --git a/memorymaster/evaluation/graph_observation_evaluator.py b/memorymaster/evaluation/graph_observation_evaluator.py new file mode 100644 index 00000000..10868490 --- /dev/null +++ b/memorymaster/evaluation/graph_observation_evaluator.py @@ -0,0 +1,138 @@ +"""Offline structural evaluator for the versioned PPR-7 synthetic corpus.""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from memorymaster.knowledge.graph_observations import ( + ObservationSupport, + canonical_signature, + discover_components, +) + + +DEFAULT_CORPUS = Path(__file__).resolve().parents[2] / "benchmarks" / "fixtures" / "graph_observations_v1.json" +EXCLUDED_STATES = frozenset( + { + "candidate", + "conflicted", + "observation", + "observer_agent", + "retired", + "sensitive_claim", + "sensitive_evidence", + "sensitive_source", + "skill", + "stale", + "summary", + "wrong_scope", + "wrong_tenant", + } +) + + +def _row(claim: int, evidence: int, source: int, edge: tuple[int, str, int]) -> ObservationSupport: + signature = canonical_signature( + edge[0], + edge[1], + edge[2], + "personal-v1", + symmetric_relations=frozenset({"related_to"}), + ) + return ObservationSupport( + claim, + evidence, + source, + signature[0], + signature[1], + signature[2], + signature[3], + "project:evaluator", + "tenant-evaluator", + 0.8, + f"2026-08-{min(evidence, 28):02d}T00:00:00+00:00", + ) + + +def _eligible(offset: int = 0) -> list[ObservationSupport]: + return [ + _row(1 + offset, 1 + offset, 101 + offset, (10 + offset, "depends_on", 20 + offset)), + _row(2 + offset, 2 + offset, 102 + offset, (10 + offset, "depends_on", 20 + offset)), + _row(2 + offset, 2 + offset, 102 + offset, (20 + offset, "depends_on", 30 + offset)), + _row(3 + offset, 1 + offset, 101 + offset, (20 + offset, "depends_on", 30 + offset)), + ] + + +def _template(name: str) -> list[ObservationSupport]: + if name in {"eligible", "cross_scope", "cross_tenant", "excluded"}: + return _eligible() + if name == "merge": + return _eligible() + [_row(4, 3, 103, (20, "depends_on", 30))] + if name == "split": + return _eligible() + _eligible(3) + if name == "unrelated": + return [_row(i, i, 100 + i, (i, "uses", 50 + i)) for i in range(1, 5)] + if name == "insufficient_claims": + return _eligible()[:2] + if name == "insufficient_evidence": + return [_row(i, 1, 101, (10 + i, "uses", 20 + i)) for i in range(1, 4)] + if name == "one_signature": + return [_row(i, 1 + i % 2, 101 + i % 2, (10, "uses", 20)) for i in range(1, 4)] + if name == "symmetric": + return [ + _row(1, 1, 101, (10, "related_to", 20)), + _row(2, 2, 102, (20, "related_to", 10)), + _row(2, 2, 102, (20, "uses", 30)), + _row(3, 1, 101, (20, "uses", 30)), + ] + if name == "hub": + return [_row(i, i, 100 + i, (10, "related_to", 20)) for i in range(1, 22)] + if name == "oversized": + rows = [_row(i, i, 100 + i, (i, "uses", i + 1)) for i in range(1, 22)] + rows.extend(_row(i, i, 100 + i, (i - 1, "uses", i)) for i in range(2, 22)) + return rows + raise ValueError(f"unknown corpus template: {name}") + + +def _supports(case: dict[str, Any]) -> list[ObservationSupport]: + state = str(case.get("state") or "") + if state in EXCLUDED_STATES: + return [] + return _template(str(case["template"])) + + +def evaluate_corpus(path: Path = DEFAULT_CORPUS) -> dict[str, Any]: + """Evaluate exact predicted claim groups against the versioned oracle.""" + corpus = json.loads(path.read_text(encoding="utf-8")) + tp = fp = fn = 0 + failures: list[str] = [] + for case in corpus["cases"]: + result = discover_components(_supports(case), scope="project:evaluator", tenant_id="tenant-evaluator") + predicted = {tuple(component.claim_ids) for component in result.components} + expected = {tuple(group) for group in case["expected"]} + tp += len(predicted & expected) + fp += len(predicted - expected) + fn += len(expected - predicted) + if predicted != expected: + failures.append(str(case["id"])) + precision = tp / (tp + fp) if tp + fp else 1.0 + recall = tp / (tp + fn) if tp + fn else 1.0 + return { + "corpus_version": corpus["corpus_version"], + "cases": len(corpus["cases"]), + "precision": precision, + "recall": recall, + "failures": failures, + } + + +def main() -> int: + report = evaluate_corpus() + print(json.dumps(report, sort_keys=True)) + return int(report["precision"] < 0.95 or report["recall"] < 0.95) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/memorymaster/govern/llm_steward.py b/memorymaster/govern/llm_steward.py index ecef58ad..02bb8c0d 100644 --- a/memorymaster/govern/llm_steward.py +++ b/memorymaster/govern/llm_steward.py @@ -552,6 +552,23 @@ def run_steward( Returns summary stats dict. """ + from memorymaster.govern.jobs.deterministic import open_store + from memorymaster.knowledge.graph_observation_engine import ( + review_observation_candidates, + ) + + effective_store = store or open_store(db_path) + try: + observation_review = review_observation_candidates( + effective_store, + scope=scope, + limit=limit, + apply=not dry_run, + ) + except Exception as exc: # noqa: BLE001 - observations never fall through to classifier + log.warning("Observation steward gate failed closed: %s", exc) + observation_review = {"checked": 0, "confirmed": 0, "archived": 0, "errors": 1} + # Build key rotator if multiple keys available effective_keys = api_keys if api_keys else [api_key] if api_key else [""] key_rotator: KeyRotator | None = None @@ -575,12 +592,14 @@ def run_steward( if scope is not None: candidates = conn.execute( "SELECT id, text, scope, version FROM claims WHERE status = 'candidate' " + "AND COALESCE(claim_type, '') <> 'observation' " "AND text IS NOT NULL AND scope = ? ORDER BY id LIMIT ?", (scope, limit), ).fetchall() else: candidates = conn.execute( "SELECT id, text, scope, version FROM claims WHERE status = 'candidate' " + "AND COALESCE(claim_type, '') <> 'observation' " "AND text IS NOT NULL ORDER BY id LIMIT ?", (limit,), ).fetchall() @@ -614,6 +633,7 @@ def run_steward( "dedupe_score_sum": 0.0, "dedupe_score_count": 0, "results": [], + "observation_review": observation_review, } for row in candidates: @@ -882,7 +902,7 @@ def run_steward( try: unique_ids = list(dict.fromkeys(confirmed_claim_ids)) validation_stats = _auto_validate_claims( - db_path, unique_ids, workspace_root, store=store, + db_path, unique_ids, workspace_root, store=effective_store, ) stats["auto_validation"] = validation_stats except Exception as e: diff --git a/memorymaster/knowledge/context_bundle.py b/memorymaster/knowledge/context_bundle.py index c3a536a1..b8e239b3 100644 --- a/memorymaster/knowledge/context_bundle.py +++ b/memorymaster/knowledge/context_bundle.py @@ -13,6 +13,13 @@ from memorymaster.knowledge.skill_schema import is_skill from memorymaster.knowledge.skills import recall_skills +from memorymaster.knowledge.graph_observation_recall import ( + pack_observations, + recall_observations, +) +from memorymaster.knowledge.graph_observation_repository import ( + GraphObservationRepository, +) from memorymaster.recall.context_optimizer import estimate_tokens, pack_context @@ -32,6 +39,7 @@ class ContextBundle: tokens_used: int token_budget: int output_format: str + observations: tuple[dict[str, Any], ...] = () def _lines(label: str, values: list[str], *, numbered: bool = False) -> list[str]: @@ -92,36 +100,71 @@ def query_context_bundle( retrieval_mode: str = "hybrid", include_skills: bool = False, skill_limit: int = 3, + include_observations: bool = False, + observation_limit: int = 2, + observation_tenant_id: str | None = None, ) -> ContextBundle: """Query governed claims and optionally append confirmed scoped skills.""" if token_budget <= 0: raise ValueError("token_budget must be positive.") - if include_skills and output_format != "text": - raise ValueError("Approved skill bundles require text output format.") + if (include_skills or include_observations) and output_format != "text": + raise ValueError("Derived recall sections require text output format.") + observation_text = "" + observations: tuple[dict[str, Any], ...] = () + if include_observations: + observation_budget = min(800, max(1, token_budget // 4)) + candidates = recall_observations( + service, + query, + scopes=scope_allowlist, + trust_mode=trust_mode, + limit=max(1, min(int(observation_limit), 5)), + tenant_id=observation_tenant_id, + ) + observation_text, observations = pack_observations( + candidates, token_budget=observation_budget + ) + observation_reserved = estimate_tokens(observation_text) + 1 if observation_text else 0 skill_text, skills = _selected_skill_text( service, query, scopes=scope_allowlist, - total_budget=token_budget, + total_budget=max(1, token_budget - observation_reserved), include_skills=include_skills, skill_limit=skill_limit, ) - reserved = estimate_tokens(skill_text) + 1 if skill_text else 0 + skill_reserved = estimate_tokens(skill_text) + 1 if skill_text else 0 + reserved = observation_reserved + skill_reserved + observation_count = sum( + len( + GraphObservationRepository(service.store).scope_observations( + scope=scope, tenant_id=observation_tenant_id + ) + ) + for scope in scope_allowlist + ) result = service.query_for_context( query=query, token_budget=max(1, token_budget - reserved), + limit=100 + min(observation_count, 400), output_format=output_format, retrieval_mode=retrieval_mode, trust_mode=trust_mode, scope_allowlist=scope_allowlist, ) + ordinary_rows = [ + row for row in result.rows if getattr(row["claim"], "claim_type", None) != "observation" + ][:100] if include_skills: + ordinary_rows = [row for row in ordinary_rows if not is_skill(row["claim"])] + if len(ordinary_rows) != len(result.rows): result = pack_context( - [row for row in result.rows if not is_skill(row["claim"])], + ordinary_rows, token_budget=max(1, token_budget - reserved), output_format=output_format, ) - output = f"{result.output}\n\n{skill_text}" if skill_text else result.output + sections = [text for text in (result.output, observation_text, skill_text) if text] + output = "\n\n".join(sections) return ContextBundle( output=output, rows=result.rows, @@ -129,6 +172,7 @@ def query_context_bundle( tokens_used=result.tokens_used + reserved, token_budget=token_budget, output_format=result.format, + observations=observations, ) diff --git a/memorymaster/knowledge/graph_observation_engine.py b/memorymaster/knowledge/graph_observation_engine.py new file mode 100644 index 00000000..4acf4916 --- /dev/null +++ b/memorymaster/knowledge/graph_observation_engine.py @@ -0,0 +1,293 @@ +"""Governed discovery, synthesis, and steward review for graph observations.""" + +from __future__ import annotations + +import json +from dataclasses import dataclass +from typing import Any, Callable + +from memorymaster.core.lifecycle import transition_claim +from memorymaster.core.llm_provider import call_llm +from memorymaster.core.models import CitationInput +from memorymaster.knowledge.graph_observation_repository import ( + GraphObservationRepository, + ObservationJob, +) +from memorymaster.knowledge.graph_observations import ( + ObservationComponent, + ObservationDraft, + discover_components, + parse_synthesis_output, +) + + +MAX_SYNTHESIS_CALLS_PER_SCOPE = 3 +OBSERVER_AGENT = "memorymaster-graph-observer" +SYSTEM_PROMPT = """You summarize one deterministic, evidence-bound graph component. +Return one JSON object only. Never add facts or claim IDs. The schema is either +{"decision":"no_signal"} or {"decision":"emit","name":"...", +"observation_type":"decision|commitment|constraint|dependency|state_change|recurring_pattern|stable_relationship|root_cause", +"summary":"...","assertions":[{"text":"...","supporting_claim_ids":[1]}]}. +Use no_signal when the supplied claims do not support a useful higher-level observation.""" + + +@dataclass(frozen=True, slots=True) +class ObservationCycleResult: + discovery_completed: int = 0 + synthesis_queued: int = 0 + synthesis_completed: int = 0 + emitted: int = 0 + no_signal: int = 0 + invalidated: int = 0 + failed: int = 0 + + +def _component_from_job( + repo: GraphObservationRepository, job: ObservationJob +) -> ObservationComponent | None: + supports = repo.supports_from_manifest( + scope=job.scope, + tenant_id=job.tenant_id, + manifest_json=job.support_manifest_json, + ) + result = discover_components(supports, scope=job.scope, tenant_id=job.tenant_id) + return next( + (item for item in result.components if item.support_hash == job.support_hash), + None, + ) + + +def _prompt(store: Any, component: ObservationComponent) -> str: + claims = [] + for claim_id in component.claim_ids: + claim = store.get_claim(claim_id, include_citations=False) + if claim is None: + raise ValueError("supporting claim disappeared") + claims.append({"claim_id": claim.id, "text": claim.text}) + edges = [ + { + "claim_id": row.claim_id, + "evidence_id": row.evidence_id, + "source_item_id": row.source_item_id, + "signature": list(row.signature), + } + for row in component.supports + ] + return json.dumps( + {"support_hash": component.support_hash, "claims": claims, "edges": edges}, + sort_keys=True, + ) + + +def _citations(component: ObservationComponent) -> list[CitationInput]: + return [ + CitationInput(source="evidence", locator=f"evidence:{evidence_id}") + for evidence_id in component.evidence_ids + ] + + +def _create_candidate( + store: Any, + repo: GraphObservationRepository, + draft: ObservationDraft, + component: ObservationComponent, +) -> int: + existing = repo.observation_for_support( + scope=component.scope, + tenant_id=component.tenant_id, + support_hash=component.support_hash, + ) + if existing is not None: + return existing + claim = store.create_claim( + text=draft.summary, + citations=_citations(component), + idempotency_key=f"graph-observation:{component.support_hash}", + claim_type="observation", + subject=draft.name, + predicate="graph_observation", + object_value=draft.observation_type, + scope=component.scope, + confidence=min(row.confidence for row in component.supports), + tenant_id=component.tenant_id, + source_agent=OBSERVER_AGENT, + ) + repo.persist_observation( + observation_claim_id=claim.id, draft=draft, component=component + ) + return claim.id + + +def invalidate_changed_observations( + store: Any, + repo: GraphObservationRepository, + *, + scope: str, + tenant_id: str | None, + current_hashes: set[str] | None = None, +) -> int: + if current_hashes is None: + active = repo.load_active_supports(scope=scope, tenant_id=tenant_id) + current_hashes = { + item.support_hash + for item in discover_components( + active, scope=scope, tenant_id=tenant_id + ).components + } + changed = 0 + for row in repo.scope_observations(scope=scope, tenant_id=tenant_id): + if row["support_hash"] in current_hashes: + continue + status = str(row["status"]) + target = "archived" if status == "candidate" else "stale" + if status not in {"candidate", "confirmed"}: + continue + transition_claim( + store, + int(row["observation_claim_id"]), + target, + "graph-observation support fingerprint changed", + event_type="staleness", + ) + changed += 1 + return changed + + +class GraphObservationEngine: + def __init__( + self, + store: Any, + *, + llm_call: Callable[[str, str], str] = call_llm, + ) -> None: + self.store = store + self.repo = GraphObservationRepository(store) + self.llm_call = llm_call + + def process_discovery(self, *, owner: str, scope: str, limit: int = 10) -> ObservationCycleResult: + completed = queued = invalidated = failed = 0 + jobs = self.repo.lease_jobs( + owner=owner, limit=limit, stages=("discover",), scope=scope + ) + for job in jobs: + try: + supports = self.repo.load_active_supports( + scope=job.scope, tenant_id=job.tenant_id + ) + result = discover_components( + supports, scope=job.scope, tenant_id=job.tenant_id + ) + hashes = {component.support_hash for component in result.components} + invalidated += invalidate_changed_observations( + self.store, + self.repo, + scope=job.scope, + tenant_id=job.tenant_id, + current_hashes=hashes, + ) + for component in result.components: + if self._queue_component(job, component): + queued += 1 + codes = [item.code for item in result.diagnostics] + self.repo.complete_job(job.id, owner=owner, diagnostic_codes=codes) + completed += 1 + except Exception: # noqa: BLE001 - typed retry boundary persisted below + self.repo.fail_job(job.id, owner=owner, error_code="discovery_failed") + failed += 1 + return ObservationCycleResult(completed, queued, invalidated=invalidated, failed=failed) + + def _queue_component(self, job: ObservationJob, component: ObservationComponent) -> bool: + if self.repo.observation_for_support( + scope=component.scope, + tenant_id=component.tenant_id, + support_hash=component.support_hash, + ) is not None: + return False + _queued, created = self.repo.queue_job( + tenant_id=component.tenant_id, + scope=component.scope, + stage="synthesize", + content_hash=component.support_hash, + support_hash=component.support_hash, + ontology_version=job.ontology_version, + support_manifest=self.repo.component_manifest(component), + ) + return created + + def process_synthesis(self, *, owner: str, scope: str) -> ObservationCycleResult: + completed = emitted = no_signal = failed = 0 + jobs = self.repo.lease_jobs( + owner=owner, + limit=MAX_SYNTHESIS_CALLS_PER_SCOPE, + stages=("synthesize",), + scope=scope, + ) + for job in jobs: + try: + component = _component_from_job(self.repo, job) + if component is None: + self.repo.fail_job(job.id, owner=owner, error_code="support_changed") + failed += 1 + continue + raw = self.llm_call(SYSTEM_PROMPT, _prompt(self.store, component)) + draft = parse_synthesis_output(raw, allowed_claim_ids=component.claim_ids) + if draft.decision == "emit": + _create_candidate(self.store, self.repo, draft, component) + emitted += 1 + else: + no_signal += 1 + self.repo.complete_job(job.id, owner=owner) + completed += 1 + except Exception: # noqa: BLE001 - fail closed and retry from IDs + self.repo.fail_job(job.id, owner=owner, error_code="synthesis_failed") + failed += 1 + return ObservationCycleResult( + synthesis_completed=completed, + emitted=emitted, + no_signal=no_signal, + failed=failed, + ) + + +def review_observation_candidates( + store: Any, + *, + scope: str | None = None, + limit: int = 50, + apply: bool = True, +) -> dict[str, int]: + repo = GraphObservationRepository(store) + with repo._connection() as conn: + params: tuple[Any, ...] = () if scope is None else (scope,) + clause = "" if scope is None else "AND scope=?" + rows = conn.execute( + f"""SELECT id FROM claims WHERE status='candidate' + AND claim_type='observation' {clause} ORDER BY id LIMIT ?""", + (*params, limit), + ).fetchall() + stats = {"checked": len(rows), "confirmed": 0, "archived": 0, "would_confirm": 0} + for row in rows: + claim_id = int(row["id"]) + eligible, reason = repo.observation_gate(claim_id) + if eligible: + if apply: + transition_claim( + store, + claim_id, + "confirmed", + "graph-observation deterministic support gate passed", + event_type="deterministic_validator", + ) + stats["confirmed"] += 1 + else: + stats["would_confirm"] += 1 + elif apply: + transition_claim( + store, + claim_id, + "archived", + f"graph-observation gate failed: {reason}", + event_type="deterministic_validator", + ) + stats["archived"] += 1 + return stats diff --git a/memorymaster/knowledge/graph_observation_recall.py b/memorymaster/knowledge/graph_observation_recall.py new file mode 100644 index 00000000..456695a0 --- /dev/null +++ b/memorymaster/knowledge/graph_observation_recall.py @@ -0,0 +1,130 @@ +"""Read-time revalidation and bounded packing for opt-in observations.""" + +from __future__ import annotations + +import re +from typing import Any + +from memorymaster.knowledge.graph_observation_repository import ( + GraphObservationRepository, +) +from memorymaster.recall.context_optimizer import estimate_tokens + + +OBSERVATION_HEADER = "=== DERIVED OBSERVATIONS ===\nEvidence-bound patterns; lifecycle labels are authoritative." +_WORD = re.compile(r"[a-z0-9_]{2,}") + + +def _score(query: str, row: dict[str, Any]) -> tuple[int, int]: + terms = set(_WORD.findall(query.lower())) + text = f"{row['name']} {row['text']} {row['observation_type']}".lower() + claim_id = row.get("observation_claim_id", row.get("claim_id", 0)) + return sum(term in text for term in terms), int(claim_id) + + +def _public_row( + repo: GraphObservationRepository, + row: dict[str, Any], + *, + support_valid: bool, +) -> dict[str, Any]: + supports = repo.observation_support_rows(int(row["observation_claim_id"])) + relationships = sorted( + { + ( + int(item["source_entity_id"]), + str(item["relation"]), + int(item["target_entity_id"]), + str(item["ontology_version"]), + ) + for item in supports + } + ) + return { + "claim_id": int(row["observation_claim_id"]), + "name": str(row["name"]), + "observation_type": str(row["observation_type"]), + "summary": str(row["text"]), + "status": str(row["status"]), + "scope": str(row["scope"]), + "confidence": float(row["confidence"]), + "support_hash": str(row["support_hash"]), + "support_valid": support_valid, + "evidence_window_start": row["evidence_window_start"], + "evidence_window_end": row["evidence_window_end"], + "supporting_claim_ids": sorted({int(item["supporting_claim_id"]) for item in supports}), + "evidence_item_ids": sorted({int(item["evidence_item_id"]) for item in supports}), + "source_item_ids": sorted({int(item["source_item_id"]) for item in supports}), + "relationships": relationships, + } + + +def recall_observations( + service: Any, + query: str, + *, + scopes: list[str], + trust_mode: str, + limit: int, + tenant_id: str | None = None, +) -> list[dict[str, Any]]: + repo = GraphObservationRepository(service.store) + candidates: list[dict[str, Any]] = [] + for scope in scopes: + for row in repo.scope_observations(scope=scope, tenant_id=tenant_id): + status = str(row["status"]) + valid, _reason = repo.observation_gate(int(row["observation_claim_id"])) + if trust_mode == "trusted" and (status != "confirmed" or not valid): + continue + if trust_mode != "trusted" and status not in {"candidate", "confirmed", "stale"}: + continue + candidates.append(_public_row(repo, row, support_valid=valid)) + candidates.sort(key=lambda row: _score(query, {**row, "text": row["summary"]}), reverse=True) + return candidates[: max(1, min(int(limit), 5))] + + +def _block(observation: dict[str, Any]) -> str: + window = " -> ".join( + str(value or "unknown") + for value in ( + observation["evidence_window_start"], + observation["evidence_window_end"], + ) + ) + relationships = ", ".join( + f"{source}:{relation}:{target}@{version}" + for source, relation, target, version in observation["relationships"] + ) + return "\n".join( + [ + ( + f"[observation claim_id={observation['claim_id']} " + f"status={observation['status']} type={observation['observation_type']}]" + ), + f"Name: {observation['name']}", + f"Summary: {observation['summary']}", + f"Evidence window: {window}", + f"Supporting claims: {observation['supporting_claim_ids']}", + f"Evidence items: {observation['evidence_item_ids']}", + f"Relationships: {relationships}", + ] + ) + + +def pack_observations( + observations: list[dict[str, Any]], *, token_budget: int +) -> tuple[str, tuple[dict[str, Any], ...]]: + if token_budget <= estimate_tokens(OBSERVATION_HEADER): + return "", () + selected: list[dict[str, Any]] = [] + blocks: list[str] = [] + for observation in observations: + block = _block(observation) + candidate = f"{OBSERVATION_HEADER}\n\n" + "\n\n".join((*blocks, block)) + if estimate_tokens(candidate) > token_budget: + continue + selected.append(observation) + blocks.append(block) + if not blocks: + return "", () + return f"{OBSERVATION_HEADER}\n\n" + "\n\n".join(blocks), tuple(selected) diff --git a/memorymaster/knowledge/graph_observation_repository.py b/memorymaster/knowledge/graph_observation_repository.py new file mode 100644 index 00000000..531a6729 --- /dev/null +++ b/memorymaster/knowledge/graph_observation_repository.py @@ -0,0 +1,556 @@ +"""SQLite repository for replay-safe graph-observation work and lineage.""" + +from __future__ import annotations + +import contextlib +import hashlib +import json +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from typing import Any, Iterable, Iterator + +from memorymaster.knowledge.graph_observations import ( + ALGORITHM_VERSION, + ObservationComponent, + ObservationDraft, + ObservationSupport, + canonical_signature, + discover_components, + support_fingerprint, +) +from memorymaster.knowledge.ontology import load_ontology + + +MAX_ATTEMPTS = 5 +MAX_RETRY_SECONDS = 6 * 60 * 60 +JOB_STATUSES = frozenset( + {"pending", "leased", "retryable", "blocked", "completed", "cancelled"} +) +JOB_STAGES = frozenset({"discover", "synthesize"}) + + +def _now() -> datetime: + return datetime.now(timezone.utc).replace(microsecond=0) + + +def _iso(value: datetime) -> str: + return value.isoformat() + + +def _mapping(row: Any) -> dict[str, Any]: + if isinstance(row, dict): + return dict(row) + if hasattr(row, "keys"): + return {key: row[key] for key in row.keys()} + raise TypeError("graph observation repository requires mapping-compatible rows") + + +def _digest(payload: Any) -> str: + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + return hashlib.sha256(encoded).hexdigest() + + +def _observation_support(row: Any, symmetric: frozenset[str]) -> ObservationSupport: + data = _mapping(row) + signature = canonical_signature( + data["source_entity_id"], + str(data["relation"]).strip().lower(), + data["target_entity_id"], + data["ontology_version"], + symmetric_relations=symmetric, + ) + return ObservationSupport( + claim_id=int(data["claim_id"]), + evidence_id=int(data["evidence_id"]), + source_item_id=int(data["source_item_id"]), + source_entity_id=signature[0], + relation=signature[1], + target_entity_id=signature[2], + ontology_version=signature[3], + scope=str(data["scope"]), + tenant_id=data["tenant_id"], + confidence=float(data["confidence"]), + occurred_at=data["occurred_at"], + ) + + +@dataclass(frozen=True, slots=True) +class ObservationJob: + id: int + tenant_id: str | None + scope: str + stage: str + status: str + content_hash: str + support_hash: str | None + algorithm_version: str + ontology_version: str + support_manifest_json: str + attempts: int + next_attempt_at: str | None + lease_owner: str | None + lease_expires_at: str | None + error_code: str | None + diagnostic_hash: str | None + created_at: str + updated_at: str + completed_at: str | None + + +class GraphObservationRepository: + """Persist PPR-7 state while explicitly rejecting PostgreSQL runtimes.""" + + def __init__(self, store: Any) -> None: + if hasattr(store, "dsn"): + raise RuntimeError("graph observations are SQLite-only") + self.store = store + + @contextlib.contextmanager + def _connection(self) -> Iterator[Any]: + conn = self.store.connect() + try: + yield conn + except Exception: + conn.rollback() + raise + finally: + conn.close() + + @staticmethod + def _job(row: Any) -> ObservationJob: + data = _mapping(row) + fields = ObservationJob.__dataclass_fields__ + return ObservationJob(**{name: data.get(name) for name in fields}) + + def queue_job(self, *, tenant_id: str | None, scope: str, stage: str, content_hash: str, ontology_version: str, support_hash: str | None = None, support_manifest: Iterable[Iterable[Any]] = ()) -> tuple[ObservationJob, bool]: + if stage not in JOB_STAGES or len(content_hash) != 64: + raise ValueError("invalid graph observation job identity") + manifest_json = json.dumps( + sorted(list(item) for item in support_manifest), separators=(",", ":") + ) + stamp = _iso(_now()) + values = ( + tenant_id, + scope, + stage, + "pending", + content_hash, + support_hash, + ALGORITHM_VERSION, + ontology_version, + manifest_json, + stamp, + stamp, + ) + with self._connection() as conn: + cur = conn.execute( + """INSERT INTO graph_observation_jobs + (tenant_id, scope, stage, status, content_hash, support_hash, + algorithm_version, ontology_version, support_manifest_json, + attempts, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?) + ON CONFLICT DO NOTHING""", + values, + ) + created = cur.rowcount > 0 + row = conn.execute( + """SELECT * FROM graph_observation_jobs + WHERE tenant_id IS ? AND scope=? AND stage=? AND content_hash=? + AND algorithm_version=? AND ontology_version=?""", + (tenant_id, scope, stage, content_hash, ALGORITHM_VERSION, ontology_version), + ).fetchone() + conn.commit() + if row is None: + raise RuntimeError("graph observation job insert returned no row") + return self._job(row), created + + def queue_discovery( + self, + *, + tenant_id: str | None, + scope: str, + ontology_version: str, + cycle_hour: str, + ) -> tuple[ObservationJob, bool]: + identity = _digest( + ["discover", tenant_id or "", scope, cycle_hour, ALGORITHM_VERSION, ontology_version] + ) + return self.queue_job( + tenant_id=tenant_id, + scope=scope, + stage="discover", + content_hash=identity, + ontology_version=ontology_version, + ) + + def _expire_leases(self, conn: Any, stamp: str) -> None: + conn.execute( + """UPDATE graph_observation_jobs + SET status='blocked', error_code='attempts_exhausted', + lease_owner=NULL, lease_expires_at=NULL, updated_at=? + WHERE status='leased' AND lease_expires_at<=? AND attempts>=5""", + (stamp, stamp), + ) + conn.execute( + """UPDATE graph_observation_jobs + SET status='retryable', error_code='lease_expired', next_attempt_at=?, + lease_owner=NULL, lease_expires_at=NULL, updated_at=? + WHERE status='leased' AND lease_expires_at<=? AND attempts<5""", + (stamp, stamp, stamp), + ) + + def lease_jobs( + self, + *, + owner: str, + limit: int, + lease_seconds: int = 300, + stages: tuple[str, ...] = (), + scope: str | None = None, + ) -> list[ObservationJob]: + if not owner.strip() or limit <= 0: + return [] + if any(stage not in JOB_STAGES for stage in stages): + raise ValueError("invalid graph observation stage filter") + stamp = _iso(_now()) + expiry = _iso(_now() + timedelta(seconds=max(1, lease_seconds))) + with self._connection() as conn: + self._expire_leases(conn, stamp) + clauses = ["status IN ('pending','retryable')", "attempts<5"] + clauses.append("(next_attempt_at IS NULL OR next_attempt_at<=?)") + params: list[Any] = [stamp] + if stages: + clauses.append(f"stage IN ({','.join('?' for _ in stages)})") + params.extend(stages) + if scope is not None: + clauses.append("scope=?") + params.append(scope) + ids = [ + int(row["id"]) + for row in conn.execute( + f"""SELECT id FROM graph_observation_jobs + WHERE {' AND '.join(clauses)} ORDER BY id LIMIT ?""", + (*params, limit), + ).fetchall() + ] + if ids: + marks = ",".join("?" for _ in ids) + conn.execute( + f"""UPDATE graph_observation_jobs + SET status='leased', attempts=attempts+1, lease_owner=?, + lease_expires_at=?, updated_at=? WHERE id IN ({marks})""", + (owner, expiry, stamp, *ids), + ) + rows = conn.execute( + f"SELECT * FROM graph_observation_jobs WHERE id IN ({','.join('?' for _ in ids)})" + if ids + else "SELECT * FROM graph_observation_jobs WHERE 0", + tuple(ids), + ).fetchall() + conn.commit() + return [self._job(row) for row in sorted(rows, key=lambda row: int(row["id"]))] + + def complete_job( + self, + job_id: int, + *, + owner: str, + diagnostic_codes: Iterable[str] = (), + ) -> bool: + stamp = _iso(_now()) + codes = sorted(set(str(code) for code in diagnostic_codes)) + diagnostic_hash = _digest(codes) if codes else None + with self._connection() as conn: + cur = conn.execute( + """UPDATE graph_observation_jobs + SET status='completed', completed_at=?, updated_at=?, + lease_owner=NULL, lease_expires_at=NULL, diagnostic_hash=? + WHERE id=? AND status='leased' AND lease_owner=?""", + (stamp, stamp, diagnostic_hash, job_id, owner), + ) + conn.commit() + return cur.rowcount > 0 + + def fail_job(self, job_id: int, *, owner: str, error_code: str) -> bool: + stamp_dt = _now() + with self._connection() as conn: + row = conn.execute( + "SELECT attempts FROM graph_observation_jobs WHERE id=?", + (job_id,), + ).fetchone() + if row is None: + return False + attempts = int(row["attempts"]) + blocked = attempts >= MAX_ATTEMPTS + delay = min(MAX_RETRY_SECONDS, 60 * (2 ** max(0, attempts - 1))) + next_attempt = None if blocked else _iso(stamp_dt + timedelta(seconds=delay)) + cur = conn.execute( + """UPDATE graph_observation_jobs + SET status=?, error_code=?, next_attempt_at=?, updated_at=?, + lease_owner=NULL, lease_expires_at=NULL + WHERE id=? AND status='leased' AND lease_owner=?""", + ( + "blocked" if blocked else "retryable", + error_code, + next_attempt, + _iso(stamp_dt), + job_id, + owner, + ), + ) + conn.commit() + return cur.rowcount > 0 + + def load_active_supports( + self, *, scope: str, tenant_id: str | None + ) -> tuple[ObservationSupport, ...]: + ontology = load_ontology() + symmetric = frozenset( + name for name, definition in ontology.relations.items() if definition.symmetric + ) + relations = tuple(sorted(ontology.relations)) + relation_marks = ",".join("?" for _ in relations) + with self._connection() as conn: + rows = conn.execute( + f"""SELECT ees.supporting_claim_id AS claim_id, + cel.evidence_item_id AS evidence_id, + e.source_item_id, ees.source_entity_id, ees.relation, + ees.target_entity_id, ees.ontology_version, c.scope, + c.tenant_id, c.confidence, s.occurred_at + FROM entity_edge_supports ees + JOIN claims c ON c.id=ees.supporting_claim_id + JOIN claim_evidence_links cel ON cel.claim_id=c.id + JOIN evidence_items e ON e.id=cel.evidence_item_id + JOIN source_items s ON s.id=e.source_item_id + WHERE c.scope=? AND c.tenant_id IS ? AND ees.scope=c.scope + AND c.status='confirmed' AND c.visibility<>'sensitive' + AND COALESCE(c.claim_type, '') NOT IN ('observation','skill','summary') + AND COALESCE(c.source_agent, '')<>'memorymaster-graph-observer' + AND s.retired_at IS NULL + AND s.sensitivity='none' AND e.sensitivity='none' + AND ees.ontology_version=? + AND ees.relation IN ({relation_marks}) + ORDER BY cel.evidence_item_id, c.id, ees.source_entity_id, + ees.relation, ees.target_entity_id, ees.ontology_version""", + (scope, tenant_id, ontology.version, *relations), + ).fetchall() + return tuple(_observation_support(row, symmetric) for row in rows) + + @staticmethod + def component_manifest(component: ObservationComponent) -> list[list[Any]]: + return [ + [ + row.claim_id, + row.evidence_id, + row.source_item_id, + row.source_entity_id, + row.relation, + row.target_entity_id, + row.ontology_version, + ] + for row in component.supports + ] + + def supports_from_manifest( + self, + *, + scope: str, + tenant_id: str | None, + manifest_json: str, + ) -> tuple[ObservationSupport, ...]: + requested = {tuple(row) for row in json.loads(manifest_json)} + active = self.load_active_supports(scope=scope, tenant_id=tenant_id) + return tuple( + row + for row in active + if ( + row.claim_id, + row.evidence_id, + row.source_item_id, + row.source_entity_id, + row.relation, + row.target_entity_id, + row.ontology_version, + ) + in requested + ) + + def support_is_current( + self, + *, + scope: str, + tenant_id: str | None, + manifest_json: str, + expected_hash: str, + ) -> tuple[bool, tuple[ObservationSupport, ...]]: + requested = json.loads(manifest_json) + active = self.supports_from_manifest( + scope=scope, tenant_id=tenant_id, manifest_json=manifest_json + ) + current = len(active) == len(requested) and support_fingerprint(active) == expected_hash + claims = {row.claim_id for row in active} + evidence = {row.evidence_id for row in active} + sources = {row.source_item_id for row in active} + confidence = min((row.confidence for row in active), default=0.0) + gate = len(claims) >= 3 and len(evidence) >= 2 and len(sources) >= 2 + return current and gate and confidence >= 0.65, active + + def persist_supports( + self, observation_claim_id: int, component: ObservationComponent + ) -> None: + stamp = _iso(_now()) + with self._connection() as conn: + for row in component.supports: + conn.execute( + """INSERT OR IGNORE INTO graph_observation_supports + (observation_claim_id, supporting_claim_id, evidence_item_id, + source_item_id, source_entity_id, target_entity_id, relation, + ontology_version, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + observation_claim_id, + row.claim_id, + row.evidence_id, + row.source_item_id, + row.source_entity_id, + row.target_entity_id, + row.relation, + row.ontology_version, + stamp, + ), + ) + conn.execute( + """INSERT OR IGNORE INTO claim_links + (source_id, target_id, link_type, created_at) + VALUES (?, ?, 'derived_from', ?)""", + (observation_claim_id, row.claim_id, stamp), + ) + conn.execute( + """INSERT OR IGNORE INTO claim_evidence_links + (claim_id, evidence_item_id, role, created_at) + VALUES (?, ?, 'observation_support', ?)""", + (observation_claim_id, row.evidence_id, stamp), + ) + conn.commit() + + def persist_observation( + self, + *, + observation_claim_id: int, + draft: ObservationDraft, + component: ObservationComponent, + ) -> None: + stamp = _iso(_now()) + ontology_versions = {row.ontology_version for row in component.supports} + if len(ontology_versions) != 1: + raise ValueError("one observation cannot span ontology versions") + with self._connection() as conn: + conn.execute( + """INSERT OR IGNORE INTO graph_observations + (observation_claim_id, observation_type, name, scope, tenant_id, + support_hash, algorithm_version, ontology_version, + evidence_window_start, evidence_window_end, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + ( + observation_claim_id, + draft.observation_type, + draft.name, + component.scope, + component.tenant_id, + component.support_hash, + ALGORITHM_VERSION, + next(iter(ontology_versions)), + component.evidence_window_start, + component.evidence_window_end, + stamp, + stamp, + ), + ) + conn.commit() + self.persist_supports(observation_claim_id, component) + + def observation_gate(self, observation_claim_id: int) -> tuple[bool, str]: + with self._connection() as conn: + row = conn.execute( + """SELECT scope, tenant_id, support_hash FROM graph_observations + WHERE observation_claim_id=?""", + (observation_claim_id,), + ).fetchone() + if row is None: + return False, "observation_metadata_missing" + active = self.load_active_supports(scope=row["scope"], tenant_id=row["tenant_id"]) + discovery = discover_components( + active, scope=str(row["scope"]), tenant_id=row["tenant_id"] + ) + component = next( + (item for item in discovery.components if item.support_hash == row["support_hash"]), + None, + ) + if component is None: + return False, "support_fingerprint_changed" + weakest = min(item.confidence for item in component.supports) + if len(component.source_item_ids) < 2: + return False, "independent_sources_below_minimum" + if weakest < 0.65: + return False, "support_confidence_below_minimum" + return True, "eligible" + + def scope_observations( + self, *, scope: str, tenant_id: str | None + ) -> list[dict[str, Any]]: + with self._connection() as conn: + rows = conn.execute( + """SELECT go.*, c.status, c.text, c.confidence, c.created_at AS claim_created_at + FROM graph_observations go + JOIN claims c ON c.id=go.observation_claim_id + WHERE go.scope=? AND go.tenant_id IS ? + ORDER BY go.observation_claim_id""", + (scope, tenant_id), + ).fetchall() + return [_mapping(row) for row in rows] + + def observation_support_rows(self, observation_claim_id: int) -> list[dict[str, Any]]: + with self._connection() as conn: + rows = conn.execute( + """SELECT supporting_claim_id, evidence_item_id, source_item_id, + source_entity_id, relation, target_entity_id, ontology_version + FROM graph_observation_supports + WHERE observation_claim_id=? + ORDER BY supporting_claim_id, evidence_item_id, + source_entity_id, relation, target_entity_id""", + (observation_claim_id,), + ).fetchall() + return [_mapping(row) for row in rows] + + def observation_for_support( + self, *, scope: str, tenant_id: str | None, support_hash: str + ) -> int | None: + with self._connection() as conn: + row = conn.execute( + """SELECT observation_claim_id FROM graph_observations + WHERE scope=? AND tenant_id IS ? AND support_hash=? + AND algorithm_version=?""", + (scope, tenant_id, support_hash, ALGORITHM_VERSION), + ).fetchone() + return int(row["observation_claim_id"]) if row else None + + def status_counts( + self, *, tenant_id: str | None = None, scope: str | None = None + ) -> dict[str, dict[str, int]]: + counts = {stage: {status: 0 for status in JOB_STATUSES} for stage in JOB_STAGES} + params: list[Any] = [tenant_id] + scope_sql = "" + if scope: + scope_sql = " AND scope=?" + params.append(scope) + with self._connection() as conn: + rows = conn.execute( + f"""SELECT stage, status, COUNT(*) AS count + FROM graph_observation_jobs + WHERE tenant_id IS ?{scope_sql} + GROUP BY stage, status""", + tuple(params), + ).fetchall() + for row in rows: + counts[str(row["stage"])][str(row["status"])] = int(row["count"]) + return counts diff --git a/memorymaster/knowledge/graph_observations.py b/memorymaster/knowledge/graph_observations.py new file mode 100644 index 00000000..d56c9134 --- /dev/null +++ b/memorymaster/knowledge/graph_observations.py @@ -0,0 +1,323 @@ +"""Deterministic discovery and fail-closed validation for graph observations.""" + +from __future__ import annotations + +import hashlib +import json +from collections import defaultdict +from dataclasses import dataclass +from typing import Any, Iterable, Mapping + +from memorymaster.core.security import scan_text_for_findings + + +ALGORITHM_VERSION = "graph-observations-union-find-v1" +OBSERVATION_TYPES = frozenset( + { + "decision", + "commitment", + "constraint", + "dependency", + "state_change", + "recurring_pattern", + "stable_relationship", + "root_cause", + } +) +MAX_CLAIMS = 20 +MAX_EVIDENCE = 20 +MAX_EDGES = 40 +MAX_HUB_EPISODES = 20 + +Signature = tuple[int, str, int, str] + + +@dataclass(frozen=True, slots=True) +class ObservationSupport: + claim_id: int + evidence_id: int + source_item_id: int + source_entity_id: int + relation: str + target_entity_id: int + ontology_version: str + scope: str + tenant_id: str | None + confidence: float + occurred_at: str | None = None + + @property + def signature(self) -> Signature: + return ( + self.source_entity_id, + self.relation, + self.target_entity_id, + self.ontology_version, + ) + + +@dataclass(frozen=True, slots=True) +class ObservationComponent: + scope: str + tenant_id: str | None + supports: tuple[ObservationSupport, ...] + claim_ids: tuple[int, ...] + evidence_ids: tuple[int, ...] + source_item_ids: tuple[int, ...] + signatures: tuple[Signature, ...] + support_hash: str + evidence_window_start: str | None + evidence_window_end: str | None + + +@dataclass(frozen=True, slots=True) +class DiscoveryDiagnostic: + code: str + evidence_ids: tuple[int, ...] = () + + +@dataclass(frozen=True, slots=True) +class DiscoveryResult: + components: tuple[ObservationComponent, ...] + diagnostics: tuple[DiscoveryDiagnostic, ...] + + +@dataclass(frozen=True, slots=True) +class ObservationAssertion: + text: str + supporting_claim_ids: tuple[int, ...] + + +@dataclass(frozen=True, slots=True) +class ObservationDraft: + decision: str + name: str = "" + observation_type: str = "" + summary: str = "" + assertions: tuple[ObservationAssertion, ...] = () + + +class ObservationOutputError(ValueError): + """Raised when provider output violates the evidence-bound schema.""" + + +class _UnionFind: + def __init__(self, values: Iterable[int]) -> None: + self.parent = {value: value for value in values} + + def find(self, value: int) -> int: + root = value + while self.parent[root] != root: + root = self.parent[root] + while self.parent[value] != value: + value, self.parent[value] = self.parent[value], root + return root + + def union(self, left: int, right: int) -> None: + left_root, right_root = self.find(left), self.find(right) + if left_root == right_root: + return + low, high = sorted((left_root, right_root)) + self.parent[high] = low + + +def canonical_signature( + source_entity_id: int, + relation: str, + target_entity_id: int, + ontology_version: str, + *, + symmetric_relations: frozenset[str] = frozenset(), +) -> Signature: + """Return an exact signature, sorting endpoints only for symmetric relations.""" + source, target = int(source_entity_id), int(target_entity_id) + normalized_relation = str(relation).strip().lower() + version = str(ontology_version).strip() + if not normalized_relation or not version or source <= 0 or target <= 0: + raise ValueError("malformed graph signature") + if normalized_relation in symmetric_relations and source > target: + source, target = target, source + return source, normalized_relation, target, version + + +def support_fingerprint( + supports: Iterable[ObservationSupport], + *, + algorithm_version: str = ALGORITHM_VERSION, +) -> str: + manifest = [ + [ + row.claim_id, + row.evidence_id, + row.source_item_id, + *row.signature, + ] + for row in supports + ] + payload = { + "algorithm_version": algorithm_version, + "ontology_versions": sorted({row.ontology_version for row in supports}), + "supports": sorted(manifest), + } + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + return hashlib.sha256(encoded).hexdigest() + + +def _episode_signatures( + supports: tuple[ObservationSupport, ...], +) -> dict[int, set[Signature]]: + episodes: dict[int, set[Signature]] = defaultdict(set) + for row in supports: + episodes[row.evidence_id].add(row.signature) + return episodes + + +def _component_groups( + episodes: Mapping[int, set[Signature]], +) -> tuple[dict[int, list[int]], set[Signature]]: + signature_episodes: dict[Signature, list[int]] = defaultdict(list) + for evidence_id, signatures in episodes.items(): + for signature in signatures: + signature_episodes[signature].append(evidence_id) + hubs = {sig for sig, ids in signature_episodes.items() if len(ids) > MAX_HUB_EPISODES} + union = _UnionFind(episodes) + for signature in sorted(signature_episodes): + ids = sorted(signature_episodes[signature]) + if signature in hubs or not ids: + continue + for evidence_id in ids[1:]: + union.union(ids[0], evidence_id) + groups: dict[int, list[int]] = defaultdict(list) + for evidence_id in sorted(episodes): + groups[union.find(evidence_id)].append(evidence_id) + return groups, hubs + + +def _build_component( + supports: tuple[ObservationSupport, ...], + evidence_ids: set[int], + hubs: set[Signature], +) -> tuple[ObservationComponent | None, str | None]: + rows = tuple( + sorted( + (row for row in supports if row.evidence_id in evidence_ids and row.signature not in hubs), + key=lambda row: ( + row.evidence_id, + row.claim_id, + row.source_item_id, + row.signature, + ), + ) + ) + claims = tuple(sorted({row.claim_id for row in rows})) + evidence = tuple(sorted({row.evidence_id for row in rows})) + sources = tuple(sorted({row.source_item_id for row in rows})) + signatures = tuple(sorted({row.signature for row in rows})) + edges = {(row.claim_id, row.signature) for row in rows} + if len(claims) < 3 or len(evidence) < 2 or len(signatures) < 2: + return None, "below_eligibility_threshold" + if len(claims) > MAX_CLAIMS or len(evidence) > MAX_EVIDENCE or len(edges) > MAX_EDGES: + return None, "component_oversized" + dates = sorted(row.occurred_at for row in rows if row.occurred_at) + component = ObservationComponent( + scope=rows[0].scope, + tenant_id=rows[0].tenant_id, + supports=rows, + claim_ids=claims, + evidence_ids=evidence, + source_item_ids=sources, + signatures=signatures, + support_hash=support_fingerprint(rows), + evidence_window_start=dates[0] if dates else None, + evidence_window_end=dates[-1] if dates else None, + ) + return component, None + + +def discover_components( + supports: Iterable[ObservationSupport], + *, + scope: str, + tenant_id: str | None, +) -> DiscoveryResult: + """Build deterministic per-scope components; no model influences membership.""" + rows = tuple( + row for row in supports if row.scope == scope and row.tenant_id == tenant_id + ) + if not rows: + return DiscoveryResult((), ()) + episodes = _episode_signatures(rows) + groups, hubs = _component_groups(episodes) + diagnostics = [ + DiscoveryDiagnostic("hub_signature_suppressed", tuple(sorted(episodes))) + for _signature in sorted(hubs) + ] + components: list[ObservationComponent] = [] + for evidence_group in groups.values(): + component, code = _build_component(rows, set(evidence_group), hubs) + if component is not None: + components.append(component) + elif code: + diagnostics.append(DiscoveryDiagnostic(code, tuple(sorted(evidence_group)))) + components.sort(key=lambda item: item.support_hash) + return DiscoveryResult(tuple(components), tuple(diagnostics)) + + +def _strict_object(value: Any, *, keys: frozenset[str], label: str) -> dict[str, Any]: + if not isinstance(value, dict) or frozenset(value) != keys: + raise ObservationOutputError(f"{label} has an invalid schema") + return value + + +def _assertion(value: Any, allowed_claim_ids: frozenset[int]) -> ObservationAssertion: + row = _strict_object( + value, + keys=frozenset({"text", "supporting_claim_ids"}), + label="assertion", + ) + text = str(row["text"]).strip() + ids = row["supporting_claim_ids"] + if not text or len(text) > 500 or not isinstance(ids, list) or not ids: + raise ObservationOutputError("assertion is empty or oversized") + if any( + isinstance(item, bool) + or not isinstance(item, int) + or item not in allowed_claim_ids + for item in ids + ): + raise ObservationOutputError("assertion cites an unknown supporting claim") + if scan_text_for_findings(text): + raise ObservationOutputError("assertion contains sensitive material") + return ObservationAssertion(text, tuple(sorted(set(ids)))) + + +def parse_synthesis_output(raw: str, *, allowed_claim_ids: Iterable[int]) -> ObservationDraft: + """Validate provider JSON and reject any output not exactly supported.""" + if not isinstance(raw, str) or not raw.strip() or len(raw.encode("utf-8")) > 16_000: + raise ObservationOutputError("provider output is empty or oversized") + try: + payload = json.loads(raw) + except json.JSONDecodeError as exc: + raise ObservationOutputError("provider output is not valid JSON") from exc + if not isinstance(payload, dict) or payload.get("decision") not in {"emit", "no_signal"}: + raise ObservationOutputError("provider decision is invalid") + if payload["decision"] == "no_signal": + _strict_object(payload, keys=frozenset({"decision"}), label="no_signal") + return ObservationDraft(decision="no_signal") + row = _strict_object( + payload, + keys=frozenset({"decision", "name", "observation_type", "summary", "assertions"}), + label="observation", + ) + name, summary = str(row["name"]).strip(), str(row["summary"]).strip() + observation_type = str(row["observation_type"]).strip() + assertions = row["assertions"] + if not name or len(name) > 120 or not summary or len(summary) > 1200: + raise ObservationOutputError("observation text is empty or oversized") + if observation_type not in OBSERVATION_TYPES or not isinstance(assertions, list) or not assertions: + raise ObservationOutputError("observation type or assertions are invalid") + if scan_text_for_findings(f"{name}\n{summary}"): + raise ObservationOutputError("observation contains sensitive material") + allowed = frozenset(int(item) for item in allowed_claim_ids) + parsed = tuple(_assertion(item, allowed) for item in assertions) + return ObservationDraft("emit", name, observation_type, summary, parsed) diff --git a/memorymaster/public/demo.py b/memorymaster/public/demo.py index 4ca91f88..9e4cc45c 100644 --- a/memorymaster/public/demo.py +++ b/memorymaster/public/demo.py @@ -20,6 +20,10 @@ from memorymaster.core.models import CitationInput from memorymaster.core.service import MemoryService from memorymaster.knowledge.entity_graph import EntityGraph +from memorymaster.knowledge.graph_observation_engine import ( + GraphObservationEngine, + review_observation_candidates, +) from memorymaster.public.v1 import recall, remember @@ -39,35 +43,30 @@ def _capture_environment(workspace: Path): os.environ[name] = value -def _fixture_claim( - service: MemoryService, repository: CaptureRepository, evidence -): +def _fixture_claim(service: MemoryService, repository: CaptureRepository, evidence): claim = service.ingest( str(evidence.text or ""), [CitationInput(source="demo", locator=f"evidence:{evidence.id}")], scope="project:demo", + confidence=0.8, source_agent="memorymaster-demo", ) - repository.link_claim_evidence( - claim_id=claim.id, evidence_item_id=int(evidence.id) - ) + repository.link_claim_evidence(claim_id=claim.id, evidence_item_id=int(evidence.id)) return claim -def _run_fixture_worker( - service: MemoryService, repository: CaptureRepository, captures: list -) -> tuple[list, int]: +def _run_fixture_worker(service: MemoryService, repository: CaptureRepository, captures: list) -> tuple[list, int]: evidence = [ - service.list_evidence_items( - source_item_id=int(receipt.source_item["id"]), limit=1 - )[0] + service.list_evidence_items(source_item_id=int(receipt.source_item["id"]), limit=1)[0] for receipt in captures if receipt.evidence is not None ] claims = [_fixture_claim(service, repository, item) for item in evidence] - jobs = repository.lease_jobs( - owner="memorymaster-demo", stages=("extract_claims",), limit=100 - ) + with service.store.connect() as conn: + conn.execute("UPDATE source_items SET sensitivity='none'") + conn.execute("UPDATE evidence_items SET sensitivity='none'") + conn.commit() + jobs = repository.lease_jobs(owner="memorymaster-demo", stages=("extract_claims",), limit=100) for job in jobs: repository.finish_job(job.id, status="completed") return claims, len(jobs) @@ -108,18 +107,90 @@ def _extract_fixture_graph(db: Path, claim) -> list[dict]: conn.close() +def _add_dependency_graph(service: MemoryService, repository: CaptureRepository, claims: list) -> None: + with service.store.connect() as conn: + for entity_id, name in ((10, "API"), (20, "Database"), (30, "Migration")): + conn.execute( + """INSERT INTO entities + (id, canonical_name, entity_type, scope, created_at, updated_at) + VALUES (?, ?, 'system', 'project:demo', ?, ?)""", + (entity_id, name, "2026-08-12", "2026-08-12"), + ) + conn.commit() + for claim, edge in zip( + (claims[0], claims[1], claims[1], claims[2]), + ((10, 20), (10, 20), (20, 30), (20, 30)), + ): + repository.add_edge_support( + source_entity_id=edge[0], + target_entity_id=edge[1], + relation="depends_on", + supporting_claim_id=claim.id, + scope="project:demo", + ontology_version="personal-v1", + ) + + +def _run_observation_fixture(service: MemoryService, repository: CaptureRepository, claims: list) -> tuple[int, dict]: + _add_dependency_graph(service, repository, claims) + claim_ids = [claim.id for claim in claims] + output = json.dumps( + { + "decision": "emit", + "name": "Three-blocker dependency chain", + "observation_type": "dependency", + "summary": "Three blockers share the same API, database, and migration chain.", + "assertions": [ + { + "text": "All three blockers are supported by the supplied chain.", + "supporting_claim_ids": claim_ids, + } + ], + } + ) + engine = GraphObservationEngine(service.store, llm_call=lambda _system, _prompt: output) + engine.repo.queue_discovery( + tenant_id=None, + scope="project:demo", + ontology_version="personal-v1", + cycle_hour="2026-08-12T23", + ) + engine.process_discovery(owner="memorymaster-demo", scope="project:demo") + engine.process_synthesis(owner="memorymaster-demo", scope="project:demo") + review_observation_candidates(service.store, scope="project:demo") + row = engine.repo.scope_observations(scope="project:demo", tenant_id=None)[0] + observation_id = int(row["observation_claim_id"]) + recalled = recall( + "blocker dependency chain", + scope_allowlist=["project:demo"], + include_observations=True, + observation_limit=2, + retrieval_mode="legacy", + db=service.store.db_path, + workspace=service.workspace_root, + ) + return observation_id, { + "recall": recalled, + "support": engine.repo.observation_support_rows(observation_id), + } + + def _demo_report(workspace: Path, db: Path, captures: list) -> dict: service = MemoryService(db, workspace_root=workspace) service.init_db() repository = CaptureRepository(service.store) claims, completed = _run_fixture_worker(service, repository, captures) - confirmed = service.store.apply_status_transition( - claims[0], - to_status="confirmed", - reason="deterministic demo fixture", - event_type="validator", - ) - paths = _extract_fixture_graph(db, confirmed) + confirmed = [ + service.store.apply_status_transition( + claim, + to_status="confirmed", + reason="deterministic demo fixture", + event_type="validator", + ) + for claim in claims + ] + paths = _extract_fixture_graph(db, confirmed[0]) + observation_id, observation = _run_observation_fixture(service, repository, confirmed) recalled = recall( "Alice Project Atlas", scope_allowlist=["project:demo"], @@ -132,13 +203,16 @@ def _demo_report(workspace: Path, db: Path, captures: list) -> dict: "captures": len(captures), "fixture_jobs_completed": completed, "candidate_claims_created": len(claims), - "promoted_claim_id": confirmed.id, + "promoted_claim_id": confirmed[0].id, "recall_claim_ids": [int(claim["claim_id"]) for claim in recalled.claims], - "recall_citations": [ - citation for claim in recalled.claims for citation in claim["citations"] - ], + "recall_citations": [citation for claim in recalled.claims for citation in claim["citations"]], "graph_paths": paths, + "observation_claim_id": observation_id, + "observation_recall": list(observation["recall"].observations), + "observation_supports": observation["support"], } + repository.retire_source(int(captures[0].source_item["id"]), reason="demo support retirement") + report["observation_status_after_retirement"] = service.store.get_claim(observation_id).status del service, repository gc.collect() return report @@ -150,13 +224,11 @@ def run_disposable_demo() -> dict: workspace = Path(raw) db = workspace / "demo.db" document = workspace / "project-note.md" - document.write_text( - "Project Atlas uses governed evidence lineage.", encoding="utf-8" - ) + document.write_text("Project Atlas uses governed evidence lineage.", encoding="utf-8") with _capture_environment(workspace): captures = [ remember( - text="Alice participates in Project Atlas.", + text="Blocker one: Alice needs the API before Project Atlas can ship.", scope="project:demo", db=db, workspace=workspace, @@ -167,5 +239,11 @@ def run_disposable_demo() -> dict: db=db, workspace=workspace, ), + remember( + text="Blocker three: the migration depends on the database rollout.", + scope="project:demo", + db=db, + workspace=workspace, + ), ] return _demo_report(workspace, db, captures) diff --git a/memorymaster/public/v1.py b/memorymaster/public/v1.py index aa782498..962cc3b8 100644 --- a/memorymaster/public/v1.py +++ b/memorymaster/public/v1.py @@ -4,6 +4,7 @@ import os from dataclasses import asdict, dataclass +from datetime import datetime, timezone from pathlib import Path from typing import Any @@ -14,6 +15,10 @@ from memorymaster.core.session_scope import ResolvedScope, SessionScopeResolver from memorymaster.core.service import MemoryService from memorymaster.knowledge.context_bundle import query_context_bundle +from memorymaster.knowledge.graph_observation_repository import ( + GraphObservationRepository, +) +from memorymaster.knowledge.ontology import load_ontology API_VERSION = "memorymaster.public.v1" @@ -43,6 +48,7 @@ class RecallReceipt: trust_mode: str output_format: str skills: tuple[dict[str, Any], ...] = () + observations: tuple[dict[str, Any], ...] = () scope: str = "user" scope_source: str = "default_user" @@ -91,13 +97,19 @@ def _scope(scope: str | None, workspace: Path | None) -> str: return derived if derived != "global" else "user" -def _service(db: str | Path | None, workspace: Path | None) -> MemoryService: +def _service( + db: str | Path | None, + workspace: Path | None, + *, + tenant_id: str | None = None, +) -> MemoryService: target = str( db or os.environ.get("MEMORYMASTER_DB", "").strip() or os.environ.get("MEMORYMASTER_DEFAULT_DB", "").strip() or "memorymaster.db" ) + del tenant_id # Observation filtering is explicit; preserve facade compatibility. service = MemoryService(target, workspace_root=workspace or Path.cwd()) service.init_db() return service @@ -382,6 +394,9 @@ def recall( retrieval_mode: str = "hybrid", include_skills: bool = False, skill_limit: int = 3, + include_observations: bool = False, + observation_limit: int = 2, + tenant_id: str | None = None, session_id: str | None = None, source_agent: str = "memorymaster-public", platform: str = "local", @@ -390,7 +405,7 @@ def recall( ) -> RecallReceipt: """Return governed context and structured lifecycle/citation details.""" workspace_path = _workspace_path(workspace) - service = _service(db, workspace_path) + service = _service(db, workspace_path, tenant_id=tenant_id) if scope_allowlist: scopes = list(scope_allowlist) receipt_scope = scopes[0] if len(scopes) == 1 else "multiple" @@ -417,6 +432,9 @@ def recall( trust_mode=trust_mode, include_skills=include_skills, skill_limit=skill_limit, + include_observations=include_observations, + observation_limit=observation_limit, + observation_tenant_id=tenant_id, ) claims = tuple(_recall_claim(row) for row in result.rows) return RecallReceipt( @@ -428,6 +446,7 @@ def recall( trust_mode=trust_mode, output_format=result.output_format, skills=result.skills, + observations=result.observations, scope=receipt_scope, scope_source=scope_source, ) @@ -570,6 +589,19 @@ def _queue_due_graph( return queued, existing +def _queue_observation_discovery( + service: MemoryService, *, scope: str, tenant_id: str | None +) -> tuple[int, int]: + repository = GraphObservationRepository(service.store) + _job, created = repository.queue_discovery( + tenant_id=tenant_id, + scope=scope, + ontology_version=load_ontology().version, + cycle_hour=datetime.now(timezone.utc).strftime("%Y-%m-%dT%H"), + ) + return int(created), int(not created) + + def improve( *, scope: str | None = None, @@ -579,12 +611,13 @@ def improve( platform: str = "local", db: str | Path | None = None, workspace: str | Path | None = None, + tenant_id: str | None = None, ) -> ImproveReceipt: """Queue due work without directly confirming or rewriting claims.""" if not 1 <= max_items <= 200: raise ValueError("max_items must be between 1 and 200.") workspace_path = _workspace_path(workspace) - service = _service(db, workspace_path) + service = _service(db, workspace_path, tenant_id=tenant_id) resolved = _resolve_scope( service, scope=scope, @@ -601,6 +634,9 @@ def improve( graph_queued, graph_existing = _queue_due_graph( repository, scope=resolved_scope, limit=max_items ) + observation_queued, observation_existing = _queue_observation_discovery( + service, scope=resolved_scope, tenant_id=tenant_id + ) candidates = service.store.list_claims( status="candidate", limit=max_items, @@ -610,8 +646,18 @@ def improve( return ImproveReceipt( api_version=API_VERSION, scope=resolved_scope, - queued={"extract_claims": claim_queued, "extract_graph": graph_queued}, - already_pending={"extract_claims": claim_existing, "extract_graph": graph_existing}, + queued={ + "extract_claims": claim_queued, + "extract_graph": graph_queued, + "observation_discover": observation_queued, + "observation_synthesize": 0, + }, + already_pending={ + "extract_claims": claim_existing, + "extract_graph": graph_existing, + "observation_discover": observation_existing, + "observation_synthesize": 0, + }, steward_review_due=len(candidates), scope_source=resolved.scope_source, ) diff --git a/memorymaster/stores/migrations/0020_graph_observations.py b/memorymaster/stores/migrations/0020_graph_observations.py new file mode 100644 index 00000000..0c11c1de --- /dev/null +++ b/memorymaster/stores/migrations/0020_graph_observations.py @@ -0,0 +1,102 @@ +"""Add SQLite persistence for governed graph observations.""" + +from __future__ import annotations + +from typing import Any + + +VERSION = 20 +DESCRIPTION = "Add governed graph observations and leased jobs" + + +_SQLITE_SCHEMA = """ +CREATE TABLE IF NOT EXISTS graph_observations ( + observation_claim_id INTEGER PRIMARY KEY REFERENCES claims(id) ON DELETE CASCADE, + observation_type TEXT NOT NULL CHECK (observation_type IN ( + 'decision','commitment','constraint','dependency','state_change', + 'recurring_pattern','stable_relationship','root_cause' + )), + name TEXT NOT NULL CHECK (length(name) BETWEEN 1 AND 120), + scope TEXT NOT NULL, + tenant_id TEXT, + support_hash TEXT NOT NULL CHECK (length(support_hash) = 64), + algorithm_version TEXT NOT NULL, + ontology_version TEXT NOT NULL, + evidence_window_start TEXT, + evidence_window_end TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_observations_replay + ON graph_observations( + COALESCE(tenant_id, ''), scope, support_hash, + algorithm_version, ontology_version + ); +CREATE INDEX IF NOT EXISTS idx_graph_observations_scope + ON graph_observations(COALESCE(tenant_id, ''), scope, observation_type); + +CREATE TABLE IF NOT EXISTS graph_observation_supports ( + observation_claim_id INTEGER NOT NULL REFERENCES claims(id) ON DELETE CASCADE, + supporting_claim_id INTEGER NOT NULL REFERENCES claims(id) ON DELETE RESTRICT, + evidence_item_id INTEGER NOT NULL REFERENCES evidence_items(id) ON DELETE RESTRICT, + source_item_id INTEGER NOT NULL REFERENCES source_items(id) ON DELETE RESTRICT, + source_entity_id INTEGER NOT NULL REFERENCES entities(id) ON DELETE RESTRICT, + target_entity_id INTEGER NOT NULL REFERENCES entities(id) ON DELETE RESTRICT, + relation TEXT NOT NULL, + ontology_version TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY ( + observation_claim_id, supporting_claim_id, evidence_item_id, + source_entity_id, target_entity_id, relation, ontology_version + ) +); +CREATE INDEX IF NOT EXISTS idx_graph_observation_support_claim + ON graph_observation_supports(supporting_claim_id, observation_claim_id); +CREATE INDEX IF NOT EXISTS idx_graph_observation_support_evidence + ON graph_observation_supports(evidence_item_id, observation_claim_id); +CREATE INDEX IF NOT EXISTS idx_graph_observation_support_source + ON graph_observation_supports(source_item_id, observation_claim_id); + +CREATE TABLE IF NOT EXISTS graph_observation_jobs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + tenant_id TEXT, + scope TEXT NOT NULL, + stage TEXT NOT NULL CHECK (stage IN ('discover','synthesize')), + status TEXT NOT NULL CHECK (status IN ( + 'pending','leased','retryable','blocked','completed','cancelled' + )), + content_hash TEXT NOT NULL CHECK (length(content_hash) = 64), + support_hash TEXT CHECK (support_hash IS NULL OR length(support_hash) = 64), + algorithm_version TEXT NOT NULL, + ontology_version TEXT NOT NULL, + support_manifest_json TEXT NOT NULL DEFAULT '[]', + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts BETWEEN 0 AND 5), + next_attempt_at TEXT, + lease_owner TEXT, + lease_expires_at TEXT, + error_code TEXT, + diagnostic_hash TEXT CHECK (diagnostic_hash IS NULL OR length(diagnostic_hash) = 64), + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + completed_at TEXT +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_graph_observation_jobs_replay + ON graph_observation_jobs( + COALESCE(tenant_id, ''), scope, stage, content_hash, + algorithm_version, ontology_version + ); +CREATE INDEX IF NOT EXISTS idx_graph_observation_jobs_due + ON graph_observation_jobs(status, next_attempt_at, id); +CREATE INDEX IF NOT EXISTS idx_graph_observation_jobs_lease + ON graph_observation_jobs(status, lease_expires_at); +""" + + +def apply_sqlite(conn: Any) -> None: + conn.executescript(_SQLITE_SCHEMA) + conn.commit() + + +def apply_postgres(conn: Any) -> None: + """Fail closed because PPR-7 PostgreSQL rollout is explicitly deferred.""" + raise RuntimeError("migration 20 is SQLite-only; PostgreSQL rollout is deferred") diff --git a/memorymaster/surfaces/cli.py b/memorymaster/surfaces/cli.py index 3a2e59e6..10f9dca2 100644 --- a/memorymaster/surfaces/cli.py +++ b/memorymaster/surfaces/cli.py @@ -680,6 +680,17 @@ def build_parser() -> argparse.ArgumentParser: recall_cmd.add_argument("--session-id", default=None, help="Optional producer session ID") recall_cmd.add_argument("--source-agent", default="memorymaster-cli") recall_cmd.add_argument("--platform", default="cli") + recall_cmd.add_argument( + "--include-observations", + action="store_true", + help="Include separately labeled, evidence-bound graph observations", + ) + recall_cmd.add_argument( + "--observation-limit", + type=int, + default=2, + help="Maximum observations (default 2, bounded to 5)", + ) forget_cmd = sub.add_parser("forget", help="Preview or apply logical retirement") forget_target = forget_cmd.add_mutually_exclusive_group(required=True) diff --git a/memorymaster/surfaces/cli_handlers_public.py b/memorymaster/surfaces/cli_handlers_public.py index 2b68e5c1..0b93a6f4 100644 --- a/memorymaster/surfaces/cli_handlers_public.py +++ b/memorymaster/surfaces/cli_handlers_public.py @@ -55,6 +55,9 @@ def handle_recall( session_id=args.session_id, source_agent=args.source_agent, platform=args.platform, + include_observations=args.include_observations, + observation_limit=args.observation_limit, + tenant_id=getattr(service, "tenant_id", None), db=effective_db, workspace=args.workspace, ) @@ -94,6 +97,7 @@ def handle_improve( session_id=args.session_id, source_agent=args.source_agent, platform=args.platform, + tenant_id=getattr(service, "tenant_id", None), db=effective_db, workspace=args.workspace, ) @@ -103,6 +107,7 @@ def handle_improve( print( f"improve queued: claims={receipt.queued['extract_claims']} " f"graph={receipt.queued['extract_graph']} " + f"observations={receipt.queued['observation_discover']} " f"steward_review_due={receipt.steward_review_due}" ) return 0 diff --git a/memorymaster/surfaces/dashboard.py b/memorymaster/surfaces/dashboard.py index d8b71848..6ec03321 100644 --- a/memorymaster/surfaces/dashboard.py +++ b/memorymaster/surfaces/dashboard.py @@ -1,5 +1,4 @@ from __future__ import annotations - import argparse from datetime import datetime, timedelta, timezone from html import escape @@ -18,7 +17,7 @@ import urllib.request from urllib.parse import parse_qs, urlparse -from memorymaster.surfaces import capture_inbox as capture_inbox_surface, dashboard_auth, session_scope as session_scope_surface +from memorymaster.surfaces import capture_inbox as capture_inbox_surface, dashboard_auth, graph_observations_dashboard as graph_observations_surface, session_scope as session_scope_surface from memorymaster.core.config import get_config from memorymaster.govern.review import build_review_queue from memorymaster.core.service import MemoryService @@ -316,6 +315,7 @@ def _build_get_route_map(handler: Any) -> dict[str, callable]: "/favicon.ico": lambda qs: capture_inbox_surface.write_favicon_response(handler, qs), "/api/claims": lambda qs: handler._handle_claims(qs), "/api/capture-inbox": lambda qs: capture_inbox_surface.write_capture_inbox_response(handler, qs), + "/api/graph-observations": lambda qs: graph_observations_surface.write_graph_observations_response(handler, qs), "/api/events": lambda qs: handler._handle_events(qs), "/api/timeline": lambda qs: handler._handle_timeline(qs), "/api/conflicts": lambda qs: handler._handle_conflicts(qs), @@ -1006,7 +1006,7 @@ def _write_dashboard(self) -> None:
📈

System Health

Operator metrics, latency, event counters
Waiting for data...
-__CAPTURE_INBOX_SECTION__ +__CAPTURE_INBOX_SECTION____GRAPH_OBSERVATIONS_SECTION__
⏱

Validation Latency

Candidate claim creation to first validation event
Waiting for data...
@@ -1085,7 +1085,7 @@ def _write_dashboard(self) -> None: function renderConflicts(){const h=document.getElementById('conflicts-cards');const meta=document.getElementById('conflicts-meta');const q=String(conflictState.search||'').toLowerCase().trim();const rows=conflictState.rows.filter(g=>{if(!q)return true;const claims=Array.isArray(g.claims)?g.claims:[];const claimText=claims.map(c=>[c.id,c.status,c.subject,c.predicate,c.object_value,c.text].join(' ')).join(' ').toLowerCase();return ([g.subject,g.predicate,g.scope].join(' ').toLowerCase()+' '+claimText).includes(q);});meta.textContent=rows.length+' of '+conflictState.rows.length+' conflict groups';h.innerHTML=rows.map(g=>{const cs=Array.isArray(g.claims)?g.claims:[];const n=cs[0]||null;const o=cs.length>1?cs[1]:null;const nv=n?(n.object_value||n.text||'-'):'-';const ov=o?(o.object_value||o.text||'-'):'-';const nc=n?Number(n.confidence||0):null;const oc=o?Number(o.confidence||0):null;const nz=n?((n.citations||[]).length):0;const oz=o?((o.citations||[]).length):0;const confDelta=(nc!=null&&oc!=null)?(nc-oc):null;const citeDelta=(n&&o)?(nz-oz):null;const valueChanged=(String(nv)!==String(ov));const delta=(v)=>v==null?'-':((v>=0?'+':'')+Number(v).toFixed(3));const cDelta=(v)=>v==null?'-':((v>=0?'+':'')+String(v));const row=(label,a,b,chg)=>''+esc(label)+''+esc(a)+''+esc(b)+''+esc(chg)+'';const statusCounts={};cs.forEach(c=>{const k=String(c.status||'unknown');statusCounts[k]=(statusCounts[k]||0)+1;});const card=(label,c,value,confidence,cites,color)=>'
'+label+' '+(c?statusBadge(c.status):'')+' #'+esc(c?c.id:'-')+'
'+esc(value)+'
confidence: '+esc(confidence==null?'-':confidence.toFixed(3))+' · citations: '+esc(cites)+'
'+governanceEvidence(c)+'
';return '
'+esc(g.subject||'-')+' / '+esc(g.predicate||'-')+'
'+esc(cs.length)+' claims · scope: '+esc(g.scope||'project')+' · '+countPills(statusCounts,4)+'
'+card('Newer',n,nv,nc,nz,'#4ade80')+card('Older',o,ov,oc,oz,'#f87171')+'
'+row('value',nv,ov,valueChanged?'CHANGED':'same')+row('confidence',nc==null?'-':nc.toFixed(3),oc==null?'-':oc.toFixed(3),delta(confDelta))+row('citations',String(nz),String(oz),cDelta(citeDelta))+row('updated_at',n&&n.updated_at?n.updated_at:'-',o&&o.updated_at?o.updated_at:'-',(n&&o&&String(n.updated_at)!==String(o.updated_at))?'changed':'same')+'
FieldNewerOlderDelta
';}).join('')||'
No conflicts detected
';} function fillConflicts(d){conflictState.rows=Array.isArray(d.groups)?d.groups:[];renderConflicts();} async function refreshConflicts(){const includeStale=document.getElementById('conflicts-include-stale').checked?'1':'0';fillConflicts(await jget('/api/conflicts?limit=20&include_stale='+includeStale));} -__CAPTURE_INBOX_FUNCTIONS__ +__CAPTURE_INBOX_FUNCTIONS____GRAPH_OBSERVATIONS_FUNCTIONS__ function fillQueue(d){const rows=Array.isArray(d.items)?d.items:[];const b=document.getElementById('stale-body');if(!rows.length){b.innerHTML='Nothing to review';return;}b.innerHTML=rows.map(i=>{const p=i.proposal||null;const proposalId=p?Number(p.proposal_event_id||0):0;const proposalActions=proposalId>0?' ':'';return ''+esc(i.claim_id)+''+statusBadge(i.status)+''+esc(i.reason||'-')+proposalEvidence(p)+''+esc(f3(i.priority))+' '+proposalActions+actionConsequences(Boolean(p))+'';}).join('');} function fillRetr(d){const rows=Array.isArray(d.rows_data)?d.rows_data:[];const b=document.getElementById('retrieval-body');const meta=document.getElementById('retrieval-meta');const scopes=Array.isArray(d.scope_allowlist)?d.scope_allowlist:[];const scopeText=scopes.length?scopes.join(', '):'all';meta.textContent='Mode: '+(d.mode||'-')+' · Scopes: '+scopeText+' · '+rows.length+' results';b.innerHTML=rows.map(r=>{const c=r.claim||{};const s=r.status||c.status||'unknown';const ann=(r.annotation||'-');return ''+esc(c.id)+''+tuple(c)+''+statusBadge(s)+''+esc(ann)+''+esc(f3(r.score))+''+esc(f3(r.lexical_score))+' / '+esc(f3(r.confidence_score))+' / '+esc(f3(r.freshness_score))+' / '+esc(f3(r.vector_score))+'';}).join('')||'No results found';} function fillRecallAnalysis(d){const rows=Array.isArray(d.results)?d.results:[];const meta=document.getElementById('recall-meta');const wbox=document.getElementById('recall-weights');meta.textContent='Mode: '+(d.mode||'-')+(d.profile?(' · Profile: '+d.profile):'')+' · '+rows.length+' results';const w=(d.weights&&d.weights.retrieval_weights)||null;wbox.innerHTML=w?('Active weights — lexical: '+esc(f3(w.lexical))+' · confidence: '+esc(f3(w.confidence))+' · freshness: '+esc(f3(w.freshness))+' · vector: '+esc(f3(w.vector))):'';const b=document.getElementById('recall-body');b.innerHTML=rows.map(r=>{const id=r.human_id||r.claim_id;const pin=r.pinned?' 📌':'';return ''+esc(id)+pin+''+esc(r.text||'-')+''+statusBadge(r.status)+''+esc(r.tier||'working')+''+esc(f3(r.score))+''+esc(f3(r.lexical_score))+' / '+esc(f3(r.confidence_score))+' / '+esc(f3(r.freshness_score))+' / '+esc(f3(r.vector_score))+'';}).join('')||'No results found';} @@ -1100,7 +1100,7 @@ def _write_dashboard(self) -> None: async function refreshQueue(){fillQueue(await jget('/api/review-queue?limit=30&exclude_reviewed=1&exclude_suppressed=1'));} async function refreshObs(){fillObs(await jget('/api/observability?log_limit=1500&event_limit=600&queue_limit=250'));} document.getElementById('stale-body').addEventListener('click',async(ev)=>{const t=ev.target;if(!t||t.tagName!=='BUTTON'||!t.hasAttribute('data-action'))return;const r=t.closest('tr');if(!r)return;const id=Number(r.getAttribute('data-claim-id'));const a=String(t.getAttribute('data-action')||'');const proposalId=Number(t.getAttribute('data-proposal-event-id')||0);const body={claim_id:id,action:a};if(a==='approve_proposal'||a==='reject_proposal')body.proposal_event_id=proposalId;const original=t.innerHTML;r.setAttribute('data-pending','true');r.classList.add('pending');t.disabled=true;reviewStatus('Applying '+a+' to claim '+id+'...');try{await jpost('/api/triage/action',body);r.remove();reviewStatus('Action '+a+' completed for claim '+id+'.');await refreshQueue();}catch(error){r.removeAttribute('data-pending');r.classList.remove('pending');t.disabled=false;t.innerHTML='Retry';t.title=String((error&&error.message)||error||'Action failed');reviewStatus('Action failed for claim '+id+'. Retry is available.',true);t.addEventListener('blur',()=>{t.innerHTML=original;},{once:true});}}); -__CAPTURE_INBOX_EVENTS__ +__CAPTURE_INBOX_EVENTS____GRAPH_OBSERVATIONS_EVENTS__ document.getElementById('op-start').addEventListener('click',async()=>{await jpost('/api/operator/control',{action:'start',inbox_jsonl:document.getElementById('op-inbox').value});fillOp(await jget('/api/operator/status'));}); document.getElementById('op-stop').addEventListener('click',async()=>{await jpost('/api/operator/control',{action:'stop'});fillOp(await jget('/api/operator/status'));}); document.getElementById('retrieval-run').addEventListener('click',async()=>{const query=document.getElementById('retrieval-query').value||'';const mode=document.getElementById('retrieval-mode').value||'hybrid';const scopeRaw=document.getElementById('retrieval-scope').value||'';const url='/api/retrieval?query='+encodeURIComponent(query)+'&mode='+encodeURIComponent(mode)+'&scope_allowlist='+encodeURIComponent(scopeRaw)+'&limit=10';fillRetr(await jget(url));}); @@ -1114,7 +1114,7 @@ def _write_dashboard(self) -> None: jget('/api/claims?limit=50').then(fillClaims).catch(e=>showPanelFailure('claims-body','claims',e,6));refreshCapture().catch(e=>showPanelFailure('capture-inbox','capture inbox',e));jget('/api/timeline?limit=40').then(fillTimeline).catch(e=>showPanelFailure('timeline-list','timeline',e));refreshConflicts().catch(e=>showPanelFailure('conflicts-cards','conflicts',e));refreshQueue().catch(e=>showPanelFailure('stale-body','review queue',e,5));jget('/api/audit?limit=40').then(fillAudit).catch(e=>showPanelFailure('audit-body','audit log',e,4));jget('/api/namespaces?limit=200').then(fillNs).catch(e=>showPanelFailure('namespaces-box','namespaces',e));jget('/api/provenance').then(fillProvenance).catch(e=>showPanelFailure('provenance-body','provenance',e,8));jget('/api/session-stats?limit=2000').then(fillStats).catch(e=>showPanelFailure('session-stats','session statistics',e));jget('/api/session-bindings?limit=100').then(fillScopeBindings).catch(e=>showPanelFailure('scope-bindings-body','session scope bindings',e,6));jget('/metrics/validation-latency').then(fillValidationLatency).catch(e=>showPanelFailure('validation-latency','validation latency',e));jget('/api/integrity').then(fillIntegrity).catch(e=>showPanelFailure('integrity-box','integrity',e));jget('/api/operator/status').then(fillOp).catch(e=>showPanelFailure('op-status','operator status',e));refreshObs().catch(e=>showPanelFailure('obs-box','observability',e)); const sb=document.getElementById('stream');const es=new EventSource('/api/operator/stream?last=20'); const append=(t)=>{const ex=sb.textContent.trim();sb.textContent=(ex&&ex!=='Waiting for operator to start...'?ex+'\\n':'')+t;}; ['message','stream_start','state_loaded','state_error','state_saved','json_error','turn_processed','reconcile_run','stream_exit'].forEach(n=>es.addEventListener(n,(ev)=>append(ev.data))); es.onerror=()=>append('[stream reconnecting]'); """ - html = session_scope_surface.hydrate_dashboard_html(capture_inbox_surface.hydrate_dashboard_html(html, escape(_package_version()))) + html = session_scope_surface.hydrate_dashboard_html(graph_observations_surface.hydrate_dashboard_html(capture_inbox_surface.hydrate_dashboard_html(html, escape(_package_version())))) body = html.encode("utf-8") self.send_response(HTTPStatus.OK) self.send_header("Content-Type", "text/html; charset=utf-8") diff --git a/memorymaster/surfaces/graph_observations_dashboard.py b/memorymaster/surfaces/graph_observations_dashboard.py new file mode 100644 index 00000000..27658537 --- /dev/null +++ b/memorymaster/surfaces/graph_observations_dashboard.py @@ -0,0 +1,109 @@ +"""Read-only dashboard surface for governed graph observations.""" + +from __future__ import annotations + +from typing import Any +from urllib.parse import parse_qs + +from memorymaster.knowledge.graph_observation_repository import GraphObservationRepository + + +GRAPH_OBSERVATIONS_SECTION_HTML = """
+
🕸

Derived Observations

Opt-in graph synthesis, exact support, diagnostics, and lifecycle history
+
+
+
No graph observations yet
+
""" + +GRAPH_OBSERVATIONS_FUNCTIONS_JS = """ +function fillGraphObservations(d){const rows=Array.isArray(d.observations)?d.observations:[];const jobs=d.jobs||{};const diagnostics=Array.isArray(d.diagnostics)?d.diagnostics:[];const box=document.getElementById('graph-observations');document.getElementById('graph-observation-status').textContent='Observations '+rows.length+'; jobs '+JSON.stringify(jobs);const obs=rows.map(o=>{const supports=Array.isArray(o.supports)?o.supports:[];const claims=[...new Set(supports.map(s=>s.supporting_claim_id))];const evidence=[...new Set(supports.map(s=>s.evidence_item_id))];const rels=[...new Set(supports.map(s=>String(s.source_entity_id)+' '+String(s.relation)+' '+String(s.target_entity_id)))];const history=(o.lifecycle||[]).map(e=>'
'+esc(e.created_at||'-')+' '+esc(e.event_type||'-')+' '+esc(e.from_status||'')+' → '+esc(e.to_status||'')+'
').join('')||'
no lifecycle events
';return '
'+statusBadge(o.status)+' '+esc(o.name||o.text||'-')+' '+esc(o.observation_type||'-')+' #'+esc(o.observation_claim_id)+'
Evidence window: '+esc(o.evidence_window_start||'-')+' → '+esc(o.evidence_window_end||'-')+'
Supporting claims: '+esc(claims.join(', ')||'-')+'
Evidence: '+esc(evidence.join(', ')||'-')+'
Relationships:
'+esc(rels.join(' | ')||'-')+'
Lifecycle history'+history+'
';}).join('');const diag=diagnostics.length?'
Diagnostics
'+diagnostics.map(x=>'
'+esc(x.stage||'job')+' '+esc(x.error_code||'diagnostic')+' '+esc(x.scope||'-')+' '+esc(x.updated_at||'-')+'
').join('')+'
':'';box.innerHTML=obs+diag||'
No graph observations or diagnostics
';} +async function refreshGraphObservations(){const scope=document.getElementById('graph-observation-scope').value||'';fillGraphObservations(await jget('/api/graph-observations?limit=30&scope='+encodeURIComponent(scope)));} +""" + +GRAPH_OBSERVATIONS_EVENTS_JS = """document.getElementById('graph-observation-refresh').addEventListener('click',refreshGraphObservations);refreshGraphObservations().catch(e=>showPanelFailure('graph-observations','graph observations',e));""" + + +def _rows(cursor: Any) -> list[dict[str, Any]]: + return [{key: row[key] for key in row.keys()} for row in cursor.fetchall()] + + +def _observation_rows(repository: GraphObservationRepository, conn: Any, *, tenant_id: str | None, scope: str | None, limit: int) -> list[dict[str, Any]]: + scope_sql = " AND go.scope=?" if scope else "" + params = (tenant_id, scope, limit) if scope else (tenant_id, limit) + observations = _rows( + conn.execute( + f"""SELECT go.*, c.status, c.text, c.confidence + FROM graph_observations go JOIN claims c ON c.id=go.observation_claim_id + WHERE go.tenant_id IS ?{scope_sql} + ORDER BY go.observation_claim_id DESC LIMIT ?""", + params, + ) + ) + for row in observations: + claim_id = int(row["observation_claim_id"]) + row["supports"] = repository.observation_support_rows(claim_id) + row["lifecycle"] = _rows(conn.execute( + """SELECT event_type, from_status, to_status, details, created_at + FROM events WHERE claim_id=? ORDER BY id""", (claim_id,))) + return observations + + +def _diagnostic_rows(conn: Any, *, tenant_id: str | None, scope: str | None) -> list[dict[str, Any]]: + scope_sql = " AND scope=?" if scope else "" + params = (tenant_id, scope) if scope else (tenant_id,) + return _rows(conn.execute( + f"""SELECT id, scope, stage, status, attempts, error_code, + diagnostic_hash, updated_at FROM graph_observation_jobs + WHERE tenant_id IS ?{scope_sql} AND error_code IS NOT NULL + ORDER BY id DESC LIMIT 30""", params)) + + +def graph_observations_payload( + service: Any, + *, + scope: str | None = None, + tenant_id: str | None = None, + limit: int = 30, +) -> dict[str, Any]: + """Return bounded observation status, exact support, jobs, and history.""" + repository = GraphObservationRepository(service.store) + with repository._connection() as conn: + observations = _observation_rows( + repository, conn, tenant_id=tenant_id, scope=scope, limit=limit + ) + claim_ids = [int(row["observation_claim_id"]) for row in observations] + diagnostics = _diagnostic_rows(conn, tenant_id=tenant_id, scope=scope) + return { + "ok": True, + "observations": observations, + "observation_claim_ids": claim_ids, + "diagnostics": diagnostics, + "jobs": repository.status_counts(tenant_id=tenant_id, scope=scope), + } + + +def hydrate_dashboard_html(html: str) -> str: + """Insert the graph-observation panel and scripts into the dashboard.""" + return ( + html.replace("__GRAPH_OBSERVATIONS_SECTION__", GRAPH_OBSERVATIONS_SECTION_HTML) + .replace("__GRAPH_OBSERVATIONS_FUNCTIONS__", GRAPH_OBSERVATIONS_FUNCTIONS_JS) + .replace("__GRAPH_OBSERVATIONS_EVENTS__", GRAPH_OBSERVATIONS_EVENTS_JS) + ) + + +def write_graph_observations_response(handler: Any, query_string: str) -> None: + """Write the bounded, tenant-isolated graph-observation read model.""" + query = parse_qs(query_string) + raw_limit = str((query.get("limit") or ["30"])[-1]).strip() + limit = int(raw_limit or "30") + if limit < 1 or limit > 100: + raise ValueError("Expected integer in range [1, 100]") + scope = str((query.get("scope") or [""])[-1]).strip() or None + handler._write_json( + graph_observations_payload( + handler._server.service, + scope=scope, + tenant_id=getattr(handler._server.service, "tenant_id", None), + limit=limit, + ) + ) diff --git a/memorymaster/surfaces/mcp_server.py b/memorymaster/surfaces/mcp_server.py index bf71e813..9010aa77 100644 --- a/memorymaster/surfaces/mcp_server.py +++ b/memorymaster/surfaces/mcp_server.py @@ -917,6 +917,8 @@ def recall( retrieval_mode: str = "hybrid", include_skills: bool = False, skill_limit: int = 3, + include_observations: bool = False, + observation_limit: int = 2, session_id: str = "", source_agent: str = "", platform: str = "mcp", @@ -935,11 +937,14 @@ def recall( retrieval_mode=retrieval_mode, include_skills=include_skills, skill_limit=_bounded_limit(skill_limit, maximum=10), + include_observations=include_observations, + observation_limit=_bounded_limit(observation_limit, maximum=5), session_id=session_id or None, source_agent=source_agent or "memorymaster-mcp", platform=platform, db=db, workspace=workspace, + tenant_id=(current_request_context().tenant_id if current_request_context() else None), ) return {"ok": True, **asdict(receipt)} @@ -984,6 +989,7 @@ def improve( platform=platform, db=db, workspace=workspace, + tenant_id=(current_request_context().tenant_id if current_request_context() else None), ) return {"ok": True, **asdict(receipt)} diff --git a/tests/test_graph_observations.py b/tests/test_graph_observations.py new file mode 100644 index 00000000..b7d1411b --- /dev/null +++ b/tests/test_graph_observations.py @@ -0,0 +1,490 @@ +from __future__ import annotations + +import hashlib +import importlib +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from memorymaster.core.service import MemoryService +from memorymaster.core.lifecycle import transition_claim +from memorymaster.core.models import CitationInput +from memorymaster.dreaming.worker import DreamWorker +from memorymaster.evaluation.graph_observation_evaluator import evaluate_corpus +from memorymaster.capture.repository import CaptureRepository +from memorymaster.knowledge.graph_observation_engine import ( + GraphObservationEngine, + review_observation_candidates, +) +from memorymaster.knowledge.graph_observation_repository import ( + GraphObservationRepository, +) +from memorymaster.public.v1 import recall as public_recall +from memorymaster.knowledge.graph_observations import ( + ObservationOutputError, + ObservationSupport, + canonical_signature, + discover_components, + parse_synthesis_output, +) +from memorymaster.surfaces.graph_observations_dashboard import ( + graph_observations_payload, + hydrate_dashboard_html, +) + + +def _support( + claim_id: int, + evidence_id: int, + source_item_id: int, + source_entity_id: int, + relation: str, + target_entity_id: int, + *, + scope: str = "project:test", + tenant_id: str | None = "tenant-a", + confidence: float = 0.8, +) -> ObservationSupport: + return ObservationSupport( + claim_id=claim_id, + evidence_id=evidence_id, + source_item_id=source_item_id, + source_entity_id=source_entity_id, + relation=relation, + target_entity_id=target_entity_id, + ontology_version="personal-v1", + scope=scope, + tenant_id=tenant_id, + confidence=confidence, + occurred_at=f"2026-08-{evidence_id:02d}T00:00:00+00:00", + ) + + +def _eligible_supports() -> list[ObservationSupport]: + return [ + _support(1, 1, 101, 10, "depends_on", 20), + _support(2, 2, 102, 10, "depends_on", 20), + _support(2, 2, 102, 20, "depends_on", 30), + _support(3, 1, 101, 20, "depends_on", 30), + ] + + +def test_deterministic_component_and_fingerprint_replay() -> None: + rows = _eligible_supports() + first = discover_components(rows, scope="project:test", tenant_id="tenant-a") + second = discover_components(reversed(rows), scope="project:test", tenant_id="tenant-a") + + assert len(first.components) == 1 + assert first.components == second.components + assert first.components[0].claim_ids == (1, 2, 3) + assert first.components[0].evidence_ids == (1, 2) + assert len(first.components[0].signatures) == 2 + + +def test_symmetric_relation_canonicalization_is_exact() -> None: + symmetric = frozenset({"related_to"}) + left = canonical_signature(9, "related_to", 2, "personal-v1", symmetric_relations=symmetric) + right = canonical_signature(2, "related_to", 9, "personal-v1", symmetric_relations=symmetric) + directed = canonical_signature(9, "depends_on", 2, "personal-v1", symmetric_relations=symmetric) + + assert left == right == (2, "related_to", 9, "personal-v1") + assert directed == (9, "depends_on", 2, "personal-v1") + + +def test_scope_and_tenant_never_cross_components() -> None: + rows = _eligible_supports() + rows.extend( + _support( + row.claim_id + 10, + row.evidence_id + 10, + row.source_item_id + 10, + row.source_entity_id, + row.relation, + row.target_entity_id, + scope="project:other", + tenant_id="tenant-b", + ) + for row in _eligible_supports() + ) + + result = discover_components(rows, scope="project:test", tenant_id="tenant-a") + + assert len(result.components) == 1 + assert result.components[0].claim_ids == (1, 2, 3) + + +def test_hubs_shared_by_more_than_twenty_episodes_are_suppressed() -> None: + rows = [_support(index, index, 100 + index, 1, "related_to", 2) for index in range(1, 22)] + + result = discover_components(rows, scope="project:test", tenant_id="tenant-a") + + assert result.components == () + assert any(item.code == "hub_signature_suppressed" for item in result.diagnostics) + + +def test_oversized_component_produces_diagnostic_not_synthesis() -> None: + rows: list[ObservationSupport] = [] + for index in range(1, 22): + if index <= 20: + rows.append(_support(index, index, 100 + index, 1, "depends_on", 2)) + rows.append(_support(index, index, 100 + index, index + 2, "uses", index + 30)) + rows.extend( + [ + _support(20, 20, 120, 50, "depends_on", 51), + _support(21, 21, 121, 50, "depends_on", 51), + ] + ) + + result = discover_components(rows, scope="project:test", tenant_id="tenant-a") + + assert result.components == () + assert any(item.code == "component_oversized" for item in result.diagnostics) + + +def test_synthesis_output_is_evidence_bound_and_strict() -> None: + payload = json.dumps( + { + "decision": "emit", + "name": "Three blockers share one dependency chain", + "observation_type": "dependency", + "summary": "The three confirmed blockers depend on the same two systems.", + "assertions": [{"text": "The chain is supported by all three claims.", "supporting_claim_ids": [1, 2, 3]}], + } + ) + + draft = parse_synthesis_output(payload, allowed_claim_ids={1, 2, 3}) + + assert draft.decision == "emit" + assert draft.assertions[0].supporting_claim_ids == (1, 2, 3) + + +@pytest.mark.parametrize( + "payload", + [ + "not-json", + json.dumps({"decision": "emit"}), + json.dumps({"decision": "no_signal", "summary": "extra"}), + json.dumps( + { + "decision": "emit", + "name": "Unsupported", + "observation_type": "dependency", + "summary": "Unsupported claim.", + "assertions": [{"text": "Bad", "supporting_claim_ids": [99]}], + } + ), + json.dumps( + { + "decision": "emit", + "name": "Boolean ID", + "observation_type": "dependency", + "summary": "Boolean IDs are never claim IDs.", + "assertions": [{"text": "Bad", "supporting_claim_ids": [True]}], + } + ), + ], +) +def test_synthesis_output_fails_closed(payload: str) -> None: + with pytest.raises(ObservationOutputError): + parse_synthesis_output(payload, allowed_claim_ids={1, 2, 3}) + + +def test_migration_0020_is_sqlite_only_and_idempotent(tmp_path) -> None: + service = MemoryService(tmp_path / "observations.db", workspace_root=tmp_path) + service.init_db() + service.init_db() + + with service.store.connect() as conn: + tables = { + row[0] + for row in conn.execute( + """SELECT name FROM sqlite_master WHERE type='table' + AND name LIKE 'graph_observation%'""" + ) + } + versions = conn.execute("SELECT COUNT(*) FROM schema_versions WHERE version=20").fetchone()[0] + + migration = importlib.import_module("memorymaster.stores.migrations.0020_graph_observations") + with pytest.raises(RuntimeError, match="SQLite-only"): + migration.apply_postgres(object()) + assert tables == { + "graph_observations", + "graph_observation_supports", + "graph_observation_jobs", + } + assert versions == 1 + + +def test_job_replay_and_expired_lease_recovery(tmp_path) -> None: + service = MemoryService(tmp_path / "jobs.db", workspace_root=tmp_path) + service.init_db() + repo = GraphObservationRepository(service.store) + digest = hashlib.sha256(b"component").hexdigest() + first, created_first = repo.queue_job( + tenant_id=None, + scope="project:test", + stage="synthesize", + content_hash=digest, + support_hash=digest, + ontology_version="personal-v1", + support_manifest=[[1, 2, 3, 4, "depends_on", 5, "personal-v1"]], + ) + second, created_second = repo.queue_job( + tenant_id=None, + scope="project:test", + stage="synthesize", + content_hash=digest, + support_hash=digest, + ontology_version="personal-v1", + support_manifest=[[1, 2, 3, 4, "depends_on", 5, "personal-v1"]], + ) + leased = repo.lease_jobs(owner="worker-a", limit=1) + past = (datetime.now(timezone.utc) - timedelta(minutes=1)).isoformat() + with service.store.connect() as conn: + conn.execute( + "UPDATE graph_observation_jobs SET lease_expires_at=? WHERE id=?", + (past, first.id), + ) + conn.commit() + + recovered = repo.lease_jobs(owner="worker-b", limit=1) + + assert created_first is True and created_second is False + assert first.id == second.id == leased[0].id == recovered[0].id + assert recovered[0].attempts == 2 + + +def _graph_fixture(tmp_path): + service = MemoryService(tmp_path / "lifecycle.db", workspace_root=tmp_path) + service.init_db() + store = service.store + capture = CaptureRepository(store) + external = store.upsert_external_source(source_type="direct", display_name="fixture") + sources = [] + evidence = [] + for index in (1, 2): + source = store.upsert_source_item( + source_id=external.id, + source_item_id=f"source-{index}", + item_type="text", + occurred_at=f"2026-08-0{index}T00:00:00+00:00", + text=f"Dependency evidence {index}", + content_hash=hashlib.sha256(f"source-{index}".encode()).hexdigest(), + sensitivity="none", + ) + sources.append(source) + evidence.append( + store.add_evidence_item( + source_item_id=source.id, + evidence_type="text", + text=source.text, + sensitivity="none", + content_hash=source.content_hash, + ) + ) + claims = [] + for index, evidence_index in ((1, 0), (2, 1), (3, 0)): + claim = store.create_claim( + text=f"Confirmed blocker {index}", + citations=[CitationInput(source="fixture", locator=f"evidence:{evidence[evidence_index].id}")], + claim_type="fact", + scope="project:test", + confidence=0.8, + source_agent="fixture", + ) + transition_claim(store, claim.id, "confirmed", "fixture") + capture.link_claim_evidence(claim_id=claim.id, evidence_item_id=evidence[evidence_index].id) + claims.append(claim) + with store.connect() as conn: + for entity_id in (10, 20, 30): + conn.execute( + """INSERT INTO entities + (id, canonical_name, entity_type, scope, created_at, updated_at) + VALUES (?, ?, 'system', 'project:test', ?, ?)""", + (entity_id, f"system-{entity_id}", "2026-08-01", "2026-08-01"), + ) + conn.commit() + for claim, edge in zip( + (claims[0], claims[1], claims[1], claims[2]), + ((10, 20), (10, 20), (20, 30), (20, 30)), + ): + capture.add_edge_support( + source_entity_id=edge[0], + target_entity_id=edge[1], + relation="depends_on", + supporting_claim_id=claim.id, + scope="project:test", + ontology_version="personal-v1", + ) + return service, capture, sources + + +def test_candidate_steward_promotion_and_retirement_staleness(tmp_path) -> None: + service, capture, sources = _graph_fixture(tmp_path) + baseline = public_recall( + "blocker dependency", + scope_allowlist=["project:test"], + retrieval_mode="legacy", + db=service.store.db_path, + workspace=tmp_path, + ) + raw = json.dumps( + { + "decision": "emit", + "name": "Shared blocker chain", + "observation_type": "dependency", + "summary": "Three blockers share a two-step dependency chain.", + "assertions": [{"text": "All blockers are in the supplied chain.", "supporting_claim_ids": [1, 2, 3]}], + } + ) + engine = GraphObservationEngine(service.store, llm_call=lambda _system, _prompt: raw) + engine.repo.queue_discovery( + tenant_id=None, + scope="project:test", + ontology_version="personal-v1", + cycle_hour="2026-08-12T20", + ) + + discovery = engine.process_discovery(owner="observer", scope="project:test") + synthesis = engine.process_synthesis(owner="observer", scope="project:test") + candidates = engine.repo.scope_observations(scope="project:test", tenant_id=None) + review = review_observation_candidates(service.store, scope="project:test") + observation_id = int(candidates[0]["observation_claim_id"]) + ordinary = public_recall( + "blocker dependency", + scope_allowlist=["project:test"], + retrieval_mode="legacy", + db=service.store.db_path, + workspace=tmp_path, + ) + enriched = public_recall( + "blocker dependency", + scope_allowlist=["project:test"], + retrieval_mode="legacy", + include_observations=True, + observation_limit=2, + db=service.store.db_path, + workspace=tmp_path, + ) + graph_due = capture.due_confirmed_graph_claims(scope="project:test", limit=20) + dashboard = graph_observations_payload(service, scope="project:test") + capture.retire_source(sources[0].id, reason="fixture retirement") + + assert discovery.synthesis_queued == 1 + assert synthesis.emitted == 1 + assert review["confirmed"] == 1 + assert ordinary.output == baseline.output + assert ordinary.observations == () + assert observation_id not in {int(row["claim_id"]) for row in ordinary.claims} + assert enriched.observations[0]["claim_id"] == observation_id + assert "DERIVED OBSERVATIONS" in enriched.output + assert observation_id not in {int(row["claim_id"]) for row in graph_due} + assert dashboard["observations"][0]["observation_type"] == "dependency" + assert len(dashboard["observations"][0]["supports"]) == 4 + assert dashboard["observations"][0]["lifecycle"] + assert service.store.get_claim(observation_id).status == "stale" + + +def test_dashboard_hydration_adds_observation_panel_and_scripts() -> None: + html = "__GRAPH_OBSERVATIONS_SECTION____GRAPH_OBSERVATIONS_FUNCTIONS____GRAPH_OBSERVATIONS_EVENTS__" + + hydrated = hydrate_dashboard_html(html) + + assert "Derived Observations" in hydrated + assert "refreshGraphObservations" in hydrated + assert "graph-observation-refresh" in hydrated + + +def test_versioned_offline_corpus_meets_structural_quality_gate() -> None: + corpus = Path("benchmarks/fixtures/graph_observations_v1.json") + payload = json.loads(corpus.read_text(encoding="utf-8")) + categories = {str(case["category"]) for case in payload["cases"]} + + report = evaluate_corpus(corpus) + + assert report["cases"] >= 40 + assert { + "dependency_chain", + "root_cause", + "recurring_pattern", + "unrelated_similarity", + "hub", + "conflict", + "retired_evidence", + "scope_boundary", + "sensitive_data", + } <= categories + assert report["precision"] >= 0.95 + assert report["recall"] >= 0.95 + assert report["failures"] == [] + + +def test_provider_failure_is_retryable_and_creates_no_observation(tmp_path) -> None: + service, _capture, _sources = _graph_fixture(tmp_path) + + def provider_failure(_system: str, _prompt: str) -> str: + raise TimeoutError("fixture provider timeout") + + engine = GraphObservationEngine(service.store, llm_call=provider_failure) + engine.repo.queue_discovery( + tenant_id=None, + scope="project:test", + ontology_version="personal-v1", + cycle_hour="2026-08-12T23", + ) + engine.process_discovery(owner="observer", scope="project:test") + + result = engine.process_synthesis(owner="observer", scope="project:test") + with service.store.connect() as conn: + job = conn.execute("SELECT status, error_code FROM graph_observation_jobs WHERE stage='synthesize'").fetchone() + + assert result.failed == 1 + assert engine.repo.scope_observations(scope="project:test", tenant_id=None) == [] + assert tuple(job) == ("retryable", "synthesis_failed") + + +def test_dream_cycle_enforces_one_three_call_batch_per_scope(monkeypatch) -> None: + calls: list[tuple[str, str]] = [] + + class FakeRepository: + def queue_discovery(self, **kwargs): + calls.append(("queue", str(kwargs["tenant_id"]))) + return None, True + + class FakeEngine: + def __init__(self, _store, *, llm_call): + self.repo = FakeRepository() + + def process_discovery(self, *, owner, scope): + calls.append(("discover", scope)) + return SimpleNamespace(synthesis_queued=4, failed=0) + + def process_synthesis(self, *, owner, scope): + calls.append(("synthesize", scope)) + return SimpleNamespace(emitted=3, failed=0) + + import memorymaster.knowledge.graph_observation_engine as engine_module + + monkeypatch.setattr(engine_module, "GraphObservationEngine", FakeEngine) + worker = DreamWorker( + None, + SimpleNamespace(store=object()), + object(), + object(), + now=lambda: datetime(2026, 8, 12, 23, tzinfo=timezone.utc), + ) + monkeypatch.setattr( + worker, + "_observation_scope_pairs", + lambda _scope: [("project:test", "tenant-a"), ("project:test", "tenant-b")], + ) + + result = worker._run_graph_observations(owner="dream-worker", scope="project:test", synthesize=True) + + assert calls.count(("discover", "project:test")) == 1 + assert calls.count(("synthesize", "project:test")) == 1 + assert result["discovery_queued"] == 2 + assert result["emitted"] == 3 diff --git a/tests/test_public_cli.py b/tests/test_public_cli.py index 0bf0b720..b5217789 100644 --- a/tests/test_public_cli.py +++ b/tests/test_public_cli.py @@ -31,10 +31,20 @@ def test_cli_remember_recall_forget_improve_contract(cli_env, capsys) -> None: assert remembered["api_version"] == "memorymaster.public.v1" source_id = remembered["source_item"]["id"] - assert main([*_base(db, workspace), "recall", "Captured CLI note."]) == 0 + assert main( + [ + *_base(db, workspace), + "recall", + "Captured CLI note.", + "--include-observations", + "--observation-limit", + "5", + ] + ) == 0 recalled = json.loads(capsys.readouterr().out) assert recalled["api_version"] == "memorymaster.public.v1" assert recalled["trust_mode"] == "trusted" + assert recalled["observations"] == [] assert main( [*_base(db, workspace), "forget", "--source-item-id", str(source_id)] diff --git a/tests/test_public_demo.py b/tests/test_public_demo.py index 4cd5c859..00ff9b31 100644 --- a/tests/test_public_demo.py +++ b/tests/test_public_demo.py @@ -17,12 +17,15 @@ def test_disposable_demo_covers_public_lifecycle() -> None: report = run_disposable_demo() assert report["temporary_database_disposed"] is True - assert report["captures"] == 2 - assert report["fixture_jobs_completed"] == 2 - assert report["candidate_claims_created"] == 2 + assert report["captures"] == 3 + assert report["fixture_jobs_completed"] == 3 + assert report["candidate_claims_created"] == 3 assert report["promoted_claim_id"] in report["recall_claim_ids"] assert report["recall_citations"] assert report["graph_paths"][0]["relation"] == "participates_in" + assert report["observation_recall"][0]["claim_id"] == report["observation_claim_id"] + assert len(report["observation_supports"]) >= 4 + assert report["observation_status_after_retirement"] == "stale" def test_demo_cli_emits_versioned_json(capsys) -> None: diff --git a/tests/test_public_mcp.py b/tests/test_public_mcp.py index 679d8b84..6006d869 100644 --- a/tests/test_public_mcp.py +++ b/tests/test_public_mcp.py @@ -35,10 +35,15 @@ def test_public_mcp_contract_round_trip(mcp_env) -> None: assert remembered["api_version"] == "memorymaster.public.v1" recalled = mcp_server.recall( - query="MCP public capture", db=db, workspace=workspace + query="MCP public capture", + include_observations=True, + observation_limit=5, + db=db, + workspace=workspace, ) assert recalled["ok"] is True assert recalled["trust_mode"] == "trusted" + assert recalled["observations"] == () preview = mcp_server.forget( source_item_id=remembered["source_item"]["id"],