Prove beta continuity with the workspace unavailable

Author: durable-workflow-ops

scripts/ci/component-release-recovery.py

@@ -36,6 +36,7 @@
 STATE_SCHEMA = "durable-workflow.component-release-recovery/v1"
 CONTROL_REPOSITORY = "durable-workflow/.github"
 PLAN_TAG_PREFIX = "release-plan/"
+CONTINUITY_TAG_PREFIX = "beta-continuity/"
 FOUNDATION_TAG = "beta-candidate/beta-continuity-foundation"
 FOUNDATION_COMMIT = "4995052410bd4301c5796ffba54e0b6d2f490ed1"
 COMMIT_PATTERN = re.compile(r"^[0-9a-f]{40}$")
@@ -1239,6 +1240,26 @@ def base_state(component: str, tag: str | None = None, plan: dict[str, Any] | No
     }
 
 
+def scheduled_continuity_pause(client: PublicClient, plan: dict[str, Any]) -> dict[str, str] | None:
+    accepted_tag = f"{CONTINUITY_TAG_PREFIX}{plan['plan']}/accepted"
+    accepted_commit = resolve_tag(client, CONTROL_REPOSITORY, accepted_tag)
+    if accepted_commit is None:
+        return None
+    accepted_plan = read_record(client, accepted_tag, accepted_commit, "release-plan.json")
+    validate_plan(accepted_plan)
+    if canonical_json(accepted_plan) != canonical_json(plan):
+        raise RecoveryError("continuity acceptance record names a different release plan", "continuity-gate")
+    resumed_tag = f"{CONTINUITY_TAG_PREFIX}{plan['plan']}/resumed"
+    resumed_commit = resolve_tag(client, CONTROL_REPOSITORY, resumed_tag)
+    if resumed_commit is not None:
+        resumed_plan = read_record(client, resumed_tag, resumed_commit, "release-plan.json")
+        validate_plan(resumed_plan)
+        if canonical_json(resumed_plan) != canonical_json(plan):
+            raise RecoveryError("continuity resume record names a different release plan", "continuity-gate")
+        return None
+    return {"accepted_tag": accepted_tag, "accepted_commit": accepted_commit, "resumed_tag": resumed_tag}
+
+
 def resolution_failure_state(
     component_name: str,
     tag: str | None,
@@ -1453,6 +1474,32 @@ def main() -> int:
                 args.plan_output.write_bytes(canonical_json(plan))
                 if preparation is not None:
                     args.preparation_output.write_bytes(canonical_json(preparation))
+                continuity_pause = scheduled_continuity_pause(client, plan) if args.plan_tag is None else None
+                if continuity_pause is not None:
+                    paused = base_state(args.component, tag, plan)
+                    paused.update(
+                        {
+                            "phase": "continuity-gate",
+                            "outcome": "paused",
+                            "plan_record_commit": record_commit,
+                            "continuity": continuity_pause,
+                            "resume_action": (
+                                f"Wait for {continuity_pause['resumed_tag']} or explicitly recover exact plan {tag}"
+                            ),
+                        }
+                    )
+                    args.evidence.write_bytes(canonical_json(paused))
+                    write_output(
+                        args.github_output,
+                        {
+                            "action": "none",
+                            "plan": str(plan["plan"]),
+                            "channel": str(plan["channel"]),
+                            "plan_tag": tag,
+                            "plan_record_commit": record_commit,
+                        },
+                    )
+                    return 0
                 state, outputs = resolve_component(
                     client,
                     args.component,

scripts/ci/test-component-release-recovery.py

@@ -68,6 +68,40 @@ def load_recovery_for_retry_tests():
     return loader()
 
 
+class ContinuityGateTest(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.recovery = load_recovery_for_retry_tests()
+
+    def test_scheduled_recovery_pauses_until_remote_resume(self) -> None:
+        plan = {"plan": "workspace-unavailable-test"}
+        with (
+            mock.patch.object(
+                self.recovery,
+                "resolve_tag",
+                side_effect=["a" * 40, None],
+            ),
+            mock.patch.object(self.recovery, "read_record", return_value=plan),
+            mock.patch.object(self.recovery, "validate_plan"),
+        ):
+            paused = self.recovery.scheduled_continuity_pause(mock.Mock(), plan)
+
+        self.assertEqual(
+            "beta-continuity/workspace-unavailable-test/resumed",
+            paused["resumed_tag"],
+        )
+        with (
+            mock.patch.object(
+                self.recovery,
+                "resolve_tag",
+                side_effect=["a" * 40, "b" * 40],
+            ),
+            mock.patch.object(self.recovery, "read_record", return_value=plan),
+            mock.patch.object(self.recovery, "validate_plan"),
+        ):
+            self.assertIsNone(self.recovery.scheduled_continuity_pause(mock.Mock(), plan))
+
+
 class PublicClientRetryTest(unittest.TestCase):
     @classmethod
     def setUpClass(cls) -> None: