fix: apply synthetic safety guards to retry execution evidence

integrations/agentrust-confidential-workflow-harness/README.md

@@ -61,8 +61,23 @@ earlier observation in that lineage. A later observation cannot overwrite an
 earlier boundary outcome. Unsupported mutation names are rejected before the
 adapter runs. Response verification points to the actual
 main-lineage execution observation, including timeout and no-dispatch outcomes.
-Retry observations record authorization from the shared synthetic inputs in
-their own lineage; they do not replace the main path's evidence.
+Retry observations apply authorization, version admission, revocation, replay
+and execution-evidence checks to the shared synthetic inputs in their own
+lineage. Their receipt and admission observations precede execution. A retry
+after a main-lineage timeout models a separately observed synthetic attempt;
+its result does not replace the main path's evidence. No live retry is run.
+`retry-dispatch` and `retry-receipt` represent scripted synthetic transport
+observations, including when authorization is contradicted; neither establishes
+authorized delivery or permission to execute. Consumers must inspect the
+lineage's authorization, admission and guarded execution outcomes. A known
+rejection (authorization, version admission, revocation or replay) produces
+`execution=contradicted` even if execution evidence is also missing. Missing
+evidence produces `unavailable` only when these guards allow execution. The
+explicit missing-evidence mutation cannot override a known guard refusal.
+The synthetic execution outcome evaluates the permitted-execution claim;
+`contradicted` does not prove that no physical side effects occurred. Retry
+attempts are independently scripted even when the main input has `dispatch=False`;
+their observations do not establish real-world dispatch.
 
 The public result must never contain protected payloads, secrets, device IDs or
 low-entropy digests of protected material.

integrations/agentrust-confidential-workflow-harness/harness/adapters/deterministic.py

@@ -58,6 +58,24 @@ def emit(boundary, outcome, source, caused_by=None, lineage="main", detail=None)
             "authorization-check",
         )
 
+        def admission_ok():
+            return (scenario.mutation == "disable_version_gate"
+                    or _meets_version_floor(i.get("software_version", "1.0.0")))
+
+        def execution_outcome(admitted):
+            revoked = (i.get("revoked_before_use", False)
+                       and scenario.mutation != "disable_revocation_gate")
+            replayed = (i.get("replay", False)
+                        and scenario.mutation != "disable_replay_gate")
+            # Known guard refusal takes precedence over missing execution evidence.
+            if not auth_ok or not admitted or replayed or revoked:
+                return Outcome.CONTRADICTED
+            if i.get("missing_execution_evidence", False):
+                return (Outcome.ESTABLISHED
+                        if scenario.mutation == "assume_missing_execution_success"
+                        else Outcome.UNAVAILABLE)
+            return Outcome.ESTABLISHED
+
         if i.get("dispatch", True):
             emit(Boundary.DELIVERY, Outcome.ESTABLISHED, "dispatch", "authorization-check")
             if i.get("timeout_after_dispatch", False):
@@ -66,37 +84,16 @@ def emit(boundary, outcome, source, caused_by=None, lineage="main", detail=None)
                 emit(Boundary.EXECUTION, Outcome.UNAVAILABLE, "execution-unavailable", "admission-unavailable")
             else:
                 emit(Boundary.RECEIPT, Outcome.ESTABLISHED, "peer-receipt", "dispatch")
-                admitted = _meets_version_floor(i.get("software_version", "1.0.0"))
-                if scenario.mutation == "disable_version_gate":
-                    admitted = True
+                admitted = admission_ok()
                 emit(
                     Boundary.ADMISSION,
                     Outcome.ESTABLISHED if admitted else Outcome.CONTRADICTED,
                     "version-admission",
                     "peer-receipt",
                 )
-                revoked = i.get("revoked_before_use", False)
-                if scenario.mutation == "disable_revocation_gate":
-                    revoked = False
-                replayed = i.get("replay", False)
-                if scenario.mutation == "disable_replay_gate":
-                    replayed = False
-                missing_execution_evidence = i.get("missing_execution_evidence", False)
-                executed = auth_ok and admitted and not replayed and not revoked
-                if missing_execution_evidence:
-                    execution_outcome = Outcome.UNAVAILABLE
-                else:
-                    execution_outcome = (
-                        Outcome.ESTABLISHED if executed else Outcome.CONTRADICTED
-                    )
-                if (
-                    scenario.mutation == "assume_missing_execution_success"
-                    and missing_execution_evidence
-                ):
-                    execution_outcome = Outcome.ESTABLISHED
                 emit(
                     Boundary.EXECUTION,
-                    execution_outcome,
+                    execution_outcome(admitted),
                     "execution-check",
                     "version-admission",
                 )
@@ -114,6 +111,7 @@ def emit(boundary, outcome, source, caused_by=None, lineage="main", detail=None)
                 "authorization-check",
                 lineage="retry-1",
             )
+            # Scripted transport observations do not assert authorized delivery.
             emit(
                 Boundary.DELIVERY,
                 Outcome.ESTABLISHED,
@@ -122,12 +120,27 @@ def emit(boundary, outcome, source, caused_by=None, lineage="main", detail=None)
                 lineage="retry-1",
             )
             emit(
-                Boundary.EXECUTION,
+                Boundary.RECEIPT,
                 Outcome.ESTABLISHED,
-                "retry-execution",
+                "retry-receipt",
                 "retry-dispatch",
                 lineage="retry-1",
             )
+            retry_admitted = admission_ok()
+            emit(
+                Boundary.ADMISSION,
+                Outcome.ESTABLISHED if retry_admitted else Outcome.CONTRADICTED,
+                "retry-admission",
+                "retry-receipt",
+                lineage="retry-1",
+            )
+            emit(
+                Boundary.EXECUTION,
+                execution_outcome(retry_admitted),
+                "retry-execution",
+                "retry-admission",
+                lineage="retry-1",
+            )
 
         execution_main = next(
             (

integrations/agentrust-confidential-workflow-harness/tests/test_harness.py

@@ -201,6 +201,93 @@ def test_retry_does_not_overwrite_main_lineage(self):
             )
         )
 
+    def test_retry_execution_reuses_guards_and_mutation_controls(self):
+        cases = [
+            ({"workload": "workload-X"}, "disable_authorization_gate", "contradicted"),
+            ({"key_id": "key-X"}, "disable_authorization_gate", "contradicted"),
+            ({"policy_version": "p0"}, "disable_authorization_gate", "contradicted"),
+            ({"software_version": "0.9.0"}, "disable_version_gate", "contradicted"),
+            ({"revoked_before_use": True}, "disable_revocation_gate", "contradicted"),
+            ({"replay": True}, "disable_replay_gate", "contradicted"),
+            ({"missing_execution_evidence": True}, "assume_missing_execution_success", "unavailable"),
+        ]
+
+        def retry_execution(result):
+            return next(o for o in result.observations
+                        if o.lineage == "retry-1" and o.boundary == Boundary.EXECUTION)
+
+        for changes, mutation, expected in cases:
+            with self.subTest(inputs=changes):
+                good = self.runner.run(scenario(inputs={**scenario().inputs, "retry_lineage": True}))
+                self.assertEqual(retry_execution(good).outcome, Outcome.ESTABLISHED)
+                inputs = {**scenario().inputs, **changes, "retry_lineage": True}
+                bad = self.runner.run(scenario(inputs=inputs))
+                self.assertEqual(retry_execution(bad).outcome.value, expected)
+                self.assertEqual(bad.boundary_outcomes["execution"], expected)
+                weakened = self.runner.run(scenario(inputs=inputs, mutation=mutation))
+                self.assertEqual(retry_execution(weakened).outcome, Outcome.ESTABLISHED)
+                execution = retry_execution(bad)
+                admission = next(o for o in bad.observations
+                                 if o.lineage == "retry-1" and o.source == execution.caused_by)
+                self.assertEqual(admission.boundary, Boundary.ADMISSION)
+
+    def test_synthetic_retry_transport_does_not_permit_invalid_authorization(self):
+        for changes in ({"workload": "workload-X"}, {"key_id": "key-X"},
+                        {"policy_version": "p0"}):
+            with self.subTest(inputs=changes):
+                result = self.runner.run(scenario(inputs={
+                    **scenario().inputs, **changes, "retry_lineage": True,
+                }))
+                retry = {o.source: o for o in result.observations if o.lineage == "retry-1"}
+                self.assertEqual(retry["retry-dispatch"].outcome, Outcome.ESTABLISHED)
+                self.assertEqual(retry["retry-receipt"].outcome, Outcome.ESTABLISHED)
+                self.assertEqual(retry["authorization-check"].outcome, Outcome.CONTRADICTED)
+                self.assertEqual(retry["retry-execution"].outcome, Outcome.CONTRADICTED)
+                self.assertEqual(result.status, "refused")
+                self.assertEqual(result.boundary_outcomes["release/disclosure"], "contradicted")
+
+    def test_missing_execution_evidence_cannot_mask_known_guard_refusal(self):
+        cases = ({"workload": "workload-X"}, {"key_id": "key-X"},
+                 {"policy_version": "p0"}, {"software_version": "0.9.0"},
+                 {"revoked_before_use": True}, {"replay": True})
+        for changes in cases:
+            for mutation in (None, "assume_missing_execution_success"):
+                with self.subTest(inputs=changes, mutation=mutation):
+                    result = self.runner.run(scenario(inputs={
+                        **scenario().inputs, **changes, "retry_lineage": True,
+                        "missing_execution_evidence": True,
+                    }, mutation=mutation))
+                    executions = [o for o in result.observations
+                                  if o.boundary == Boundary.EXECUTION]
+                    self.assertEqual({o.lineage for o in executions}, {"main", "retry-1"})
+                    self.assertTrue(all(o.outcome == Outcome.CONTRADICTED for o in executions))
+                    self.assertEqual(result.boundary_outcomes["execution"], "contradicted")
+                    self.assertEqual(result.status, "refused")
+
+    def test_retry_rejection_dominates_missing_evidence(self):
+        cases = (
+            ({"workload": "workload-X"}, "disable_authorization_gate"),
+            ({"key_id": "key-X"}, "disable_authorization_gate"),
+            ({"policy_version": "p0"}, "disable_authorization_gate"),
+            ({"software_version": "0.9.0"}, "disable_version_gate"),
+            ({"revoked_before_use": True}, "disable_revocation_gate"),
+            ({"replay": True}, "disable_replay_gate"),
+        )
+        for invalid, mutation in cases:
+            with self.subTest(invalid=invalid):
+                inputs = {**scenario().inputs, **invalid,
+                          "retry_lineage": True, "missing_execution_evidence": True}
+                result = self.runner.run(scenario(inputs=inputs))
+                retry = next(o for o in result.observations
+                             if o.lineage == "retry-1" and o.boundary == Boundary.EXECUTION)
+                self.assertEqual(retry.outcome, Outcome.CONTRADICTED)
+                self.assertEqual(result.status, "refused")
+                weakened = self.runner.run(scenario(inputs=inputs, mutation=mutation))
+                retry_weakened = next(o for o in weakened.observations
+                                      if o.lineage == "retry-1" and o.boundary == Boundary.EXECUTION)
+                self.assertEqual(retry_weakened.outcome, Outcome.UNAVAILABLE)
+                self.assertEqual(weakened.status, "unknown")
+
     def test_response_parent_is_actual_main_execution(self):
         for overrides, expected_source in (
             ({}, "execution-check"),

marketplace/catalog.json

@@ -1,6 +1,6 @@
 {
   "catalog_version": 1,
-  "count": 46,
+  "count": 47,
   "integrations": [
     {
       "name": "Claude Code",