diff --git a/depone/fixtures/code_health/tiered_mixed/expected-verdict.json b/depone/fixtures/code_health/tiered_mixed/expected-verdict.json index 00f5f78..1a4d108 100644 --- a/depone/fixtures/code_health/tiered_mixed/expected-verdict.json +++ b/depone/fixtures/code_health/tiered_mixed/expected-verdict.json @@ -1,5 +1,5 @@ { - "decision": "fail", + "decision": "pass", "error_codes": [ "ERR_HEALTH_GATE_VIOLATION", "ERR_HEALTH_GATE_VIOLATION" @@ -12,6 +12,8 @@ "tool": "black", "status": "pass", "enforcement": "block", + "evidence_substrate": "producer-transcribed", + "means": "producer-reported exit code; not bundle-bound", "blocks_handoff": false, "error_code": null, "evidence_path": null @@ -21,6 +23,8 @@ "tool": "ruff-c901", "status": "fail", "enforcement": "advisory", + "evidence_substrate": "producer-transcribed", + "means": "producer-reported exit code; not bundle-bound", "blocks_handoff": false, "error_code": "ERR_HEALTH_GATE_VIOLATION", "evidence_path": "health/complexity.exit" @@ -30,7 +34,9 @@ "tool": "import-linter", "status": "fail", "enforcement": "block", - "blocks_handoff": true, + "evidence_substrate": "producer-transcribed", + "means": "producer-reported exit code; not bundle-bound", + "blocks_handoff": false, "error_code": "ERR_HEALTH_GATE_VIOLATION", "evidence_path": "health/architecture.exit" } diff --git a/depone/verify/engine.py b/depone/verify/engine.py index 7a75982..c62d47f 100644 --- a/depone/verify/engine.py +++ b/depone/verify/engine.py @@ -19,6 +19,7 @@ from depone.verify.evidence_contract import ( EvidenceContractEntry, _read_evidence_contract, + _health_evidence_substrates, validate_advisory_provenance, validate_evidence_contract, ) @@ -144,6 +145,8 @@ class HealthAxisConformance: tool: str status: Literal["pass", "fail"] enforcement: Literal["block", "advisory"] + evidence_substrate: Literal["bound", "producer-transcribed"] + means: str blocks_handoff: bool error_code: str | None = None evidence_path: str | None = None @@ -234,6 +237,7 @@ def _health_entry_matches_gate( def _is_advisory_health_entry( contract: dict[str, Any] | None, entry: EvidenceContractEntry, + health_substrates: dict[str, str] | None = None, ) -> bool: if entry.code != "ERR_HEALTH_GATE_VIOLATION" or contract is None: return False @@ -243,23 +247,33 @@ def _is_advisory_health_entry( gates = directive.get("gates") if not isinstance(gates, list): return False - return any( - isinstance(gate, dict) - and gate.get("enforcement") == "advisory" - and _health_entry_matches_gate(entry, gate) - for gate in gates + matching_gate = next( + ( + gate + for gate in gates + if isinstance(gate, dict) and _health_entry_matches_gate(entry, gate) + ), + None, ) + if not matching_gate: + return False + if health_substrates is not None: + gate_id = entry.health_gate.get("gate") if entry.health_gate else None + if health_substrates.get(gate_id) == "producer-transcribed": + return True + return matching_gate.get("enforcement") == "advisory" def _blocking_evidence_contract_entries( contract: dict[str, Any] | None, evidence_contract: list[EvidenceContractEntry], + health_substrates: dict[str, str] | None = None, ) -> list[EvidenceContractEntry]: return [ entry for entry in evidence_contract if not _is_advisory_skill_routing_entry(contract, entry) - and not _is_advisory_health_entry(contract, entry) + and not _is_advisory_health_entry(contract, entry, health_substrates) ] @@ -416,6 +430,7 @@ def _policy_conformance( def _health_conformance( contract: dict[str, Any] | None, evidence_contract: list[EvidenceContractEntry], + health_substrates: dict[str, str] | None = None, ) -> HealthConformance | None: if contract is None: return None @@ -450,13 +465,29 @@ def _health_conformance( None, ) status: Literal["pass", "fail"] = "fail" if failure else "pass" + substrate: Literal["bound", "producer-transcribed"] = ( + "bound" + if health_substrates is not None + and health_substrates.get(gate_id) == "bound" + else "producer-transcribed" + ) axes.append( HealthAxisConformance( gate=gate_id, tool=tool, status=status, enforcement=enforcement, - blocks_handoff=status == "fail" and enforcement == "block", + evidence_substrate=substrate, + means=( + "bundle-subject-verified gate artifacts" + if substrate == "bound" + else "producer-reported exit code; not bundle-bound" + ), + blocks_handoff=( + status == "fail" + and enforcement == "block" + and substrate == "bound" + ), error_code=failure.code if failure else None, evidence_path=failure.evidence_path if failure else None, ) @@ -913,6 +944,11 @@ def run_verification( evidence, verified_signature_anchors=contract_signature_anchors, ) + health_substrates, _health_binding_errors = ( + _health_evidence_substrates(evidence, contract) + if contract is not None + else ({}, []) + ) evidence_contract_schema_version = _validated_evidence_contract_schema_version( contract, evidence_contract, @@ -983,7 +1019,9 @@ def run_verification( ) ) - if _blocking_evidence_contract_entries(contract, evidence_contract): + if _blocking_evidence_contract_entries( + contract, evidence_contract, health_substrates + ): any_refuted = True if any( entry.error_code @@ -1052,6 +1090,8 @@ def run_verification( review_signals=review_signals, role_capability_conformance=role_capability_conformance, policy_conformance=_policy_conformance(role_capability_conformance, contract), - health_conformance=_health_conformance(contract, evidence_contract), + health_conformance=_health_conformance( + contract, evidence_contract, health_substrates + ), verdict=overall, ) diff --git a/depone/verify/evidence_contract.py b/depone/verify/evidence_contract.py index 577e062..ce9b513 100644 --- a/depone/verify/evidence_contract.py +++ b/depone/verify/evidence_contract.py @@ -70,6 +70,13 @@ class EvidenceContractEntry: "ERR_ROLE_CAPABILITY_SKILL_ROUTING_VIOLATION" ) _ERR_HEALTH_GATE_VIOLATION = "ERR_HEALTH_GATE_VIOLATION" +_ERR_HEALTH_GATE_MANIFEST_NOT_BUNDLE_SUBJECT = ( + "ERR_HEALTH_GATE_MANIFEST_NOT_BUNDLE_SUBJECT" +) +_ERR_HEALTH_GATE_ARTIFACT_MISSING = "ERR_HEALTH_GATE_ARTIFACT_MISSING" +_ERR_HEALTH_GATE_ARTIFACT_DIGEST_MISMATCH = ( + "ERR_HEALTH_GATE_ARTIFACT_DIGEST_MISMATCH" +) _ERR_ROLE_CAPABILITY_OBSERVATION_UNBOUND = "ERR_ROLE_CAPABILITY_OBSERVATION_UNBOUND" _ERR_ROLE_CAPABILITY_OBSERVATION_DIGEST_MISMATCH = ( "ERR_ROLE_CAPABILITY_OBSERVATION_DIGEST_MISMATCH" @@ -466,6 +473,126 @@ def _raw_artifact_digest_matches(expected_digest: str, content: str) -> bool: return expected_digest == hashlib.sha256(content.encode("utf-8")).hexdigest() +def _health_evidence_substrates( + evidence: EvidenceContext, + contract: dict[str, Any], +) -> tuple[dict[str, str], list[EvidenceContractEntry]]: + directive = contract.get("code_health") + if not isinstance(directive, dict): + return {}, [] + gates = directive.get("gates") + if not isinstance(gates, list): + return {}, [] + substrates = { + gate["gate"]: "producer-transcribed" + for gate in gates + if isinstance(gate, dict) and isinstance(gate.get("gate"), str) + } + manifest_path = directive.get("manifest_path") + if not isinstance(manifest_path, str) or not manifest_path: + manifest_path = "health-gate-artifacts.json" + manifest_entry = _evidence_file_entry(evidence, manifest_path) + if manifest_entry is None: + return substrates, [] + + bundle_path = directive.get("bundle_path") + if not isinstance(bundle_path, str) or not bundle_path: + bundle_path = "bundle.json" + bundle_entry = _evidence_file_entry(evidence, bundle_path) + if bundle_entry is None: + return substrates, [ + EvidenceContractEntry( + code=_ERR_HEALTH_GATE_MANIFEST_NOT_BUNDLE_SUBJECT, + message=f"health manifest {manifest_path} is not a subject in {bundle_path}", + evidence_path=bundle_path, + ) + ] + bundle, invalid = _json_object(bundle_entry.content, bundle_path) + if invalid is not None or bundle is None: + return substrates, [ + EvidenceContractEntry( + code=_ERR_HEALTH_GATE_MANIFEST_NOT_BUNDLE_SUBJECT, + message=f"health manifest {manifest_path} is not a subject in {bundle_path}", + evidence_path=bundle_path, + ) + ] + verified_bundle, signature_error = _verified_role_capability_bundle( + evidence, bundle, bundle_path + ) + if signature_error is not None or verified_bundle is None: + return substrates, [ + signature_error + if signature_error is not None + else EvidenceContractEntry( + code=_ERR_HEALTH_GATE_MANIFEST_NOT_BUNDLE_SUBJECT, + message=f"health manifest {manifest_path} is not a verified bundle subject", + evidence_path=bundle_path, + ) + ] + bundle = verified_bundle + expected_manifest_digest = _subject_digest(bundle, manifest_path) + if expected_manifest_digest is None or not _raw_artifact_digest_matches( + expected_manifest_digest, manifest_entry.content + ): + return substrates, [ + EvidenceContractEntry( + code=_ERR_HEALTH_GATE_MANIFEST_NOT_BUNDLE_SUBJECT, + message=f"health manifest {manifest_path} is not a verified bundle subject", + evidence_path=bundle_path, + ) + ] + manifest, invalid = _json_object( + manifest_entry.content, + manifest_path, + _ERR_HEALTH_GATE_MANIFEST_NOT_BUNDLE_SUBJECT, + ) + if invalid is not None or manifest is None: + return substrates, [invalid] if invalid is not None else [] + manifest_gates = manifest.get("gates") + if not isinstance(manifest_gates, list): + return substrates, [ + EvidenceContractEntry( + code=_ERR_HEALTH_GATE_MANIFEST_NOT_BUNDLE_SUBJECT, + message=f"health manifest {manifest_path} must contain gates", + evidence_path=manifest_path, + ) + ] + for item in manifest_gates: + if not isinstance(item, dict): + continue + gate_id = item.get("gate") + if not isinstance(gate_id, str): + continue + for path_key, digest_key in ( + ("exit_code_path", "exit_code_sha256"), + ("log_path", "log_sha256"), + ): + path = item.get(path_key) + expected_digest = item.get(digest_key) + if not isinstance(path, str) or not isinstance(expected_digest, str): + continue + artifact = _evidence_file_entry(evidence, path) + if artifact is None: + return substrates, [ + EvidenceContractEntry( + code=_ERR_HEALTH_GATE_ARTIFACT_MISSING, + message=f"health gate artifact missing: {path}", + evidence_path=path, + ) + ] + if not _raw_artifact_digest_matches(expected_digest, artifact.content): + return substrates, [ + EvidenceContractEntry( + code=_ERR_HEALTH_GATE_ARTIFACT_DIGEST_MISMATCH, + message=f"health gate artifact digest mismatch: {path}", + evidence_path=path, + ) + ] + if gate_id in substrates: + substrates[gate_id] = "bound" + return substrates, [] + + def _canonical_decision_hash(decision: dict[str, Any]) -> str: return hashlib.sha256( json.dumps( @@ -799,7 +926,8 @@ def _validate_code_health( ) ] - results: list[EvidenceContractEntry] = [] + _substrates, binding_errors = _health_evidence_substrates(evidence, contract) + results: list[EvidenceContractEntry] = list(binding_errors) for index, gate in enumerate(gates): prefix = f"code_health.gates[{index}]" if not isinstance(gate, dict): diff --git a/scripts/revalidate_code_health.py b/scripts/revalidate_code_health.py index e8728b2..07fb0a4 100755 --- a/scripts/revalidate_code_health.py +++ b/scripts/revalidate_code_health.py @@ -63,7 +63,7 @@ def revalidate() -> dict[str, Any]: def self_test() -> None: actual = revalidate() wrong = dict(actual) - wrong["decision"] = "pass" + wrong["decision"] = "fail" try: _assert_result(actual, wrong) except AssertionError: diff --git a/tests/test_code_health_contract.py b/tests/test_code_health_contract.py index 6cac668..6f7bf9b 100644 --- a/tests/test_code_health_contract.py +++ b/tests/test_code_health_contract.py @@ -2,11 +2,14 @@ import hashlib import json +import tempfile import unittest from dataclasses import asdict from pathlib import Path from depone.verify.adapters.base import EvidenceContext, EvidenceFile +from depone.agent_fabric.evidence_substrate import wrap_statement_in_dsse +from depone.agent_fabric.sign import _generate_ed25519_keypair, sign_dsse_envelope from depone.verify.adapters.generic import read_evidence from depone.verify.engine import ( _health_conformance, @@ -228,6 +231,8 @@ def test_health_conformance_rolls_up_three_declared_gates(self) -> None: "tool": "black", "status": "pass", "enforcement": "block", + "evidence_substrate": "producer-transcribed", + "means": "producer-reported exit code; not bundle-bound", "blocks_handoff": False, "error_code": None, "evidence_path": None, @@ -237,6 +242,8 @@ def test_health_conformance_rolls_up_three_declared_gates(self) -> None: "tool": "ruff-c901", "status": "fail", "enforcement": "advisory", + "evidence_substrate": "producer-transcribed", + "means": "producer-reported exit code; not bundle-bound", "blocks_handoff": False, "error_code": "ERR_HEALTH_GATE_VIOLATION", "evidence_path": "health/complexity.exit", @@ -246,7 +253,9 @@ def test_health_conformance_rolls_up_three_declared_gates(self) -> None: "tool": "import-linter", "status": "fail", "enforcement": "block", - "blocks_handoff": True, + "evidence_substrate": "producer-transcribed", + "means": "producer-reported exit code; not bundle-bound", + "blocks_handoff": False, "error_code": "ERR_HEALTH_GATE_VIOLATION", "evidence_path": "health/architecture.exit", }, @@ -268,7 +277,7 @@ def test_advisory_health_failure_is_reported_without_blocking_decision(self) -> self.assertEqual(report.health_conformance.overall, "fail") self.assertFalse(report.health_conformance.axes[0].blocks_handoff) - def test_block_health_failure_refutes_decision(self) -> None: + def test_transcribed_block_health_failure_is_advisory(self) -> None: report = run_verification( _plan(), _evidence( @@ -277,10 +286,72 @@ def test_block_health_failure_refutes_decision(self) -> None: ), ) - self.assertEqual(report.decision, "fail") - self.assertEqual(report.verdict, "refuted") + self.assertEqual(report.decision, "pass") + self.assertEqual(report.verdict, "verified") self.assertEqual(report.health_conformance.overall, "fail") - self.assertTrue(report.health_conformance.axes[0].blocks_handoff) + self.assertFalse(report.health_conformance.axes[0].blocks_handoff) + + def test_bound_block_health_failure_honors_declared_enforcement(self) -> None: + gate = _gate("architecture", "import-linter", "block") + evidence = _evidence([gate], {"architecture": 1}) + contract = json.loads(evidence.files[0].content) + entries = validate_evidence_contract(evidence) + + conformance = _health_conformance( + contract, + entries, + {"architecture": "bound"}, + ) + + self.assertTrue(conformance.axes[0].blocks_handoff) + self.assertEqual(conformance.axes[0].evidence_substrate, "bound") + + def test_bound_manifest_tampered_exit_is_refuted(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + gate = _gate("architecture", "import-linter", "block") + evidence = _evidence([gate], {"architecture": 0}) + files = {entry.path: entry for entry in evidence.files} + manifest = { + "gates": [{ + "gate": "architecture", + "exit_code_path": "health/architecture.exit", + "exit_code_sha256": _sha("0\n"), + "log_path": "health/architecture.log", + "log_sha256": _sha("recorded gate output\n"), + }] + } + manifest_content = _file("health-gate-artifacts.json", manifest).content + statement = { + "subject": [{ + "name": "health-gate-artifacts.json", + "digest": {"sha256": _sha(manifest_content)}, + }] + } + private_key, public_key = _generate_ed25519_keypair(Path(tmp)) + bundle = { + "statement": statement, + "dsse_envelope": sign_dsse_envelope( + wrap_statement_in_dsse(statement), + str(private_key), + key_id="health-test-key", + ), + } + files["health-gate-artifacts.json"] = _file( + "health-gate-artifacts.json", manifest + ) + files["bundle.json"] = _file("bundle.json", bundle) + files["health/architecture.exit"] = _file( + "health/architecture.exit", "1\n" + ) + evidence.files = list(files.values()) + evidence.raw["trusted_observer_public_key_file"] = str(public_key) + + errors = validate_evidence_contract(evidence) + + self.assertIn( + "ERR_HEALTH_GATE_ARTIFACT_DIGEST_MISMATCH", + [error.code for error in errors], + ) def test_health_conformance_is_none_without_code_health_directive(self) -> None: evidence = _evidence([_gate("format", "black", "block")]) @@ -296,7 +367,7 @@ def test_committed_tiered_fixture_rederives_mixed_health_result(self) -> None: report = run_verification(_plan(), evidence) - self.assertEqual(report.decision, "fail") + self.assertEqual(report.decision, "pass") self.assertEqual( [entry.code for entry in report.evidence_contract], ["ERR_HEALTH_GATE_VIOLATION", "ERR_HEALTH_GATE_VIOLATION"], @@ -309,7 +380,7 @@ def test_committed_tiered_fixture_rederives_mixed_health_result(self) -> None: [ ("format", "pass", "block", False), ("complexity", "fail", "advisory", False), - ("architecture", "fail", "block", True), + ("architecture", "fail", "block", False), ], )