feat(ai): verify canonical learning writeback

apps/api/src/services/executor_trust_boundary_readback.py

@@ -50,6 +50,14 @@ _INDEPENDENT_POST_VERIFIER_REQUIRED_TRUE_FIELDS = (
     "verifier_failure_retry_first",
     "post_verifier_raw_output_not_persisted",
 )
+_CANONICAL_LEARNING_REQUIRED_TRUE_FIELDS = (
+    "canonical_playbook_resolver_complete",
+    "km_row_readback_required",
+    "playbook_row_readback_required",
+    "learning_failure_fail_closed",
+    "trust_failure_fail_closed",
+    "learning_receipts_public_safe",
+)
 _DB_WORKLOAD_IDS = ("api", "worker", "broker")
 _ROLE_FINGERPRINT_PATTERN = re.compile(r"^[0-9a-f]{64}$")
 _cache: tuple[float, dict[str, Any]] | None = None
@@ -357,6 +365,41 @@ def _build_independent_post_verifier_source_boundary(
     }
 
 
+def _build_canonical_learning_source_boundary(
+    receipt: Mapping[str, str],
+    *,
+    receipt_ref: str,
+    source_sha_matches_deployment: bool,
+) -> dict[str, Any]:
+    controls = {
+        field: _truthy(receipt.get(field))
+        for field in _CANONICAL_LEARNING_REQUIRED_TRUE_FIELDS
+    }
+    blockers = [
+        f"{field}_not_verified"
+        for field, verified in controls.items()
+        if verified is not True
+    ]
+    verifier = receipt.get("canonical_learning_source_verifier") or None
+    if verifier != "cd_tests_and_production_source_sha_v1":
+        blockers.append("canonical_learning_source_not_verified")
+    if not source_sha_matches_deployment:
+        blockers.append("canonical_learning_source_sha_not_deployed")
+    verified = not blockers
+    return {
+        "schema_version": "awoooi_canonical_learning_source_boundary_v1",
+        "status": "verified_ready" if verified else "in_progress",
+        "receipt_ref": receipt_ref,
+        "verifier": verifier,
+        "source_boundary_verified": verified,
+        "source_sha_matches_deployment": source_sha_matches_deployment,
+        "controls": controls,
+        "active_blockers": blockers,
+        "writes_on_read": False,
+        "secret_values_read": False,
+    }
+
+
 def build_executor_trust_boundary_readback(
     data: Mapping[str, Any] | None,
     *,
@@ -405,6 +448,13 @@ def build_executor_trust_boundary_readback(
             source_sha_matches_deployment=source_sha_matches_deployment,
         )
     )
+    canonical_learning_source_boundary = (
+        _build_canonical_learning_source_boundary(
+            receipt,
+            receipt_ref=receipt_ref,
+            source_sha_matches_deployment=source_sha_matches_deployment,
+        )
+    )
     full_workload_boundary_verified = bool(
         ready
         and database_identity_boundary["db_identity_isolated"] is True
@@ -448,6 +498,9 @@ def build_executor_trust_boundary_readback(
         "independent_post_verifier_source_boundary": (
             independent_post_verifier_source_boundary
         ),
+        "canonical_learning_source_boundary": (
+            canonical_learning_source_boundary
+        ),
         "active_blockers": blockers,
         "full_boundary_active_blockers": [
             *blockers,