Keep feature discovery reports compact and retain full audit artifacts (#303)

A full JiuwenSwarm discovery scan produces a projected 12–13 MB report
before its final boundary decisions, with repeated receipt metadata and
thousands of candidate descriptions. Feature discovery now publishes a
compact summary and retains the complete report as an immutable,
content-addressed artifact outside Git. The checkpoint and summary bind
its exact path, byte size and SHA-256; cached and prepared publication
reject missing, altered or escaped artifacts.

Completed unpublished legacy previews re-render through the native stage
using their saved scans, independent reviews and boundary audit without
calling either model again. Published historical catalogs remain
immutable. Source scope, prompts, model configuration, feature
definitions, budgets and correction limits are unchanged.

Validation: 176 focused tests pass in both implementation and
independent runs; native receipt/input assertions now inspect the real
complete artifact. Regressions cover large-report size, canonical bytes
across recovery, unchanged catalogs, containment, missing/tampered
archives and exact restoration, prepared publication, native boundary
failures, and legacy preview migration. Independent code review found no
remaining actionable issue. Full CI is required before merge.

Author: tzhouam

src/infermatrix_copilot/kb_service/init_feature_discovery.py

@@ -16,6 +16,7 @@
 import os
 from pathlib import Path
 import re
+import tempfile
 import threading
 import unicodedata
 
@@ -34,6 +35,7 @@
 MAX_ATTEMPTS = 4
 MAX_BASELINE_CHARS = 16000
 CATALOG_BOUNDARY_VERSION = "catalog-boundary-v1"
+COMPACT_REPORT_VERSION = "feature-discovery-compact-v1"
 RELATIONS = {"new", "implementation_supplement", "alias", "subcapability",
              "shared_component", "outdated", "unknown"}
 _SLUG = re.compile(r"[a-z0-9][a-z0-9-]{0,39}\Z")
@@ -110,6 +112,89 @@ def _json(value):
     return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
 
 
+def _report_archive_root(state_dir):
+    state_root = Path(state_dir).resolve()
+    root = state_root / "init" / "artifacts" / "feature-discovery"
+    if not root.resolve().is_relative_to(state_root):
+        raise InitError("discovery report archive escapes its state directory")
+    return root.resolve()
+
+
+def write_full_discovery_report(state_dir, report):
+    """Atomically create immutable full evidence outside publication files."""
+    data = (json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode()
+    digest = hashlib.sha256(data).hexdigest()
+    root = _report_archive_root(state_dir)
+    root.mkdir(parents=True, exist_ok=True)
+    path = root / (digest + ".json")
+    with tempfile.NamedTemporaryFile(dir=root, delete=False) as handle:
+        temporary = Path(handle.name)
+        handle.write(data)
+        handle.flush()
+        os.fsync(handle.fileno())
+    try:
+        try:
+            os.link(temporary, path)  # Exclusive atomic creation, never replacement.
+        except FileExistsError:
+            pass
+        if path.is_symlink() or path.read_bytes() != data:
+            raise InitError("immutable discovery report artifact differs; restore its original bytes")
+    finally:
+        temporary.unlink()
+    return {"path": str(path), "sha256": digest, "size_bytes": len(data)}
+
+
+def compact_discovery_report(report, artifact):
+    keys = ("schema_version", "repo", "pin", "run_config", "complete", "done", "catalog_sha256",
+            "feature_ids", "features", "owner_requests", "counts", "statistics", "scan_summary",
+            "index_sha256", "limits", "historical_denominator", "facet_denominator", "pacing")
+    compact = {key: deepcopy(report[key]) for key in keys if key in report}
+    audit = report["catalog_boundary_audit"]
+    compact["catalog_boundary_audit"] = {key: deepcopy(audit[key]) for key in ("identity", "done", "counts") if key in audit}
+    compact["catalog_boundary_audit"]["candidate_count"] = len(audit.get("candidate_ids", []))
+    compact.update(report_format=COMPACT_REPORT_VERSION, full_artifact=deepcopy(artifact),
+                   candidate_count=len(report.get("candidates", [])),
+                   unassociated_implementation_count=len(report.get("unassociated_implementation_paths", [])),
+                   unassociated_entry_lead_count=len(report.get("unassociated_entry_leads", [])))
+    for key in ("failures", "scope_suggestions", "candidate_limit_task_ids", "invalid_candidates"):
+        values = report.get(key, [])
+        compact[key] = {"count": len(values), "examples": deepcopy(values[:5]),
+                        "complete_list_in_full_artifact": True}
+    return compact
+
+
+def verify_full_discovery_report(state_dir, report, state):
+    """Check byte identity, containment and the genuine checkpoint binding."""
+    try:
+        if report.get("report_format") != COMPACT_REPORT_VERSION or state.get("report_format") != COMPACT_REPORT_VERSION:
+            raise ValueError("compact report marker is missing")
+        artifact = report["full_artifact"]
+        if artifact != state.get("full_artifact") or set(artifact) != {"path", "sha256", "size_bytes"}:
+            raise ValueError("report and checkpoint artifact bindings differ")
+        digest, size = artifact["sha256"], artifact["size_bytes"]
+        if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest) \
+                or type(size) is not int or size <= 0:
+            raise ValueError("artifact hash or size is invalid")
+        path = Path(artifact["path"])
+        expected = _report_archive_root(state_dir) / (digest + ".json")
+        if not path.is_absolute() or path != expected or path.is_symlink() or path.resolve() != expected:
+            raise ValueError("artifact path escapes the immutable report archive")
+        data = path.read_bytes()
+        if len(data) != size or hashlib.sha256(data).hexdigest() != digest:
+            raise ValueError("artifact bytes or hash differ")
+        full = json.loads(data)
+        if compact_discovery_report(full, artifact) != report:
+            raise ValueError("compact report differs from its full artifact")
+        for key in ("pin", "catalog_sha256", "index_sha256", "run_config"):
+            if full.get(key) != state.get(key):
+                raise ValueError("full report and checkpoint identity differ")
+        if full.get("complete") is not True or full.get("done") is not True:
+            raise ValueError("full report is unfinished")
+        return full
+    except (OSError, ValueError, TypeError, KeyError) as exc:
+        raise InitError(f"discovery full report artifact could not be verified: {exc}") from exc
+
+
 def _hash(value):
     return hashlib.sha256(_json(value).encode()).hexdigest()
 
@@ -765,13 +850,53 @@ def _chain(self):
         return super()._chain()
 
     def _cache_reusable(self, previous):
-        if previous.discovery.get("done") and self._validate_saved_archives(previous.discovery):
+        audited = deepcopy(previous)
+        if previous.discovery.get("done") and (self._validate_saved_archives(audited.discovery)
+                or previous.discovery.get("catalog_boundary_audit", {}).get("done")
+                and not audited.discovery.get("catalog_boundary_audit", {}).get("done")):
             raise InitError("completed discovery native archives are unavailable; restore the archive before reuse")
         if not previous.discovery.get("done") or self.retry_unfinished:
             return False
+        if previous.discovery.get("report_format") == COMPACT_REPORT_VERSION:
+            self._verify_report_record(previous)
+        elif previous.status != "published":
+            return False  # Re-render saved unpublished results through the official stage.
         # Published historical catalogs remain immutable. Unpublished previews
         # must acquire the additive audit before publication, reusing their scan.
-        return previous.status == "published" or self._boundary_record_ready(previous)
+        return previous.status == "published" or self._boundary_record_ready(audited)
+
+    def _verify_report_record(self, record, files=None):
+        from .init_stages import _dry_run_files
+        from .init_support import load_prepared
+        if files is None:
+            if record.pr.get("prepared"):
+                files = load_prepared(record.pr["prepared"])["files"]
+            elif record.pr.get("dry_run_dir"):
+                files = _dry_run_files(record)
+            else:
+                files = {path: self.rt.knowledge.show(record.kb_base_sha, path) for path in
+                         (record.discovery["report_path"], self._coverage_policy_path())}
+        text = files.get(record.discovery["report_path"])
+        catalog = files.get(self._coverage_policy_path())
+        if catalog is None:
+            catalog = self.rt.knowledge.show(record.kb_base_sha, self._coverage_policy_path())
+        if not isinstance(text, str) or hashlib.sha256(text.encode()).hexdigest() != record.discovery.get("report_sha256") \
+                or not isinstance(catalog, str) or hashlib.sha256(catalog.encode()).hexdigest() != record.discovery.get("catalog_sha256"):
+            raise InitError("discovery checkpoint and compact catalog/report bytes differ")
+        try:
+            report = json.loads(text)
+        except ValueError as exc:
+            raise InitError("discovery compact report is not JSON") from exc
+        if not isinstance(report, dict) or report.get("repo") != record.repo or report.get("pin") != record.pin:
+            raise InitError("discovery compact report identity differs")
+        return verify_full_discovery_report(self.rt.state_dir, report, record.discovery)
+
+    def _publish(self, changed):
+        try:
+            self._verify_report_record(self.record, changed)
+        except InitError as exc:
+            return self._blocked([str(exc)])
+        return super()._publish(changed)
 
     def _boundary_record_ready(self, previous):
         state = previous.discovery
@@ -811,6 +936,10 @@ def _resume_input_problems(self, previous, digest):
         if not self._boundary_record_ready(audited):
             return ["prepared discovery has no complete catalog boundary audit; preserve its publication and checkpoint "
                     "and audit a copy before creating a new publication"]
+        try:
+            self._verify_report_record(previous)
+        except InitError as exc:
+            return [str(exc)]
         return []
 
     def _init_identity(self):
@@ -1139,8 +1268,15 @@ def save():
                   "limits": "Declared inventory processed; not proof all repository capabilities were found. Tests not executed.",
                   "historical_denominator": len(seeds) * 7, "facet_denominator": len(features) * 7}
         report_path = f"eval/feature-discovery/{self.lifecycle.repo}-{self.record.pin[:12]}.json"
-        report_text = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
+        try:
+            if _report_archive_root(self.rt.state_dir).is_relative_to(Path(self.rt.knowledge.path).resolve()):
+                raise InitError("full discovery reports require a state directory outside the knowledge checkout")
+            artifact = write_full_discovery_report(self.rt.state_dir, report)
+        except (InitError, OSError) as exc:
+            return self._blocked([str(exc)])
+        report_text = json.dumps(compact_discovery_report(report, artifact), ensure_ascii=False, indent=2) + "\n"
         state.update(done=True, pin=self.record.pin, catalog_sha256=catalog_sha, report_path=report_path,
+                     report_format=COMPACT_REPORT_VERSION, full_artifact=artifact,
                      report_sha256=hashlib.sha256(report_text.encode()).hexdigest())
         other = {path: (original_text, rendered), report_path: (self.rt.knowledge.show(self._base_sha, report_path), report_text)}
 

test/test_kb_feature_discovery_report.py

@@ -0,0 +1,74 @@
+"""Compact publication preserves full evidence in an immutable external file."""
+from copy import deepcopy
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+
+from infermatrix_copilot.kb_service.init_feature_discovery import (
+    compact_discovery_report, write_full_discovery_report, verify_full_discovery_report,
+)
+from infermatrix_copilot.kb_service.init_support import InitError
+
+
+def full_report():
+    return {'schema_version': 1, 'repo': 'library', 'pin': 'a' * 40,
+            'complete': True, 'done': True, 'catalog_sha256': 'b' * 64,
+            'index_sha256': 'c' * 64, 'run_config': {'packet_chars': 96000},
+            'feature_ids': ['library'], 'features': [{'id': 'library', 'owner': 'core', 'title': 'Library'}],
+            'owner_requests': [], 'counts': {'unknown': 5000}, 'statistics': {'unknown': 5000},
+            'catalog_boundary_audit': {'done': True, 'identity': {'version': 'boundary-v1'},
+                                       'candidate_ids': ['proposal'] * 5000, 'counts': {'yes': 5000}},
+            'candidates': [{'id': f'proposal-{n}', 'description': 'evidence ' * 250,
+                            'reason': 'unknown ' * 250, 'generator_receipts': [{'trace_id': 'native'}],
+                            'evidence': [{'path': f'src/file-{n}.py', 'start': 1, 'end': 2}]} for n in range(5000)],
+            'unassociated_implementation_paths': [f'src/unmapped-{n}.py' for n in range(5000)],
+            'unassociated_entry_leads': [{'path': f'src/unmapped-{n}.py', 'symbols': ['entry']} for n in range(5000)],
+            'failures': [], 'scope_suggestions': [{'path': f'outside/file-{n}.py'} for n in range(5000)],
+            'candidate_limit_task_ids': [f'source:{n}' for n in range(500)],
+            'invalid_candidates': [{'reason': 'invalid citation'} for n in range(5000)]}
+
+
+def test_large_report_is_compact_and_full_bytes_are_preserved(tmp_path):
+    full = full_report()
+    artifact = write_full_discovery_report(tmp_path, full)
+    compact = compact_discovery_report(full, artifact)
+    assert len(json.dumps(compact, ensure_ascii=False).encode()) < 100_000
+    assert artifact['size_bytes'] > 10_000_000
+    assert compact['candidate_count'] == 5000 and 'candidates' not in compact
+    assert compact['catalog_boundary_audit']['candidate_count'] == 5000
+    assert 'candidate_ids' not in compact['catalog_boundary_audit']
+    assert compact['scope_suggestions']['count'] == 5000
+    state = {**{k: full[k] for k in ('pin', 'catalog_sha256', 'index_sha256', 'run_config')},
+             'report_format': compact['report_format'], 'full_artifact': artifact}
+    assert verify_full_discovery_report(tmp_path, compact, state) == full
+    raw = Path(artifact['path']).read_bytes()
+    assert hashlib.sha256(raw).hexdigest() == artifact['sha256']
+    assert write_full_discovery_report(tmp_path, full) == artifact
+    assert Path(artifact['path']).read_bytes() == raw
+
+
+@pytest.mark.parametrize('damage', ['escape', 'symlink', 'checkpoint_binding', 'summary', 'size', 'identity'])
+def test_full_report_binding_and_containment_refuse_tampering(tmp_path, damage):
+    full = full_report(); full['candidates'] = []
+    artifact = write_full_discovery_report(tmp_path, full)
+    compact = compact_discovery_report(full, artifact)
+    state = {**{k: full[k] for k in ('pin', 'catalog_sha256', 'index_sha256', 'run_config')},
+             'report_format': compact['report_format'], 'full_artifact': deepcopy(artifact)}
+    if damage == 'escape':
+        outside = tmp_path / 'elsewhere.json'; outside.write_bytes(Path(artifact['path']).read_bytes())
+        compact['full_artifact']['path'] = str(outside); state['full_artifact'] = deepcopy(compact['full_artifact'])
+    elif damage == 'symlink':
+        path = Path(artifact['path']); outside = tmp_path / 'elsewhere.json'
+        path.rename(outside); path.symlink_to(outside)
+    elif damage == 'checkpoint_binding':
+        state['full_artifact']['sha256'] = 'd' * 64
+    elif damage == 'summary':
+        compact['counts']['unknown'] = 0
+    elif damage == 'size':
+        compact['full_artifact']['size_bytes'] += 1; state['full_artifact'] = deepcopy(compact['full_artifact'])
+    else:
+        state['catalog_sha256'] = 'd' * 64
+    with pytest.raises(InitError, match='full report artifact'):
+        verify_full_discovery_report(tmp_path, compact, state)