Complete release dispatch groundwork and mark remaining work blocked

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e429-436c-73f1-9144-bac4336c06e8

src/activity_core/release_dispatch.py

@@ -0,0 +1,135 @@
+"""Explicitly registered release dispatcher; no production credentials or admission."""
+from __future__ import annotations
+
+import asyncio
+import re
+from datetime import timedelta
+
+from temporalio import activity, workflow
+from temporalio.common import RetryPolicy
+from temporalio.exceptions import ActivityError, ApplicationError
+
+with workflow.unsafe.imports_passed_through():
+    from activity_core.release_broker import TERMINAL, Broker
+
+
+class ReleaseActivities:
+    """Trusted worker construction binds the ledger, adapter factory and audit sink.
+
+    The factory receives a ledger plan, never workflow-supplied configuration.
+    The sink must durably upsert by event_id and return only after acknowledgement.
+    This is at-least-once delivery: a lost acknowledgement can repeat an event.
+    """
+
+    def __init__(self, broker: Broker, backend_factory, sink):
+        if not broker.admitted:
+            raise ValueError("broker identity not admitted")
+        self.broker, self.backend_factory, self.sink = broker, backend_factory, sink
+        with broker.connect() as db:
+            db.execute("CREATE TABLE IF NOT EXISTS release_audit_delivered "
+                       "(sequence INTEGER PRIMARY KEY)")
+
+    def _plan(self, release_id):
+        import json
+
+        if not self.broker.admitted:
+            raise ApplicationError("release identity disabled", non_retryable=True)
+        if not isinstance(release_id, str) or not re.fullmatch(r"[0-9a-f]{64}", release_id):
+            raise ApplicationError("invalid release id", non_retryable=True)
+        with self.broker.connect() as db:
+            row = db.execute("SELECT plan FROM releases WHERE id=?", (release_id,)).fetchone()
+        if row is None:
+            raise ApplicationError("unknown admitted release", non_retryable=True)
+        return json.loads(row[0])
+
+    def _advance(self, release_id, plan):
+        return self.broker.advance(release_id, self.backend_factory(plan))
+
+    def _deliver(self, release_id, plan):
+        with self.broker.connect() as db:
+            rows = db.execute(
+                "SELECT sequence, phase, observed_at FROM transitions "
+                "WHERE release_id=? AND sequence NOT IN "
+                "(SELECT sequence FROM release_audit_delivered) "
+                "ORDER BY sequence LIMIT 100", (release_id,),
+            ).fetchall()
+        for sequence, phase, observed_at in rows:
+            # Only normalized facts leave the ledger. No envelopes or adapter errors.
+            self.sink({
+                "event_id": f"release:{release_id}:{sequence}",
+                "event_type": "release_transition", "release_id": release_id,
+                "application": "activity-core", "phase": phase,
+                "observed_at": observed_at, "candidate": plan["candidate"],
+                "rollback": plan["rollback"],
+            })
+            with self.broker.connect() as db:
+                db.execute("INSERT OR IGNORE INTO release_audit_delivered VALUES (?)",
+                           (sequence,))
+        return len(rows)
+
+    @activity.defn(name="advance_admitted_release")
+    async def advance(self, release_id: str) -> str:
+        plan = self._plan(release_id)
+        try:
+            return await asyncio.to_thread(self._advance, release_id, plan)
+        except Exception:
+            # Do not put transport exception text or credentials in Temporal history.
+            raise ApplicationError("release step unavailable", type="ReleaseStepUnavailable") from None
+
+    @activity.defn(name="deliver_release_audit")
+    async def deliver(self, release_id: str) -> int:
+        plan = self._plan(release_id)
+        try:
+            return await asyncio.to_thread(self._deliver, release_id, plan)
+        except Exception:
+            raise ApplicationError("release audit unavailable", type="ReleaseAuditUnavailable") from None
+
+
+@workflow.defn
+class ReleaseDispatchWorkflow:
+    """Resume one admitted ledger entry; receipt admission stays outside Temporal."""
+
+    @workflow.run
+    async def run(self, release_id: str) -> str:
+        for _ in range(100):
+            phase = await workflow.execute_activity(
+                "advance_admitted_release", release_id,
+                start_to_close_timeout=timedelta(minutes=10),
+                retry_policy=RetryPolicy(maximum_interval=timedelta(seconds=30)),
+            )
+            try:
+                await workflow.execute_activity(
+                    "deliver_release_audit", release_id,
+                    start_to_close_timeout=timedelta(minutes=1),
+                    retry_policy=RetryPolicy(
+                        maximum_interval=timedelta(seconds=30),
+                        maximum_attempts=0 if phase in TERMINAL else 1,
+                    ),
+                )
+            except ActivityError:
+                if phase in TERMINAL:
+                    raise
+                # A sink outage must not prevent rollback. Committed transitions
+                # stay pending and terminal completion waits for audit delivery.
+                workflow.logger.warning("Release audit pending; ledger retains transitions")
+            if phase in TERMINAL:
+                return phase
+            if phase == "rollback_synced":
+                await workflow.sleep(timedelta(seconds=30))
+        workflow.continue_as_new(release_id)
+
+
+def release_worker(client, activities: ReleaseActivities):
+    """Dedicated queue: never add privileged release activities to orchestrator-tq.
+
+    Call only from an admitted deployment with durable shared ledger storage,
+    immutable transport configuration and a bounded, idempotent audit sink.
+    This helper starts no worker, workflow or schedule by itself.
+    """
+    from temporalio.worker import Worker
+
+    if not activities.broker.admitted:
+        raise ValueError("broker identity not admitted")
+    return Worker(client, task_queue="activity-core-release-tq",
+                  workflows=[ReleaseDispatchWorkflow],
+                  activities=[activities.advance, activities.deliver])