feat(backfill): opt-in consolidation guards: content scope, duplicate filter, sensitivity rejection (T-1794)

ROADMAP.md

@@ -1,4 +1,4 @@
-<!-- doc-head: sole roadmap; installed recall integrity and AO-7 PR build with lane semaphore test evidence -->
+<!-- doc-head: sole roadmap; installed recall integrity and default-off AO-7 consolidation guards -->
 # 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.

artifacts/T-1794-verification.md

@@ -0,0 +1,90 @@
+<!-- doc-head: T-1794 source verification and retained execution failures -->
+<!-- Covers: disposable checks, fixture corrections and navigation limits. -->
+<!-- Read when: reviewing the opt-in AO-7 consolidation admission guards. -->
+<!-- /doc-head -->
+# 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.

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": [

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
 

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,18 +988,20 @@ 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}
                     pending = manifest.setdefault(key, {"candidate_id": decision.candidate_id,
                                                        "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)