wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 3 Packages Projects Releases Wiki Activity

fix(ai): separate runtime contracts from deploy SHA churn

Browse Source
This commit is contained in:
ogt
2026-07-11 11:34:24 +08:00
parent 478be85854
commit cf23e2b230
5 changed files with 316 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

125
apps/api/src/services/ai_agent_autonomous_runtime_control.py
View File

@@ -4496,6 +4496,72 @@ def _runtime_truthy(value: Any) -> bool:
return str(value or "").strip().lower() == "true"
def _build_runtime_source_compatibility(
*,
runtime: Mapping[str, Any],
source_boundary: Mapping[str, Any],
runtime_schema_version: str,
source_boundary_schema_version: str,
runtime_source_sha: str,
deployed_source_sha: str,
) -> dict[str, Any]:
"""Separate runtime contract proof from exact deploy-SHA freshness.
A newer unrelated deployment must not force another production apply just
to refresh a receipt SHA. The live runtime chain still has to satisfy the
versioned runtime contract, while the currently deployed source must pass
its independently verified source boundary.
"""
exact_sha_match = bool(
runtime_source_sha
and deployed_source_sha
and runtime_source_sha == deployed_source_sha
)
runtime_contract_verified = bool(
runtime.get("schema_version") == runtime_schema_version
and runtime.get("runtime_evidence_ready") is True
and runtime_source_sha
)
deployed_source_boundary_verified = bool(
source_boundary.get("schema_version") == source_boundary_schema_version
and source_boundary.get("status") == "verified_ready"
and source_boundary.get("source_boundary_verified") is True
and source_boundary.get("source_sha_matches_deployment") is True
and not list(source_boundary.get("active_blockers") or [])
and deployed_source_sha
)
compatible = bool(
runtime_contract_verified and deployed_source_boundary_verified
)
return {
"schema_version": "awoooi_runtime_source_compatibility_v1",
"status": (
"exact_sha_match"
if compatible and exact_sha_match
else "contract_compatible"
if compatible
else "unverified"
),
"runtime_contract_schema_version": runtime_schema_version,
"deployed_source_boundary_schema_version": (
source_boundary_schema_version
),
"runtime_source_sha": runtime_source_sha or None,
"deployed_source_sha": deployed_source_sha or None,
"exact_sha_match": exact_sha_match,
"runtime_contract_verified": runtime_contract_verified,
"deployed_source_boundary_verified": (
deployed_source_boundary_verified
),
"runtime_contract_compatible_with_deployed_source_boundary": (
compatible
),
"completion_requires_exact_sha_match": False,
"writes_on_read": False,
}
def _build_autonomous_single_writer_runtime_readback(
*,
operation_latest_rows: Iterable[Mapping[str, Any] | Any],
@@ -6695,15 +6761,27 @@ def _attach_autonomous_single_writer_readback(
boundary_readback.get("deployed_source_sha") or ""
).lower()
router_source_sha = str(runtime.get("router_source_sha") or "").lower()
source_compatibility = _build_runtime_source_compatibility(
runtime=runtime,
source_boundary=source_boundary,
runtime_schema_version="awoooi_autonomous_single_writer_runtime_v1",
source_boundary_schema_version=(
"awoooi_autonomous_single_writer_source_boundary_v1"
),
runtime_source_sha=router_source_sha,
deployed_source_sha=deployed_source_sha,
)
controls = {
"strict_runtime_same_run_closed": bool(
strict_runtime.get("closed") is True
and runtime_run_id
and runtime_run_id == strict_run_id
),
"router_source_sha_matches_deployment": bool(
deployed_source_sha
and router_source_sha == deployed_source_sha
"runtime_contract_compatible_with_deployed_source_boundary": bool(
source_compatibility.get(
"runtime_contract_compatible_with_deployed_source_boundary"
)
is True
),
**{
str(control_id): verified is True
@@ -6738,6 +6816,7 @@ def _attach_autonomous_single_writer_readback(
"apply_op_id": runtime.get("apply_op_id"),
"source_receipt_ref": source_boundary.get("receipt_ref"),
"source_verifier": source_boundary.get("verifier"),
"source_compatibility": source_compatibility,
"controls": controls,
"active_blockers": blockers,
"writes_on_read": False,
@@ -6780,15 +6859,29 @@ def _attach_independent_post_verifier_readback(
verifier_source_sha = str(
runtime.get("verifier_source_sha") or ""
).lower()
source_compatibility = _build_runtime_source_compatibility(
runtime=runtime,
source_boundary=source_boundary,
runtime_schema_version=(
"awoooi_independent_post_verifier_runtime_v1"
),
source_boundary_schema_version=(
"awoooi_independent_post_verifier_source_boundary_v1"
),
runtime_source_sha=verifier_source_sha,
deployed_source_sha=deployed_source_sha,
)
controls = {
"strict_runtime_same_run_closed": bool(
strict_runtime.get("closed") is True
and runtime_run_id
and runtime_run_id == strict_run_id
),
"verifier_source_sha_matches_deployment": bool(
deployed_source_sha
and verifier_source_sha == deployed_source_sha
"runtime_contract_compatible_with_deployed_source_boundary": bool(
source_compatibility.get(
"runtime_contract_compatible_with_deployed_source_boundary"
)
is True
),
**{
str(control_id): verified is True
@@ -6831,6 +6924,7 @@ def _attach_independent_post_verifier_readback(
),
"source_receipt_ref": source_boundary.get("receipt_ref"),
"source_verifier": source_boundary.get("verifier"),
"source_compatibility": source_compatibility,
"controls": controls,
"active_blockers": blockers,
"writes_on_read": False,
@@ -6881,15 +6975,27 @@ def _attach_canonical_learning_readback(
writer_source_sha = str(
runtime.get("writer_source_sha") or ""
).lower()
source_compatibility = _build_runtime_source_compatibility(
runtime=runtime,
source_boundary=source_boundary,
runtime_schema_version="awoooi_canonical_learning_runtime_v1",
source_boundary_schema_version=(
"awoooi_canonical_learning_source_boundary_v1"
),
runtime_source_sha=writer_source_sha,
deployed_source_sha=deployed_source_sha,
)
controls = {
"strict_runtime_same_run_closed": bool(
strict_runtime.get("closed") is True
and runtime_run_id
and runtime_run_id == strict_run_id
),
"writer_source_sha_matches_deployment": bool(
deployed_source_sha
and writer_source_sha == deployed_source_sha
"runtime_contract_compatible_with_deployed_source_boundary": bool(
source_compatibility.get(
"runtime_contract_compatible_with_deployed_source_boundary"
)
is True
),
**{
str(control_id): verified is True
@@ -6930,6 +7036,7 @@ def _attach_canonical_learning_readback(
),
"source_receipt_ref": source_boundary.get("receipt_ref"),
"source_verifier": source_boundary.get("verifier"),
"source_compatibility": source_compatibility,
"controls": controls,
"active_blockers": blockers,
"writes_on_read": False,

39
apps/api/src/services/awoooi_priority_work_order_readback.py
View File

@@ -273,7 +273,7 @@ _AI_AUTOMATION_PROGRAM_DEPENDENCIES = {
}
_AI_SINGLE_WRITER_REQUIRED_CONTROL_IDS = (
"strict_runtime_same_run_closed",
"router_source_sha_matches_deployment",
"runtime_contract_compatible_with_deployed_source_boundary",
"controlled_queue_decision_path",
"single_writer_executor",
"candidate_idempotency_key_present",
@@ -293,7 +293,7 @@ _AI_SINGLE_WRITER_REQUIRED_CONTROL_IDS = (
)
_AI_INDEPENDENT_POST_VERIFIER_REQUIRED_CONTROL_IDS = (
"strict_runtime_same_run_closed",
"verifier_source_sha_matches_deployment",
"runtime_contract_compatible_with_deployed_source_boundary",
"same_run_verifier_receipt",
"verification_result_success",
"asset_post_verifier_schema",
@@ -314,7 +314,7 @@ _AI_INDEPENDENT_POST_VERIFIER_REQUIRED_CONTROL_IDS = (
)
_AI_CANONICAL_LEARNING_REQUIRED_CONTROL_IDS = (
"strict_runtime_same_run_closed",
"writer_source_sha_matches_deployment",
"runtime_contract_compatible_with_deployed_source_boundary",
"same_run_learning_operation",
"learning_operation_success",
"learning_parent_apply_matches",
@@ -7502,6 +7502,9 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None:
"controls": single_writer_controls,
"missing": list(dict.fromkeys(single_writer_blockers)),
"runtime_closed": single_writer_closed,
"source_compatibility": _dict(
summary.get("ai_agent_single_writer_source_compatibility")
),
}
if single_writer_closed and all(single_writer_controls.values()):
single_writer_work_item["status"] = "done"
@@ -7529,6 +7532,9 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None:
summary.get("ai_agent_single_writer_apply_op_id") or ""
),
"single_writer_executor": "awoooi-ansible-executor-broker",
"source_compatibility": _dict(
summary.get("ai_agent_single_writer_source_compatibility")
),
}
single_writer_work_item["runtime_evidence_refs"] = [
"automation-run:"
@@ -7580,6 +7586,11 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None:
"controls": post_verifier_controls,
"missing": list(dict.fromkeys(post_verifier_blockers)),
"runtime_closed": post_verifier_closed,
"source_compatibility": _dict(
summary.get(
"ai_agent_independent_post_verifier_source_compatibility"
)
),
}
if (
single_writer_work_item["status"] == "done"
@@ -7634,6 +7645,11 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None:
)
),
"verifier": "asset_specific_read_only_host_postconditions",
"source_compatibility": _dict(
summary.get(
"ai_agent_independent_post_verifier_source_compatibility"
)
),
}
post_verifier_work_item["runtime_evidence_refs"] = [
"automation-run:"
@@ -7694,6 +7710,9 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None:
"controls": canonical_learning_controls,
"missing": list(dict.fromkeys(canonical_learning_blockers)),
"runtime_closed": canonical_learning_closed,
"source_compatibility": _dict(
summary.get("ai_agent_canonical_learning_source_compatibility")
),
}
if (
post_verifier_work_item["status"] == "done"
@@ -7753,6 +7772,9 @@ def _apply_ai_automation_program_ledger(payload: dict[str, Any]) -> None:
)
or ""
),
"source_compatibility": _dict(
summary.get("ai_agent_canonical_learning_source_compatibility")
),
}
canonical_learning_work_item["runtime_evidence_refs"] = [
"automation-run:"
@@ -8287,6 +8309,9 @@ def apply_ai_automation_live_closure_readbacks(
)
summary["ai_agent_single_writer_runtime_closed"] = single_writer_closed
summary["ai_agent_single_writer_controls"] = single_writer_controls
summary["ai_agent_single_writer_source_compatibility"] = _dict(
autonomous_single_writer.get("source_compatibility")
)
summary["ai_agent_single_writer_active_blockers"] = list(
autonomous_single_writer.get("active_blockers") or []
)
@@ -8336,6 +8361,11 @@ def apply_ai_automation_live_closure_readbacks(
summary["ai_agent_independent_post_verifier_controls"] = (
independent_post_verifier_controls
)
summary[
"ai_agent_independent_post_verifier_source_compatibility"
] = _dict(
autonomous_independent_post_verifier.get("source_compatibility")
)
summary["ai_agent_independent_post_verifier_active_blockers"] = list(
autonomous_independent_post_verifier.get("active_blockers") or []
)
@@ -8396,6 +8426,9 @@ def apply_ai_automation_live_closure_readbacks(
summary["ai_agent_canonical_learning_controls"] = (
canonical_learning_controls
)
summary["ai_agent_canonical_learning_source_compatibility"] = _dict(
autonomous_canonical_learning.get("source_compatibility")
)
summary["ai_agent_canonical_learning_active_blockers"] = list(
autonomous_canonical_learning.get("active_blockers") or []
)