diff --git a/src/infermatrix_copilot/kb_service/init_feature_discovery.py b/src/infermatrix_copilot/kb_service/init_feature_discovery.py index c285f401..3bb3b49c 100644 --- a/src/infermatrix_copilot/kb_service/init_feature_discovery.py +++ b/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)} diff --git a/test/test_kb_feature_discovery_report.py b/test/test_kb_feature_discovery_report.py new file mode 100644 index 00000000..797ac382 --- /dev/null +++ b/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) diff --git a/test/test_kb_feature_discovery_stage.py b/test/test_kb_feature_discovery_stage.py index be0cefdf..101c3d47 100644 --- a/test/test_kb_feature_discovery_stage.py +++ b/test/test_kb_feature_discovery_stage.py @@ -97,6 +97,16 @@ def _report(record): return files, json.loads(files[path]), path +def _full_report(record): + _, report, _ = _report(record) + artifact = report['full_artifact'] + data = Path(artifact['path']).read_bytes() + assert artifact == record.discovery['full_artifact'] + assert len(data) == artifact['size_bytes'] + assert hashlib.sha256(data).hexdigest() == artifact['sha256'] + return json.loads(data) + + def _policy(): return {"schema_version": 1, "required": False, "semantic_depth": {"per_facet_gt": 0.92}, "core": {"roots": ["pkg/"], "exclude": ["*/tests/*"], "target": 0.97, "suffixes": [".py"]}, @@ -190,7 +200,7 @@ def test_effective_native_reasoning_is_bound_to_report_and_both_archives(world, _, report, _ = _report(record) assert report['run_config'] == record.discovery['run_config'] == { 'packet_chars': 96000, 'zcode_start_interval_s': 10.0, 'zcode_reasoning_level': 'low'} - for candidate in report['candidates']: + for candidate in _full_report(record)['candidates']: for proof in candidate['generator_receipts']: archived = store.get(proof['trace_id']) assert archived['model']['effort'] == '' @@ -215,7 +225,7 @@ def test_zcode_role_effort_does_not_mislabel_effective_native_configuration(worl assert record.status == 'dry_run', record.problems _, report, _ = _report(record) assert report['run_config']['zcode_reasoning_level'] == 'max' - for candidate in report['candidates']: + for candidate in _full_report(record)['candidates']: for proof in candidate['generator_receipts']: model = store.get(proof['trace_id'])['model'] assert model['effort'] == 'low' @@ -256,7 +266,7 @@ def test_first_source_only_catalog_has_native_review_and_no_knowledge_rewrite(wo assert report["complete"] and report["done"] and report["feature_ids"] == ["engine-step"] assert report["historical_denominator"] == 0 and report["facet_denominator"] == 7 assert report["catalog_sha256"] == hashlib.sha256(files["adapters/toy/knowledge-coverage.yaml"].encode()).hexdigest() - candidate = report["candidates"][0] + candidate = _full_report(record)["candidates"][0] assert candidate["status"] == "accepted" receipts = candidate["generator_receipts"] + [candidate["judge_receipt"]] for receipt in receipts: @@ -438,7 +448,7 @@ def test_completed_empty_shard_requires_native_archive_before_cache_reuse(world) (store.root/'blobs'/digest[:2]/f'{digest}.gz').unlink() with pytest.raises(InitError,match='archives'): stage._cache_reusable(record) - assert task['status']=='unknown' + assert task['status']=='complete' # Cache refusal never rewrites retained proofs. @pytest.mark.parametrize('tamper',['decision','candidate']) @@ -460,3 +470,104 @@ def test_restored_approval_binds_native_verdict_and_input_candidate(world,tamper assert 'engine-step' in failures and checked['judge_receipt'] assert state['reviews']['engine-step']['supported']=='unsure' assert 'native approval binding' in row['invalid_reason'] + + +@pytest.mark.parametrize('damage', ['missing', 'tampered']) +def test_cached_compact_report_requires_original_external_artifact(world, damage): + from infermatrix_copilot.kb_service.init_support import InitRecord + rt, lifecycle, script, _ = _setup(world, policy=_policy(), script=NativeScript(candidates=False)) + first = _run(rt, lifecycle) + assert first.status == 'dry_run', first.problems + files, compact, _ = _report(first) + assert 'adapters/toy/knowledge-coverage.yaml' not in files # Unchanged catalog comes from base. + path = Path(compact['full_artifact']['path']); original = path.read_bytes() + checkpoint = InitRecord.path(rt.state_dir, lifecycle.repo, 'feature-discovery') + checkpoint_bytes = checkpoint.read_bytes(); calls = len(script.calls) + if damage == 'missing': path.unlink() + else: path.write_bytes(original + b'corruption') + with pytest.raises(InitError, match='full report artifact'): + _run(rt, lifecycle) + assert checkpoint.read_bytes() == checkpoint_bytes and len(script.calls) == calls + path.write_bytes(original) + restored = _run(rt, lifecycle) + assert restored == first and len(script.calls) == calls + + +@pytest.mark.parametrize('damage', ['missing', 'tampered']) +def test_prepared_compact_artifact_blocks_publication_until_exact_restore(world, damage): + from infermatrix_copilot.kb_service.init_support import load_prepared + rt, lifecycle, script, _ = _setup(world) + rt.environ.update(ALLOW_PUSH='1', ALLOW_POST='1', KB_INIT_GIT_AUTHOR='t ') + gh = FakeGh(fail_create=1); rt.gh_run = gh + def publish(): + return run_stage(rt, lifecycle, 'feature-discovery', dry_run=False, from_existing=True, + unlimited_subscription=True) + first = publish() + assert first.status == 'blocked' and first.pr.get('prepared'), first.problems + artifact = Path(first.discovery['full_artifact']['path']); original = artifact.read_bytes() + prepared = Path(first.pr['prepared']); prepared_bytes = prepared.read_bytes() + calls, pushes = len(script.calls), list(gh.pushed) + if damage == 'missing': artifact.unlink() + else: artifact.write_bytes(original + b'corruption') + blocked = publish() + assert blocked.status == 'blocked' and any('full report artifact' in p for p in blocked.problems) + assert prepared.read_bytes() == prepared_bytes and len(script.calls) == calls and gh.pushed == pushes + artifact.write_bytes(original) + restored = publish() + assert restored.status == 'published', restored.problems + assert prepared.read_bytes() == prepared_bytes and len(script.calls) == calls and gh.pushed == pushes + compact = json.loads(load_prepared(prepared)['files'][restored.discovery['report_path']]) + assert compact['full_artifact'] == restored.discovery['full_artifact'] + + +def test_completed_legacy_preview_rerenders_without_new_native_calls(world, monkeypatch): + from infermatrix_copilot.kb_service.init_support import InitRecord + rt, lifecycle, script, _ = _setup(world) + first = _run(rt, lifecycle) + assert first.status == 'dry_run' + _, compact, path = _report(first) + full = _full_report(first) + # Represent a genuine old-format completed preview, retaining all native + # scan, primary and boundary receipts; this is not a prepared publication. + old_text = json.dumps(full, ensure_ascii=False, indent=2) + '\n' + (Path(first.pr['dry_run_dir']) / 'tree' / path).write_text(old_text) + first.discovery.pop('report_format'); first.discovery.pop('full_artifact') + first.discovery['report_sha256'] = hashlib.sha256(old_text.encode()).hexdigest() + first.save(rt.state_dir) + receipts = json.dumps(first.discovery['reviews'], sort_keys=True) + calls = len(script.calls) + monkeypatch.setattr(script, 'complete', lambda **kwargs: pytest.fail('legacy report rerender called a model')) + restored = _run(rt, lifecycle) + assert restored.status == 'dry_run', restored.problems + assert len(script.calls) == calls + assert json.dumps(restored.discovery['reviews'], sort_keys=True) == receipts + assert _report(restored)[1] == compact + assert _full_report(restored) == full + assert InitRecord.load(rt.state_dir, lifecycle.repo, 'feature-discovery').discovery['full_artifact'] == compact['full_artifact'] + + +def test_legacy_published_report_is_immutable_without_compact_marker(world, monkeypatch): + rt, lifecycle, script, _ = _setup(world) + first = _run(rt, lifecycle) + first.status = 'published' + first.discovery.pop('report_format'); first.discovery.pop('full_artifact') + first.save(rt.state_dir) + calls = len(script.calls) + monkeypatch.setattr(script, 'complete', lambda **kwargs: pytest.fail('published history called a model')) + assert _run(rt, lifecycle) == first and len(script.calls) == calls + + +def test_cached_missing_boundary_archive_preserves_proofs_and_blocks_resampling(world): + from infermatrix_copilot.kb_service.init_support import InitRecord + rt, lifecycle, script, store = _setup(world) + first = _run(rt, lifecycle) + checked = first.discovery['boundary_reviews']['engine-step'] + archive = store.get(checked['judge_receipt']['trace_id']) + # Responses may share a blob with the primary review; remove only the + # distinct boundary attempt so primary validation still succeeds. + (store.root / 'attempts' / archive['result']['native_attempt_id'] / 'attempt.json').unlink() + path = InitRecord.path(rt.state_dir, lifecycle.repo, 'feature-discovery') + original = path.read_bytes(); calls = len(script.calls) + with pytest.raises(InitError, match='archives'): + _run(rt, lifecycle) + assert path.read_bytes() == original and len(script.calls) == calls