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
31 changes: 31 additions & 0 deletions depone/fixtures/code_health/tiered_mixed/evidence-contract.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
{
"schema_version": "v111.code_health",
"code_health": {
"gates": [
{
"gate": "format",
"tool": "black",
"enforcement": "block",
"expected_exit_code": 0,
"exit_code_path": "health/format.exit",
"log_path": "health/format.log"
},
{
"gate": "complexity",
"tool": "ruff-c901",
"enforcement": "advisory",
"expected_exit_code": 0,
"exit_code_path": "health/complexity.exit",
"log_path": "health/complexity.log"
},
{
"gate": "architecture",
"tool": "import-linter",
"enforcement": "block",
"expected_exit_code": 0,
"exit_code_path": "health/architecture.exit",
"log_path": "health/architecture.log"
}
]
}
}
39 changes: 39 additions & 0 deletions depone/fixtures/code_health/tiered_mixed/expected-verdict.json
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
{
"decision": "fail",
"error_codes": [
"ERR_HEALTH_GATE_VIOLATION",
"ERR_HEALTH_GATE_VIOLATION"
],
"health_conformance": {
"overall": "fail",
"axes": [
{
"gate": "format",
"tool": "black",
"status": "pass",
"enforcement": "block",
"blocks_handoff": false,
"error_code": null,
"evidence_path": null
},
{
"gate": "complexity",
"tool": "ruff-c901",
"status": "fail",
"enforcement": "advisory",
"blocks_handoff": false,
"error_code": "ERR_HEALTH_GATE_VIOLATION",
"evidence_path": "health/complexity.exit"
},
{
"gate": "architecture",
"tool": "import-linter",
"status": "fail",
"enforcement": "block",
"blocks_handoff": true,
"error_code": "ERR_HEALTH_GATE_VIOLATION",
"evidence_path": "health/architecture.exit"
}
]
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
2
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
import-linter observed the declared block architecture gate failing
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
1
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
ruff-c901 observed the declared advisory complexity gate failing
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
0
1 change: 1 addition & 0 deletions depone/fixtures/code_health/tiered_mixed/health/format.log
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
black observed the declared format gate passing
108 changes: 108 additions & 0 deletions depone/verify/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -138,6 +138,23 @@ class PolicyConformance:
axes: list[PolicyAxisConformance] = field(default_factory=list)


@dataclass
class HealthAxisConformance:
gate: str
tool: str
status: Literal["pass", "fail"]
enforcement: Literal["block", "advisory"]
blocks_handoff: bool
error_code: str | None = None
evidence_path: str | None = None


@dataclass
class HealthConformance:
overall: Literal["pass", "fail"]
axes: list[HealthAxisConformance] = field(default_factory=list)


@dataclass
class VerificationReport:
"""Verification result.
Expand Down Expand Up @@ -165,6 +182,7 @@ class VerificationReport:
default_factory=list
)
policy_conformance: PolicyConformance | None = None
health_conformance: HealthConformance | None = None
verdict: Literal["verified", "refuted", "insufficient-evidence"] = "verified"


Expand Down Expand Up @@ -198,6 +216,39 @@ def _is_advisory_skill_routing_entry(
return isinstance(directive, dict) and directive.get("enforcement") == "advisory"


def _health_entry_matches_gate(
entry: EvidenceContractEntry,
gate: dict[str, Any],
) -> bool:
return (
entry.code == "ERR_HEALTH_GATE_VIOLATION"
and gate.get("exit_code_path") == entry.evidence_path
and f"gate={gate.get('gate')!r}" in entry.message
and f"tool={gate.get('tool')!r}" in entry.message
and f"enforcement={gate.get('enforcement')!r}" in entry.message
)


def _is_advisory_health_entry(
contract: dict[str, Any] | None,
entry: EvidenceContractEntry,
) -> bool:
if entry.code != "ERR_HEALTH_GATE_VIOLATION" or contract is None:
return False
directive = contract.get("code_health")
if not isinstance(directive, dict):
return False
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
)


def _blocking_evidence_contract_entries(
contract: dict[str, Any] | None,
evidence_contract: list[EvidenceContractEntry],
Expand All @@ -206,6 +257,7 @@ def _blocking_evidence_contract_entries(
entry
for entry in evidence_contract
if not _is_advisory_skill_routing_entry(contract, entry)
and not _is_advisory_health_entry(contract, entry)
]


Expand Down Expand Up @@ -359,6 +411,61 @@ def _policy_conformance(
return PolicyConformance(overall=overall, axes=axes)


def _health_conformance(
contract: dict[str, Any] | None,
evidence_contract: list[EvidenceContractEntry],
) -> HealthConformance | None:
if contract is None:
return None
directive = contract.get("code_health")
if not isinstance(directive, dict):
return None
gates = directive.get("gates")
if not isinstance(gates, list):
return None

axes: list[HealthAxisConformance] = []
for gate in gates:
if not isinstance(gate, dict):
continue
gate_id = gate.get("gate")
tool = gate.get("tool")
enforcement = gate.get("enforcement")
exit_code_path = gate.get("exit_code_path")
if (
not isinstance(gate_id, str)
or not isinstance(tool, str)
or enforcement not in {"block", "advisory"}
or not isinstance(exit_code_path, str)
):
continue
failure = next(
(
entry
for entry in evidence_contract
if _health_entry_matches_gate(entry, gate)
),
None,
)
status: Literal["pass", "fail"] = "fail" if failure else "pass"
axes.append(
HealthAxisConformance(
gate=gate_id,
tool=tool,
status=status,
enforcement=enforcement,
blocks_handoff=status == "fail" and enforcement == "block",
error_code=failure.code if failure else None,
evidence_path=failure.evidence_path if failure else None,
)
)

overall: Literal["pass", "fail"] = (
"fail" if any(axis.status == "fail" for axis in axes) else "pass"
)
return HealthConformance(overall=overall, axes=axes)


def _resolve_handoff_path(
handoff: dict[str, Any],
evidence_map: dict[str, Any],
Expand Down Expand Up @@ -943,5 +1050,6 @@ 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),
verdict=overall,
)
107 changes: 106 additions & 1 deletion depone/verify/evidence_contract.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@ class EvidenceContractEntry:
_ROLE_CAPABILITY_SKILL_ROUTING_CONTRACT_SCHEMA_VERSION = (
"v110.role_capability_skill_routing"
)
_CODE_HEALTH_CONTRACT_SCHEMA_VERSION = "v111.code_health"
_OBSERVED_TOUCHED_FILES_FILENAME = "observed-touched-files.txt"
_OBSERVED_SKILLS_FILENAME = "observed-skills.txt"
# Deprecated compatibility alias for evidence sealed before the honest rename.
Expand Down Expand Up @@ -67,6 +68,7 @@ class EvidenceContractEntry:
_ERR_ROLE_CAPABILITY_SKILL_ROUTING_VIOLATION = (
"ERR_ROLE_CAPABILITY_SKILL_ROUTING_VIOLATION"
)
_ERR_HEALTH_GATE_VIOLATION = "ERR_HEALTH_GATE_VIOLATION"
_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 @@ -218,6 +220,8 @@ def _has_enforcement_directive(contract: dict[str, Any]) -> bool:
return True
if isinstance(contract.get("role_capability_skill_routing"), dict):
return True
if "code_health" in contract:
return True
if isinstance(contract.get("advisory_provenance"), dict):
return True
return contract.get("forbid_test_weakening") is True and _has_non_empty_str_list(
Expand All @@ -238,6 +242,7 @@ def _validate_contract_semantics(
_ADVISORY_PROVENANCE_EXECUTED_RED_CONTRACT_SCHEMA_VERSION,
_ROLE_CAPABILITY_BOUND_OBSERVATION_CONTRACT_SCHEMA_VERSION,
_ROLE_CAPABILITY_SKILL_ROUTING_CONTRACT_SCHEMA_VERSION,
_CODE_HEALTH_CONTRACT_SCHEMA_VERSION,
}:
return EvidenceContractEntry(
code=_ERR_CONTRACT_INVALID,
Expand All @@ -249,7 +254,8 @@ def _validate_contract_semantics(
f"{_ADVISORY_PROVENANCE_CONTRACT_SCHEMA_VERSION!r} or "
f"{_ADVISORY_PROVENANCE_EXECUTED_RED_CONTRACT_SCHEMA_VERSION!r} or "
f"{_ROLE_CAPABILITY_BOUND_OBSERVATION_CONTRACT_SCHEMA_VERSION!r} or "
f"{_ROLE_CAPABILITY_SKILL_ROUTING_CONTRACT_SCHEMA_VERSION!r}"
f"{_ROLE_CAPABILITY_SKILL_ROUTING_CONTRACT_SCHEMA_VERSION!r} or "
f"{_CODE_HEALTH_CONTRACT_SCHEMA_VERSION!r}"
),
evidence_path=_EVIDENCE_CONTRACT_FILENAME,
)
Expand Down Expand Up @@ -303,6 +309,18 @@ def _validate_contract_semantics(
),
evidence_path=_EVIDENCE_CONTRACT_FILENAME,
)
if (
"code_health" in contract
and schema_version != _CODE_HEALTH_CONTRACT_SCHEMA_VERSION
):
return EvidenceContractEntry(
code=_ERR_CONTRACT_INVALID,
message=(
"code_health requires schema_version "
f"{_CODE_HEALTH_CONTRACT_SCHEMA_VERSION!r}"
),
evidence_path=_EVIDENCE_CONTRACT_FILENAME,
)
if isinstance(contract.get("advisory_provenance"), dict) and schema_version not in {
_ADVISORY_PROVENANCE_CONTRACT_SCHEMA_VERSION,
_ADVISORY_PROVENANCE_EXECUTED_RED_CONTRACT_SCHEMA_VERSION,
Expand Down Expand Up @@ -754,6 +772,91 @@ def _validate_role_capability_skill_routing(
return []


def _validate_code_health(
evidence: EvidenceContext,
contract: dict[str, Any],
) -> list[EvidenceContractEntry]:
if "code_health" not in contract:
return []
directive = contract.get("code_health")
if not isinstance(directive, dict):
return [
EvidenceContractEntry(
code=_ERR_CONTRACT_INVALID,
message="code_health must be an object",
evidence_path=_EVIDENCE_CONTRACT_FILENAME,
)
]

gates = directive.get("gates")
if not isinstance(gates, list) or not gates:
return [
EvidenceContractEntry(
code=_ERR_CONTRACT_INVALID,
message="code_health.gates must be a non-empty list",
evidence_path=_EVIDENCE_CONTRACT_FILENAME,
)
]

results: list[EvidenceContractEntry] = []
for index, gate in enumerate(gates):
prefix = f"code_health.gates[{index}]"
if not isinstance(gate, dict):
return [
EvidenceContractEntry(
code=_ERR_CONTRACT_INVALID,
message=f"{prefix} must be an object",
evidence_path=_EVIDENCE_CONTRACT_FILENAME,
)
]
for key in ("gate", "tool", "exit_code_path", "log_path"):
if not isinstance(gate.get(key), str) or not gate[key]:
return [
EvidenceContractEntry(
code=_ERR_CONTRACT_INVALID,
message=f"{prefix}.{key} must be a non-empty string",
evidence_path=_EVIDENCE_CONTRACT_FILENAME,
)
]
enforcement = gate.get("enforcement")
if enforcement not in {"block", "advisory"}:
return [
EvidenceContractEntry(
code=_ERR_CONTRACT_INVALID,
message=(
f"{prefix}.enforcement must be 'block' or 'advisory'"
),
evidence_path=_EVIDENCE_CONTRACT_FILENAME,
)
]
expected_exit_code = gate.get("expected_exit_code")
if type(expected_exit_code) is not int:
return [
EvidenceContractEntry(
code=_ERR_CONTRACT_INVALID,
message=f"{prefix}.expected_exit_code must be an integer",
evidence_path=_EVIDENCE_CONTRACT_FILENAME,
)
]

exit_code_path = gate["exit_code_path"]
actual_exit_code = _read_exit_code(evidence, exit_code_path)
if actual_exit_code != expected_exit_code:
results.append(
EvidenceContractEntry(
code=_ERR_HEALTH_GATE_VIOLATION,
message=(
"code health gate exit code mismatch: "
f"gate={gate['gate']!r}, tool={gate['tool']!r}, "
f"enforcement={enforcement!r}, "
f"expected={expected_exit_code}, got={actual_exit_code}"
),
evidence_path=exit_code_path,
)
)
return results


def _load_bound_run_intent(
evidence: EvidenceContext,
run_intent_path: str,
Expand Down Expand Up @@ -1757,6 +1860,8 @@ def validate_evidence_contract(
verified_signature_anchors=verified_signature_anchors,
):
_append_unique_entry(results, entry)
for entry in _validate_code_health(evidence, contract):
_append_unique_entry(results, entry)

test_patterns = _as_str_list(contract.get("test_file_patterns"))
forbidden_test_files = set(_as_str_list(contract.get("forbidden_test_files")))
Expand Down
Loading
Loading