Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 8 additions & 2 deletions depone/fixtures/code_health/tiered_mixed/expected-verdict.json
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
{
"decision": "fail",
"decision": "pass",
"error_codes": [
"ERR_HEALTH_GATE_VIOLATION",
"ERR_HEALTH_GATE_VIOLATION"
Expand All @@ -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
Expand All @@ -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"
Expand All @@ -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"
}
Expand Down
58 changes: 49 additions & 9 deletions depone/verify/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
from depone.verify.evidence_contract import (
EvidenceContractEntry,
_read_evidence_contract,
_health_evidence_substrates,
validate_advisory_provenance,
validate_evidence_contract,
)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand All @@ -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)
]


Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
)
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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,
)
130 changes: 129 additions & 1 deletion depone/verify/evidence_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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):
Expand Down
2 changes: 1 addition & 1 deletion scripts/revalidate_code_health.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
Loading
Loading