diff --git a/ROADMAP.md b/ROADMAP.md index 20929fa9..166ed207 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -1,4 +1,4 @@ - + # MemoryMaster roadmap # Covers: live Jev decision ledger (4.9.0 checklist), useful-memory delivery/evaluation, workflow analytics, governed observations, always-on context layer program and deferrals. # Key terms: Workflow Intelligence, governed skills, graph observations, user profile, code navigation. @@ -754,6 +754,16 @@ reading and Sonnet only for consolidating kept candidates, judge Haiku batch M5 per 1,000 sessions. Depends on: AO-2; starts as soon as AO-2 ships (operator decision). +T-1794 source follow-up (2026-10-25): `MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS=1` +adds content scope resolution, counted generic/duplicate/sensitive rejection, +and reviewed scope propagation through application. Default is off. The prior +run recorded a passed independent source review. Fresh disposable verification +passed: guards 22, CLI 40, quality 13 and public demo 2; Ruff, release-truth and +contract-witness also passed. Permission failures and corrected test fixtures +are retained in [the verification record](artifacts/T-1794-verification.md). +GitNexus/Serena tools and a local index were unavailable; scoped source inspection +covered the changed callers. No live campaign or authoritative database was opened. + - [ ] Routed. - [ ] Red test: not applicable. - [ ] Green locally and in CI. diff --git a/artifacts/T-1794-verification.md b/artifacts/T-1794-verification.md new file mode 100644 index 00000000..8f6b1538 --- /dev/null +++ b/artifacts/T-1794-verification.md @@ -0,0 +1,90 @@ + + + + +# T-1794 verification + +Resumed the existing test-first commit `73ff057` and implementation commit +`84fcf56` without replacing their work. The flag remains default off; no live +campaign, authoritative database or operator home configuration was inspected. +Consolidation activation and real-output remeasurement remain with the lane. + +## Fresh checks, 2026-10-25 + +Each pytest invocation selected exactly one file and used disposable fixtures. + +| Check | Result | +|---|---| +| `tests/test_dreaming_backfill_consolidate_guards.py` | 22 passed | +| `tests/test_dreaming_backfill_cli.py` | 40 passed | +| `tests/test_dreaming_backfill_quality.py` | 13 passed | +| `tests/test_public_demo.py` | 2 passed | +| Ruff on both changed production modules and the new test file | Passed | +| `python scripts/generate_release_truth.py`, then `--check` | Generated and verified | +| `python scripts/check_contract_witness.py --base origin/main` | Passed; no orphaned witnesses | +| `python scripts/check_branch_freshness.py` | Zero commits behind origin/main | + +The scope tests cover alias keys in subjects and quotes, verified worktree Git +ownership, known canonical repository names, absent signals, generic rules, +ambiguous ownership and reviewed scope propagation through application. +Resolution reuses `_scope_aliases` and `resolve_transcript_workspace_scope`. +Duplicate checks reuse `govern.jobs.extractor.normalize_claim_text`; tests also +cover pagination and scope/tenant boundaries. Sensitivity tests inject a +synthetic credential into each of five candidate fields at the read boundary, +assert the existing `redact_text` filter detects it, verify zero provider calls +and scan every fixture file to prove the rejected text was never persisted. +Unset and zero-valued flags both retain the original consolidation behavior. + +## Preserved failures and corrections + +The previous run recorded 13 fixture permission errors before behavior ran. +On resume, ordinary pytest again produced 22 setup errors and a cleanup +`PermissionError`. A worktree-local temporary directory alone was insufficient: +Python's Windows owner-only directory mode prevented sandbox access. + +A process-local bootstrap preserved inherited directory permissions for test +fixtures (`0o777` instead of `0o700`). This exposed 9 failing tests and 13 passing +tests: four failures omitted the required `citations` argument when seeding +claims, and five used an email address that the existing security filter does +not detect. The fixtures now pass explicit empty citations and use a synthetic +credential with an assertion that the existing filter detects it. Production +security patterns were unchanged. The corrected guard suite passed all 22. + +The first public demo attempt failed both tests because `TemporaryDirectory` +also creates owner-only directories through `os.mkdir`. Its successful retry +set `tempfile.tempdir` to a disposable worktree directory and adjusted only +owner-only directory creation beneath that directory. Application code and +machine permissions were unchanged. A pytest-asyncio fixture-scope deprecation +warning was present in the runs. + +The successful demo bootstrap was: + +```python +import os +import tempfile +from pathlib import Path + +root = (Path.cwd() / "artifacts" / "t1794-demo-temp").resolve() +root.mkdir(parents=True, exist_ok=True) +tempfile.tempdir = str(root) +original = os.mkdir +os.mkdir = lambda path, mode=0o777, *, dir_fd=None: original( + path, + 0o777 if mode == 0o700 and Path(path).is_relative_to(root) else mode, + dir_fd=dir_fd, +) +import pytest + +raise SystemExit(pytest.main([ + "tests/test_public_demo.py", "-q", "--basetemp=" + str(root / "pytest"), +])) +``` + +## Navigation limits + +Deferred tool discovery exposed no GitNexus or Serena tools. +`python scripts/check_code_navigation.py` reported `INDEX_MISSING`. +Impact/change detection therefore used scoped source searches and Git diffs: +the helper's callers are the consolidation, status, quality population and +application paths in `backfill.py`. This resume changed only tests, generated +release metadata and documentation. No production symbols were edited here. diff --git a/docs/generated/release-truth.json b/docs/generated/release-truth.json index aad82190..3f3e8e5b 100644 --- a/docs/generated/release-truth.json +++ b/docs/generated/release-truth.json @@ -144,7 +144,7 @@ "console_entrypoints": 9, "mcp_tools": 51, "ops_cli_commands": 5, - "pytest_test_functions": 5935 + "pytest_test_functions": 5950 }, "feature_profile_matrix": { "capture_hook": [ diff --git a/docs/generated/release-truth.md b/docs/generated/release-truth.md index 4d92331a..663ecf99 100644 --- a/docs/generated/release-truth.md +++ b/docs/generated/release-truth.md @@ -13,7 +13,7 @@ Do not edit this file by hand. Run `python scripts/generate_release_truth.py`. - Main CLI commands: **127** - Operations CLI commands: **5** - Console entrypoints: **9** -- Pytest source test functions: **5935** +- Pytest source test functions: **5950** ## MCP tools diff --git a/memorymaster/dreaming/backfill.py b/memorymaster/dreaming/backfill.py index d550392e..91d50396 100644 --- a/memorymaster/dreaming/backfill.py +++ b/memorymaster/dreaming/backfill.py @@ -19,6 +19,7 @@ from typing import Any, Callable, Mapping from memorymaster.dreaming.backfill_guard import PauseLatch, evaluate_backfill_guard +from memorymaster.dreaming import backfill_consolidate_guards as consolidate_guards from memorymaster.dreaming.providers import ProviderCallError, ProviderOutputError from memorymaster.dreaming.backfill_quality import evaluate_cohort_quality, fingerprint, receipt_is_current from memorymaster.dreaming.backfill_selection import ( @@ -138,7 +139,8 @@ def read_backfill_status(state_dir: Path, *, campaign_id: str | None = None, dry "COALESCE(json_extract(metadata_json,'$.selection.eligible_chars'),0) AS eligible_chars," "COALESCE(json_extract(metadata_json,'$.selection.selected_chars'),0) AS selected_chars," "COALESCE(json_extract(metadata_json,'$.selection.sampled_out_chars'),0) AS sampled_out_chars," - "COALESCE(json_extract(metadata_json,'$.selection.gap_count'),0) AS coverage_gaps " + "COALESCE(json_extract(metadata_json,'$.selection.gap_count'),0) AS coverage_gaps," + "json_extract(metadata_json,'$.consolidate_guards') AS consolidate_guards " "FROM backfill_inventory" + where, params).fetchall() reservations = conn.execute("SELECT stage,COUNT(*) AS calls,SUM(input_tokens) AS input_tokens,SUM(output_tokens) AS output_tokens FROM backfill_reservations" + where + " GROUP BY stage", params).fetchall() dispositions: dict[str, int] = {} @@ -174,7 +176,7 @@ def read_backfill_status(state_dir: Path, *, campaign_id: str | None = None, dry guards = json.loads(guard_path.read_text(encoding="utf-8")) if guard_path.exists() else {} pauses = {key: value["state"] for key, value in guards.items() if (campaign_id is None or key.startswith(campaign_id + ":")) and value.get("state") != "running"} - return {"state": "inventoried", "files": len(rows), "sessions": len(sessions), + return {**consolidate_guards.status_counts(rows), "state": "inventoried", "files": len(rows), "sessions": len(sessions), "completed_sessions": len(completed), "progress": len(completed) / len(sessions) if sessions else 0, "dispositions": dispositions, "projects": projects, "sources": sources, "eligible_chars": sum(r["eligible_chars"] for r in rows), @@ -671,19 +673,24 @@ def _consolidate(self, rows: list[dict[str, Any]], run_id: str) -> None: captures = self._captures(rows) groups: dict[str, list] = {} decisions: dict[int, list[dict[str, Any]]] = {} - for capture in captures: - for raw in capture.get("extraction") or []: - candidate = DreamCandidate(**raw) - scope = "personal" if candidate.scope_class == "personal" else capture["scope"] - groups.setdefault(scope, []).append((capture, bind_source(candidate, capture))) - decisions[capture["id"]] = [] own_ids = self._owned_ids(rows) + guards_changed = False + if consolidate_guards.enabled(): + groups, decisions, guards_changed = consolidate_guards.prepare(rows, captures, service, own_ids) + else: + guards_changed = rows[0]["metadata"].pop("consolidate_guards", None) is not None + for capture in captures: + for raw in capture.get("extraction") or []: + candidate = DreamCandidate(**raw) + scope = "personal" if candidate.scope_class == "personal" else capture["scope"] + groups.setdefault(scope, []).append((capture, bind_source(candidate, capture))) + decisions[capture["id"]] = [] snapshots = {scope: [item for item in worker._current_claims(scope, [c for _, c in pairs]) if item["id"] not in own_ids] for scope, pairs in groups.items()} hashes = {scope: fingerprint(claims) for scope, claims in snapshots.items()} prior = rows[0]["metadata"].get("reference_hashes") - if prior is not None and prior != hashes: - # References changed: retain extraction but renew decisions and M5. + if guards_changed or (prior is not None and prior != hashes): + # References or guard verdicts changed: renew decisions and M5. with self.ledger._connect() as conn: for capture in captures: conn.execute("UPDATE dream_captures SET decisions_json=NULL,state='extracted' WHERE id=?", (capture["id"],)) @@ -761,7 +768,10 @@ def _population(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: index = next((i for i, m in enumerate(window["messages"]) if m["message_id"] == candidate["evidence_message_id"]), 0) origin = window["ranges"][index]["stream"] family = next(r["metadata"]["source_family"] for r in rows if r["stream_id"] == origin) - bound = bind_source(DreamCandidate(**candidate), capture) + effective, item = consolidate_guards.candidate_context(rows, capture, DreamCandidate(**candidate)) + if effective is None: + continue + bound = bind_source(item, effective) population.append({**bound.to_dict(), "decision": accepted[candidate["candidate_id"]], "capture_hash": capture["content_hash"], "source_family": family}) return population @@ -936,8 +946,10 @@ def _apply(self, rows: list[dict[str, Any]], run_id: str) -> None: grouped: dict[str, list[DreamCandidate]] = {} for capture in self._captures(rows): for raw in capture["extraction"]: - candidate = DreamCandidate(**raw) - scope = "personal" if candidate.scope_class == "personal" else capture["scope"] + effective, candidate = consolidate_guards.candidate_context(rows, capture, DreamCandidate(**raw)) + if effective is None: + continue + scope = "personal" if candidate.scope_class == "personal" else effective["scope"] grouped.setdefault(scope, []).append(candidate) def verify_references(): own_ids = self._owned_ids(rows) @@ -976,7 +988,9 @@ def verify_references(): if not self._source_valid(row): raise ValueError("source changed before application") self._reserve(row["cohort_id"], "apply", 0, 0) - candidate = candidates[decision.candidate_id] + effective, candidate = consolidate_guards.candidate_context(rows, capture, candidates[decision.candidate_id]) + if effective is None: + raise ValueError("guard-rejected candidate cannot apply") batch = f"ao7-{self.config.campaign_id}-{row['cohort_id']}-{self._application_ordinal(rows, key) // 200}" operation_id = f"ao7-{self.config.campaign_id}-{row['cohort_id']}-{key}" outcome: dict[str, Any] = {"operation_id": operation_id} @@ -984,10 +998,10 @@ def verify_references(): "source_hash": window["identity"], "intake_batch_id": batch, "operation_id": operation_id, "tenant_id": service.tenant_id, - "scope": "personal" if candidate.scope_class == "personal" else row["scope"], + "scope": "personal" if candidate.scope_class == "personal" else effective["scope"], "created_claim_ids": [], "reused_claim_ids": [], "proposal_ids": []}) self._save(row) # intent precedes cross-database writes - application_row = {**capture, "intake_batch_id": batch, "outcome_receipt": outcome} + application_row = {**effective, "intake_batch_id": batch, "outcome_receipt": outcome} # Child ranges retain the parent's independent evidence identity. ranges = window["ranges"] message_index = next((i for i, m in enumerate(window["messages"]) if m["message_id"] == candidate.evidence_message_id), 0) diff --git a/memorymaster/dreaming/backfill_consolidate_guards.py b/memorymaster/dreaming/backfill_consolidate_guards.py new file mode 100644 index 00000000..c1a7c2a8 --- /dev/null +++ b/memorymaster/dreaming/backfill_consolidate_guards.py @@ -0,0 +1,152 @@ +"""Opt-in AO-7 admission checks; receipts contain identities and verdicts only.""" +from __future__ import annotations + +import json +import os +import re +from dataclasses import replace + +from memorymaster.core.scope_utils import _scope_aliases, resolve_transcript_workspace_scope +from memorymaster.core.security import redact_text, scan_persisted_value, validate_persisted_metadata +from memorymaster.dreaming.models import DreamCandidate, DreamDecision +from memorymaster.dreaming.source_review import bind_source +from memorymaster.govern.jobs.extractor import normalize_claim_text + +FLAG = "MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS" +COUNTERS = ("generic", "duplicate", "sensitive", "ambiguous_scope", "rescoped") +_PATH = re.compile(r'''["'`]((?:[A-Za-z]:[/\\]|/)[^"'`\r\n]+)["'`]|(?:[A-Za-z]:[/\\]|/)[^\s"'`<>]+''') +_RULE = re.compile(r"^(?:(?:always|never|siempre|nunca)\s+)?(?:verify|check|run|verificar)\s+(?:the\s+)?(?:evidence|tests|checks|evidencia|pruebas)\b", re.I) +_GENERAL = {"agent", "agents", "assistant", "assistants", "developer", "developers", "user", "users"} + + +def enabled() -> bool: + return os.environ.get(FLAG) == "1" + + +def _names(rows, known_scopes): + aliases = _scope_aliases() + names = dict(aliases) + for scope in {*aliases.values(), *known_scopes, *(row["scope"] for row in rows)}: + if scope and scope.startswith("project:"): + names.setdefault(scope.removeprefix("project:"), scope) + return names + + +def _content_scopes(candidate, names): + # Only the cited quote and subject establish ownership, not a generated text hint. + evidence = candidate.subject + "\n" + candidate.evidence_quote + scopes = set() + for match in _PATH.finditer(evidence): + path = match.group(1) or match.group().rstrip(".,;:)") + scope = resolve_transcript_workspace_scope(path) + if scope: + scopes.add(scope) + # A verified worktree path is stronger than its disposable checkout name. + without_paths = _PATH.sub(" ", evidence) + for name, scope in names.items(): + if re.search(r"(? 1: + note["reason"] = "ambiguous_scope" + elif scopes: + note.update(scope=scopes.pop(), scope_class="project") + elif _generic(candidate, scope): + note.update(scope_class="generic", reason="generic") + if note["reason"] == "accepted": + scope = note["scope"] + if normalize_claim_text(candidate.text) in texts.get(scope, set()): + note["reason"] = "duplicate" + notes[candidate.candidate_id] = note + validate_persisted_metadata({"guard_scope": note["scope"], "guard_scope_class": note["scope_class"]}) + if note["reason"] != "accepted": + metadata[note["reason"]] += 1 + decisions[capture["id"]].append(DreamDecision( + candidate.candidate_id, "ignore", "backfill guard: " + note["reason"], 1.0, + ).to_dict()) + continue + metadata["rescoped"] += int(note["scope"] != capture["scope"] and note["scope_class"] == "project") + candidate = replace(candidate, scope_class=note["scope_class"]) + effective = {**capture, "scope": note["scope"]} + groups.setdefault(note["scope"], []).append((effective, bind_source(candidate, effective))) + previous = rows[0]["metadata"].get("consolidate_guards") + rows[0]["metadata"]["consolidate_guards"] = metadata + return groups, decisions, previous != metadata + + +def candidate_context(rows, capture, candidate): + """Carry a reviewed scope through quality receipts and eventual application.""" + for row in rows: + note = (row["metadata"].get("consolidate_guards", {}).get("candidates", {}) + .get(str(capture["id"]), {}).get(candidate.candidate_id)) + if note: + if note["reason"] != "accepted": + return None, candidate + return {**capture, "scope": note["scope"]}, replace(candidate, scope_class=note["scope_class"]) + return capture, candidate + + +def status_counts(rows): + receipts = [json_value for row in rows if (json_value := row["consolidate_guards"])] + if not receipts: + return {} + parsed = [json.loads(value) for value in receipts] + return {"consolidate_guards": {key: sum(item.get(key, 0) for item in parsed) for key in COUNTERS}} diff --git a/tests/test_dreaming_backfill_consolidate_guards.py b/tests/test_dreaming_backfill_consolidate_guards.py new file mode 100644 index 00000000..7ff5f635 --- /dev/null +++ b/tests/test_dreaming_backfill_consolidate_guards.py @@ -0,0 +1,271 @@ +"""Default-off AO-7 guards, exercised only against disposable campaigns.""" + +import json +from dataclasses import replace + +import pytest + +from memorymaster.dreaming.backfill import read_backfill_status +from memorymaster.core.security import redact_text +from test_dreaming_backfill_cli import _campaign, isolated_configuration as isolated_configuration + + +def _extracted(tmp_path): + fixture = _campaign(tmp_path) + fixture.runner.inventory() + fixture.runner.run(stage="extract") + capture = fixture.runner._captures(fixture.runner._rows())[0] + return fixture, capture + + +def _candidate(fixture, capture, **changes): + candidate = {**capture["extraction"][0], **changes} + fixture.runner.ledger.set_extraction(capture["id"], [candidate], "fixture") + # Keep the fixture's cited message exact without rewriting frozen transcripts. + original = fixture.runner._captures + + def captures(rows): + result = original(rows) + for item in result: + item["messages"][0]["text"] = candidate["evidence_quote"] + return result + + fixture.runner._captures = captures + return candidate + + +@pytest.mark.parametrize("signal", ["subject", "quote", "worktree"]) +def test_content_scope_uses_project_alias(tmp_path, monkeypatch, signal): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + aliases = tmp_path / "aliases.json" + aliases.write_text(json.dumps({"aliases": {"bridge-checkout": "project:wezbridge"}})) + monkeypatch.setenv("MEMORYMASTER_SCOPE_ALIASES", str(aliases)) + fixture, capture = _extracted(tmp_path) + changes = {"subject": "bridge-checkout"} if signal == "subject" else { + "evidence_quote": "Use bridge-checkout for relay delivery."} + if signal == "worktree": + repo = tmp_path / "bridge-checkout" + (repo / ".git" / "worktrees" / "relay").mkdir(parents=True) + worktree = tmp_path / "_wt" / "relay" + worktree.mkdir(parents=True) + gitdir = repo / ".git" / "worktrees" / "relay" + (worktree / ".git").write_text(f"gitdir: {gitdir.as_posix()}") + (gitdir / "commondir").write_text("../..") + (gitdir / "gitdir").write_text(str(worktree / ".git")) + changes = {"evidence_quote": f'Use the relay checkout "{worktree.as_posix()}".'} + original = fixture.consolidator.consolidate + + def privacy_review(*args, **kwargs): + result = original(*args, **kwargs) + decisions = tuple(replace(d, source_review={ + **d.source_review, "verdict": "reject", + "checks": {**d.source_review["checks"], "privacy": False}, + }) for d in result.decisions) + return replace(result, decisions=decisions) + + fixture.consolidator.consolidate = privacy_review + _candidate(fixture, capture, **changes) + result = fixture.runner.run(stage="consolidate") + assert fixture.consolidator.calls[0][2] == "project:wezbridge" + decision = fixture.runner._captures(fixture.runner._rows())[0]["decisions"][0] + assert decision["action"] == ("ignore" if signal == "worktree" else "add") + assert result["consolidate_guards"]["rescoped"] == 1 + population = fixture.runner._population(fixture.runner._rows()) + if signal == "worktree": + assert not population # Existing source-review privacy policy still rejects absolute paths. + else: + assert population[0]["source_context"]["capture_scope"] == "project:wezbridge" + # Re-running keeps counts stable and reuses the reviewed decisions. + fixture.runner.run(stage="consolidate") + assert len(fixture.consolidator.calls) == 1 + + +def test_rescoped_candidate_is_applied_in_reviewed_scope(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + aliases = tmp_path / "aliases.json" + aliases.write_text(json.dumps({"aliases": {"bridge-checkout": "project:wezbridge"}})) + monkeypatch.setenv("MEMORYMASTER_SCOPE_ALIASES", str(aliases)) + fixture, capture = _extracted(tmp_path) + _candidate(fixture, capture, subject="bridge-checkout") + fixture.runner.run(stage="consolidate") + fixture.runner.run(stage="quality") + fixture.runner.run(stage="apply", apply_candidates=True) + claims = fixture.service.list_claims(status="candidate", requesting_agent="dream-worker") + assert len(claims) == 1 and claims[0].scope == "project:wezbridge" + + +def test_no_repo_signal_retains_inventory_scope(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + fixture, _ = _extracted(tmp_path) + fixture.runner.run(stage="consolidate") + assert fixture.consolidator.calls[0][2] == "project:repo" + + +def test_bare_name_of_known_repo_resolves_without_folder_alias(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + fixture, capture = _extracted(tmp_path) + fixture.service.ingest(text="Relay deliveries retain recipient identity.", + citations=[], scope="project:wezbridge", source_agent="fixture") + _candidate(fixture, capture, subject="wezbridge") + result = fixture.runner.run(stage="consolidate") + assert fixture.consolidator.calls[0][2] == "project:wezbridge" + assert result["consolidate_guards"]["rescoped"] == 1 + + +def test_generic_rule_is_classified_and_counted(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + fixture, capture = _extracted(tmp_path) + candidate = _candidate(fixture, capture, text="Always verify evidence before making a claim.", + subject="agents", predicate="must", object_value="verify evidence", + evidence_quote="Always verify evidence before making a claim.") + result = fixture.runner.run(stage="consolidate") + assert not fixture.consolidator.calls + assert result["consolidate_guards"]["generic"] == 1 + metadata = fixture.runner._rows()[0]["metadata"]["consolidate_guards"] + assert metadata["candidates"][str(capture["id"])][candidate["candidate_id"]]["scope_class"] == "generic" + assert fixture.runner._cohort_complete(fixture.runner._rows(), decisions=True) + + +def test_normalized_duplicate_is_skipped_and_counted(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + fixture, capture = _extracted(tmp_path) + text = capture["extraction"][0]["text"] + fixture.service.ingest(text=" " + text.upper().replace(" ", " ") + " ", + citations=[], scope="project:repo", source_agent="fixture") + result = fixture.runner.run(stage="consolidate") + assert not fixture.consolidator.calls + assert result["consolidate_guards"]["duplicate"] == 1 + assert read_backfill_status(fixture.runner.config.state_dir)["consolidate_guards"]["duplicate"] == 1 + + +@pytest.mark.parametrize("field", ["text", "subject", "predicate", "object_value", "evidence_quote"]) +def test_sensitive_candidate_is_rejected_without_persisting_text(tmp_path, monkeypatch, field): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + fixture, _ = _extracted(tmp_path) + secret = "Incident password=" + "SyntheticOnly_unsafe_fixture" + assert redact_text(secret)[1], "The fixture must trigger the existing ingest filter." + # Inject an old, unsafe extraction at the read boundary; never seed it on disk. + original = fixture.runner._captures + + def unsafe(rows): + captures = original(rows) + captures[0]["extraction"][0][field] = secret + return captures + + monkeypatch.setattr(fixture.runner, "_captures", unsafe) + result = fixture.runner.run(stage="consolidate") + assert not fixture.consolidator.calls + assert result["consolidate_guards"]["sensitive"] == 1 + for path in tmp_path.rglob("*"): + if path.is_file(): + assert secret.encode() not in path.read_bytes() + + +@pytest.mark.parametrize("flag", [None, "0"]) +def test_flag_off_preserves_consolidation(tmp_path, monkeypatch, flag): + if flag is None: + monkeypatch.delenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", raising=False) + else: + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", flag) + fixture, capture = _extracted(tmp_path) + _candidate(fixture, capture, subject="wezbridge", text="Always verify evidence before making a claim.") + result = fixture.runner.run(stage="consolidate") + assert fixture.consolidator.calls[0][2] == "project:repo" + assert "consolidate_guards" not in result + + +def test_enabling_guards_replaces_existing_unguarded_decisions(tmp_path, monkeypatch): + monkeypatch.delenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", raising=False) + fixture, capture = _extracted(tmp_path) + _candidate(fixture, capture, subject="agents", predicate="must", + text="Always verify evidence before making a claim.", + evidence_quote="Always verify evidence before making a claim.") + fixture.runner.run(stage="consolidate") + assert len(fixture.runner._population(fixture.runner._rows())) == 1 + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + result = fixture.runner.run(stage="consolidate") + assert result["consolidate_guards"]["generic"] == 1 + assert not fixture.runner._population(fixture.runner._rows()) + + +def test_duplicate_filter_preserves_other_scopes_and_tenants(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + fixture, capture = _extracted(tmp_path) + text = capture["extraction"][0]["text"] + fixture.service.ingest(text=text, citations=[], scope="project:elsewhere", source_agent="fixture") + from memorymaster.core.service import MemoryService + + other = MemoryService(fixture.db, workspace_root=fixture.repo, tenant_id="other") + other.ingest(text=text, citations=[], scope="project:repo", source_agent="fixture") + result = fixture.runner.run(stage="consolidate") + assert len(fixture.consolidator.calls) == 1 + assert result["consolidate_guards"]["duplicate"] == 0 + + +def test_scope_alias_does_not_match_inside_another_word(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + aliases = tmp_path / "aliases.json" + aliases.write_text(json.dumps({"aliases": {"bridge": "project:wezbridge"}})) + monkeypatch.setenv("MEMORYMASTER_SCOPE_ALIASES", str(aliases)) + fixture, capture = _extracted(tmp_path) + _candidate(fixture, capture, subject="cambridge", evidence_quote="The cambridge schema retains citations.") + fixture.runner.run(stage="consolidate") + assert fixture.consolidator.calls[0][2] == "project:repo" + + +def test_concrete_decision_without_repo_signal_keeps_inventory_scope(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + fixture, capture = _extracted(tmp_path) + _candidate(fixture, capture, subject="storage", text="Use SQLite WAL for concurrent readers.", + evidence_quote="Use SQLite WAL for concurrent readers.") + fixture.runner.run(stage="consolidate") + assert fixture.consolidator.calls[0][2] == "project:repo" + + +def test_rule_using_this_repos_tool_is_not_generic(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + aliases = tmp_path / "aliases.json" + aliases.write_text(json.dumps({"aliases": {"repo": "project:memorymaster"}})) + monkeypatch.setenv("MEMORYMASTER_SCOPE_ALIASES", str(aliases)) + fixture, capture = _extracted(tmp_path) + _candidate(fixture, capture, subject="agents", text="Always call recall() before reusing a claim.", + evidence_quote="Always call recall() before reusing a claim.") + result = fixture.runner.run(stage="consolidate") + assert fixture.consolidator.calls[0][2] == "project:memorymaster" + assert result["consolidate_guards"]["generic"] == 0 + + +def test_ambiguous_repo_signals_are_skipped(tmp_path, monkeypatch): + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + aliases = tmp_path / "aliases.json" + aliases.write_text(json.dumps({"aliases": {"alpha": "project:alpha", "beta": "project:beta"}})) + monkeypatch.setenv("MEMORYMASTER_SCOPE_ALIASES", str(aliases)) + fixture, capture = _extracted(tmp_path) + _candidate(fixture, capture, subject="alpha", evidence_quote="Use beta for delivery.") + result = fixture.runner.run(stage="consolidate") + assert not fixture.consolidator.calls + assert result["consolidate_guards"]["ambiguous_scope"] == 1 + + +def test_duplicate_lookup_continues_after_empty_authorized_page(tmp_path, monkeypatch): + from memorymaster.core.service import MemoryService + + monkeypatch.setenv("MEMORYMASTER_BACKFILL_CONSOLIDATE_GUARDS", "1") + fixture, capture = _extracted(tmp_path) + fixture.service.ingest(text=capture["extraction"][0]["text"], citations=[], scope="project:repo", source_agent="fixture") + original = MemoryService.list_claims_page + cursors = [] + + def pages(self, **kwargs): + if kwargs["status"] == "candidate": + cursors.append(kwargs["cursor"]) + if not kwargs["cursor"]: + return [], "next" + kwargs["cursor"] = "" + return original(self, **kwargs) + + monkeypatch.setattr(MemoryService, "list_claims_page", pages) + result = fixture.runner.run(stage="consolidate") + assert cursors == ["", "next"] + assert not fixture.consolidator.calls + assert result["consolidate_guards"]["duplicate"] == 1