Milestone 4: bind approvals to canonical execution

ROADMAP.md

@@ -13,7 +13,7 @@ The target is an **Executable Reference Implementation**: one offline, determini
 1. **Repository truth and terminology** — reconcile maturity, versioning, paths, licensing, unsupported claims, and private-reasoning terminology.
 2. **Canonical data contracts** — define deterministic request, context, plan, proposed action, approval, decision, result, and audit records.
 3. **Canonical Python + OPA path** — one offline, fail-closed Layers 1–7 workflow is implemented for the bounded synthetic action.
-4. **Approval binding** — bind synthetic approval to the exact request/action digest, scope, policy version, and expiry.
+4. **Approval binding** — synthetic approval is bound to the exact request/action digests, scope, complete policy reference, authority, and expiry in the canonical path.
 5. **Behavioral scenarios** — test permitted, denied, malformed, unauthorized, unavailable, tampered, recovery, and other required paths.
 6. **Evidence and replay** — correlate records, detect tampering, and reproduce deterministic outcomes.
 7. **Supply chain and CI** — pin dependencies and actions; add integrity, vulnerability, secret, and SBOM checks.

config/policies/canonical-workflow.rego

@@ -14,25 +14,45 @@ expected_policy_ref := {
 
 result := {
     "decision": "permit",
-    "reason_code": "permitted-synthetic-read",
+    "reason_code": "approved-synthetic-read",
     "obligations": ["audit-required", "offline-runtime-required"],
 } if {
     input.policy_ref == expected_policy_ref
-    input.approval_status == "not_required"
     input.request.dry_run == false
     input.request.scope.scope_id == "lab-boundary-0001"
     input.request.scope.environment == "lab"
     input.request.scope.resource_ids == ["workspace-0001"]
     input.actor.subject_id == "investigator-0001"
     input.actor.role == "investigator"
     regex.match("^sha256:[0-9a-f]{64}$", input.action.action_digest)
-    input.action.action_digest == input.authorized_action_digest
+    input.approval.status == "approved"
+    input.approval.request_ref == input.request.reference
+    input.action.request_ref == input.request.reference
+    input.approval.action_ref == {
+        "record_id": input.action.record_id,
+        "record_digest": input.action.record_digest,
+        "action_digest": input.action.action_digest,
+    }
+    input.approval.authority == {
+        "subject_id": "approver-0001",
+        "subject_type": "human",
+        "role": "security-reviewer",
+    }
+    input.approval.policy_ref == input.policy_ref
+    input.approval.approved_scope == input.action.target
+    issued_at := time.parse_rfc3339_ns(input.approval.issued_at)
+    decision_at := time.parse_rfc3339_ns(input.decision_at)
+    execution_at := time.parse_rfc3339_ns(input.execution_at)
+    expires_at := time.parse_rfc3339_ns(input.approval.expires_at)
+    issued_at <= decision_at
+    decision_at <= execution_at
+    execution_at < expires_at
     input.action.actor_id == input.actor.subject_id
     input.action.step_id == "step-0001"
     input.action.tool == {"tool_id": "siem-query", "version": "1.0.0"}
     input.action.operation == "workspace-health"
     input.action.arguments == {"workspace_id": "workspace-0001"}
     input.action.target == input.request.scope
     input.action.risk_tier == "low"
-    input.action.requires_approval == false
+    input.action.requires_approval == true
 }

docs/F7-LAS-QA.md

@@ -32,9 +32,15 @@ executable path is the offline, deterministic Python + OPA workflow under
 Other policy-engine examples remain non-canonical illustrative patterns unless
 explicitly reclassified later.
 
-## How will human approval work?
-
-The approved design requires synthetic approval evidence bound to the exact request/action digest, arguments, target, environment, policy version, and expiry. Approval must return through PDP evaluation and PEP enforcement; it must never bypass policy or invoke a tool directly. This behavior is planned and is not yet implemented.
+## How does approval work in the canonical path?
+
+The canonical path creates deterministic synthetic approval evidence bound to
+the exact request and action references and digests, arguments, complete scope,
+complete policy reference, approving authority, and validity window. Approval
+returns through OPA evaluation and independent executor enforcement; it never
+bypasses policy or invokes a tool directly. This demonstrates the binding and
+enforcement pattern, not an interactive human-approval service or proof that a
+real person's identity was verified.
 
 ## How are Layer 4 and Layer 6 separated?
 

examples/canonical-workflow/README.md

@@ -1,6 +1,6 @@
 # Canonical Offline Workflow
 
-This is the single Milestone 3 executable path across F7-LAS Layers 1–7. It is
+This is the single canonical executable path across F7-LAS Layers 1–7. It is
 deterministic, synthetic, offline, and fail-closed. It does not call an LLM,
 cloud API, production service, or network tool.
 
@@ -11,9 +11,12 @@ synthetic in-process executor that makes no network calls (Layer 6), and
 correlated canonical audit records (Layer 7). Layer 6 here is not an OS or
 container sandbox and does not enforce a network-isolation boundary.
 
-Milestone 3 uses `not_required` for the low-risk synthetic read. Binding an
-explicit approval to the complete request/action/policy/scope/expiry tuple is
-Milestone 4 and is intentionally not claimed here.
+The synthetic approval is bound to the exact request and action references and
+digests, complete scope, complete policy reference, approving authority, issue
+time, and expiry. OPA validates the binding before permitting, and the
+in-process executor independently revalidates it at execution time. This is
+deterministic approval evidence for the reference workflow, not an interactive
+human-approval service or identity proofing system.
 
 ## Requirements
 

schemas/contracts/README.md

@@ -69,10 +69,11 @@ defaults, and self-inclusion of a digest field are not permitted.
 both `record_id` and `record_digest`; action-specific records additionally bind
 the exact `action_digest`.
 
-Milestone 2 defined these contracts, and the Milestone 3 canonical workflow now
-emits and validates them around a real offline OPA decision. This does not claim
-that explicit approval binding, OS/container sandbox containment, enforced
-network isolation, or replay is operational.
+Milestone 2 defined these contracts. The canonical workflow emits and validates
+them around a real offline OPA decision and binds its deterministic synthetic
+approval through PDP and PEP enforcement. This does not claim an interactive
+human-approval service, identity proofing, OS/container sandbox containment,
+enforced network isolation, or replay.
 
 ## Validate
 

scripts/validate-contracts.py

@@ -273,10 +273,13 @@ def records_for_action(collection: list[dict[str, Any]], action_id: str) -> list
             errors.append(f"{decision['record_id']} occurs before its approval disposition")
 
         if approval["status"] == "approved":
+            expires_time = parse_timestamp(approval["expires_at"])
             if approval["approved_scope"] != action["target"]:
                 errors.append(f"{approval['record_id']}.approved_scope must exactly equal action target")
-            if decision_time > parse_timestamp(approval["expires_at"]):
-                errors.append(f"{decision['record_id']} occurs after approval expiry")
+            if expires_time <= approval_time:
+                errors.append(f"{approval['record_id']} validity window must have positive duration")
+            if decision_time >= expires_time:
+                errors.append(f"{decision['record_id']} occurs at or after approval expiry")
         if action["requires_approval"] and approval["status"] != "approved":
             errors.append(f"{action['record_id']} requires an approved approval record")
         if approval["status"] == "not_required" and action["requires_approval"]:
@@ -306,8 +309,8 @@ def records_for_action(collection: list[dict[str, Any]], action_id: str) -> list
             complete = parse_timestamp(result["completed_at"])
             if start < decision_time:
                 errors.append(f"{result['record_id']} starts before the policy decision")
-            if approval["status"] == "approved" and start > parse_timestamp(approval["expires_at"]):
-                errors.append(f"{result['record_id']} starts after approval expiry")
+            if approval["status"] == "approved" and start >= parse_timestamp(approval["expires_at"]):
+                errors.append(f"{result['record_id']} starts at or after approval expiry")
             if complete < start:
                 errors.append(f"{result['record_id']} completes before it starts")
             if parse_timestamp(result["occurred_at"]) < complete:

src/canonical/workflow.py

@@ -31,6 +31,11 @@
     "subject_type": "agent",
     "role": "investigator",
 }
+EXPECTED_APPROVER = {
+    "subject_id": "approver-0001",
+    "subject_type": "human",
+    "role": "security-reviewer",
+}
 EXPECTED_TOOL = {"tool_id": "siem-query", "version": "1.0.0"}
 EXPECTED_OPERATION = "workspace-health"
 EXPECTED_ARGUMENTS = {"workspace_id": "workspace-0001"}
@@ -73,7 +78,7 @@ def _validate_input(workflow_input: dict[str, Any]) -> datetime:
         if set(workflow_input) != required:
             raise WorkflowError("workflow input fields do not match the canonical interface")
         if workflow_input["workflow_id"] != "workflow-0001":
-            raise WorkflowError("Milestone 3 accepts only workflow-0001")
+            raise WorkflowError("canonical workflow accepts only workflow-0001")
         if workflow_input["mission"] != EXPECTED_MISSION:
             raise WorkflowError("mission is outside the canonical offline path")
         if workflow_input["requester"] != EXPECTED_REQUESTER:
@@ -109,7 +114,13 @@ def _finalize(
         return record
 
     @staticmethod
-    def _execute(action: dict[str, Any], decision: dict[str, Any]) -> dict[str, Any]:
+    def _execute(
+        request: dict[str, Any],
+        action: dict[str, Any],
+        approval: dict[str, Any],
+        decision: dict[str, Any],
+        execution_started_at: str,
+    ) -> dict[str, Any]:
         if decision["decision"] != "permit":
             return {"status": "not_executed", "output": {"reason_code": decision["reason_code"]}}
         obligations = set(decision["obligations"])
@@ -127,13 +138,55 @@ def _execute(action: dict[str, Any], decision: dict[str, Any]) -> dict[str, Any]
         expected_digest = calculate_action_digest(action)
         if (
             action["action_digest"] != expected_digest
+            or action["request_ref"] != _reference(request)
             or decision["action_ref"] != _action_reference(action)
+            or not action["requires_approval"]
             or action["tool"] != EXPECTED_TOOL
             or action["operation"] != EXPECTED_OPERATION
             or action["arguments"] != EXPECTED_ARGUMENTS
             or action["target"] != EXPECTED_TARGET
         ):
             return {"status": "not_executed", "output": {"reason_code": "executor-binding-mismatch"}}
+        if (
+            approval["status"] != "approved"
+            or approval["request_ref"] != _reference(request)
+            or approval["action_ref"] != _action_reference(action)
+            or approval["issued_at"] != approval["occurred_at"]
+            or approval["authority"] != EXPECTED_APPROVER
+            or approval["approved_scope"] != action["target"]
+            or approval["policy_ref"] != CanonicalWorkflow._policy_ref()
+            or approval["policy_ref"] != decision["policy_ref"]
+            or decision["request_ref"] != _reference(request)
+            or decision["approval_ref"] != _reference(approval)
+            or decision["authorization_basis"] != "approved"
+        ):
+            return {
+                "status": "not_executed",
+                "output": {"reason_code": "executor-approval-binding-mismatch"},
+            }
+        try:
+            issued_at = datetime.strptime(approval["issued_at"], "%Y-%m-%dT%H:%M:%SZ").replace(
+                tzinfo=timezone.utc
+            )
+            decision_at = datetime.strptime(
+                decision["occurred_at"], "%Y-%m-%dT%H:%M:%SZ"
+            ).replace(tzinfo=timezone.utc)
+            execution_at = datetime.strptime(
+                execution_started_at, "%Y-%m-%dT%H:%M:%SZ"
+            ).replace(tzinfo=timezone.utc)
+            expires_at = datetime.strptime(
+                approval["expires_at"], "%Y-%m-%dT%H:%M:%SZ"
+            ).replace(tzinfo=timezone.utc)
+        except (TypeError, ValueError):
+            return {
+                "status": "not_executed",
+                "output": {"reason_code": "executor-approval-invalid-at-execution"},
+            }
+        if not issued_at <= decision_at <= execution_at < expires_at:
+            return {
+                "status": "not_executed",
+                "output": {"reason_code": "executor-approval-invalid-at-execution"},
+            }
         return {
             "status": "succeeded",
             "output": {
@@ -229,7 +282,7 @@ def header(record_type: str, sequence: int) -> dict[str, Any]:
             "arguments": dict(EXPECTED_ARGUMENTS),
             "target": scope,
             "risk_tier": "low",
-            "requires_approval": False,
+            "requires_approval": True,
         }
         action["action_digest"] = calculate_action_digest(action)
         action = self._finalize(action, plan)
@@ -241,34 +294,43 @@ def header(record_type: str, sequence: int) -> dict[str, Any]:
                 **header("approval", 5),
                 "request_ref": _reference(request),
                 "action_ref": _action_reference(action),
-                "status": "not_required",
-                "authority": {
-                    "subject_id": "canonical-workflow",
-                    "subject_type": "service",
-                    "role": "policy-enforcement-point",
-                },
-                "reason_code": "low-risk-read-only",
+                "status": "approved",
+                "authority": dict(EXPECTED_APPROVER),
+                "reason_code": "approved-synthetic-read",
                 "policy_ref": policy_ref,
                 "issued_at": _timestamp(started_at, 4),
-                "expires_at": None,
-                "approved_scope": None,
+                "expires_at": _timestamp(started_at, 34),
+                "approved_scope": scope,
             },
             action,
         )
         records.append(approval)
 
         policy_action = {field: action[field] for field in ACTION_DIGEST_FIELDS}
+        policy_action["record_id"] = action["record_id"]
+        policy_action["record_digest"] = action["record_digest"]
         policy_action["action_digest"] = action["action_digest"]
         opa_result = self.opa.evaluate(
             {
                 "request": {
+                    "reference": _reference(request),
                     "dry_run": request["constraints"]["dry_run"],
                     "scope": request["scope"],
                 },
                 "actor": context["actor"],
                 "action": policy_action,
-                "authorized_action_digest": approval["action_ref"]["action_digest"],
-                "approval_status": approval["status"],
+                "approval": {
+                    "request_ref": approval["request_ref"],
+                    "action_ref": approval["action_ref"],
+                    "status": approval["status"],
+                    "authority": approval["authority"],
+                    "policy_ref": approval["policy_ref"],
+                    "issued_at": approval["issued_at"],
+                    "expires_at": approval["expires_at"],
+                    "approved_scope": approval["approved_scope"],
+                },
+                "decision_at": _timestamp(started_at, 5),
+                "execution_at": _timestamp(started_at, 6),
                 "policy_ref": policy_ref,
             }
         )
@@ -281,7 +343,7 @@ def header(record_type: str, sequence: int) -> dict[str, Any]:
                 "approval_ref": _reference(approval),
                 "pdp_id": "opa-cli",
                 "decision": decision_value,
-                "authorization_basis": "not_required" if decision_value == "permit" else "none",
+                "authorization_basis": "approved" if decision_value == "permit" else "none",
                 "reason_code": opa_result["reason_code"],
                 "policy_ref": policy_ref,
                 "obligations": sorted(opa_result["obligations"]),
@@ -290,7 +352,14 @@ def header(record_type: str, sequence: int) -> dict[str, Any]:
         )
         records.append(decision)
 
-        execution = self._execute(action, decision)
+        execution_started_at = _timestamp(started_at, 6)
+        execution = self._execute(
+            request,
+            action,
+            approval,
+            decision,
+            execution_started_at,
+        )
         execution_status = execution["status"]
         was_executed = execution_status == "succeeded"
         result = self._finalize(
@@ -312,8 +381,8 @@ def header(record_type: str, sequence: int) -> dict[str, Any]:
                     ),
                 },
                 "status": execution_status,
-                "started_at": _timestamp(started_at, 6) if was_executed else None,
-                "completed_at": _timestamp(started_at, 6) if was_executed else None,
+                "started_at": execution_started_at if was_executed else None,
+                "completed_at": execution_started_at if was_executed else None,
                 "output": execution["output"],
                 "output_digest": calculate_output_digest(execution["output"]),
                 "error_code": None,