feat(v2): persist execution event revisions

backend/alembic/versions/0003_execution_events.py

@@ -0,0 +1,37 @@
+"""add durable execution event revision
+
+Revision ID: 0003_execution_events
+Revises: 0002_sessions
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0003_execution_events"
+down_revision = "0002_sessions"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("executions") as batch:
+        batch.add_column(sa.Column("revision", sa.Integer(), nullable=False, server_default="0"))
+    op.create_table(
+        "execution_events",
+        sa.Column("id", sa.Integer(), primary_key=True, autoincrement=True),
+        sa.Column(
+            "execution_id",
+            sa.String(36),
+            sa.ForeignKey("executions.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("revision", sa.Integer(), nullable=False),
+        sa.Column("payload", sa.JSON(), nullable=False),
+        sa.UniqueConstraint("execution_id", "revision", name="uq_execution_event_revision"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("execution_events")
+    with op.batch_alter_table("executions") as batch:
+        batch.drop_column("revision")

backend/src/backup_tool/api/app.py

@@ -23,6 +23,7 @@ from backup_tool.db.models import (
     ApiToken,
     AuditEvent,
     Execution,
+    ExecutionEvent,
     IdempotencyRecord,
     Job,
     Repository,
@@ -58,28 +59,34 @@ from backup_tool.security.redaction import redact
 from backup_tool.security.secrets import EnvelopeCipher
 
 
-def execution_event_id(execution: Execution) -> str:
-    """Stable monotonic cursor for an execution's current durable revision."""
-    updated = execution.updated_at.astimezone(UTC).isoformat()
-    return f"{updated}:{execution.state}:{execution.attempt}"
-
-
 async def execution_events_stream(
     sessions: async_sessionmaker[AsyncSession], execution_id: str, last_event_id: str | None
 ) -> AsyncGenerator[str, None]:
-    """Poll durable execution state and yield each changed redacted revision as SSE."""
+    """Replay every durable event after Last-Event-ID, then poll for appended events."""
-    last_sent = last_event_id
+    try:
+        last_revision = int(last_event_id or "0")
+    except ValueError:
+        last_revision = 0
     while True:
         async with sessions() as stream_db:
+            events = (
+                await stream_db.scalars(
+                    select(ExecutionEvent)
+                    .where(
+                        ExecutionEvent.execution_id == execution_id,
+                        ExecutionEvent.revision > last_revision,
+                    )
+                    .order_by(ExecutionEvent.revision)
+                )
+            ).all()
+            for event in events:
+                yield (
+                    f"id: {event.revision}\\nevent: execution\\n"
+                    f"data: {json.dumps(event.payload)}\\n\\n"
+                )
+                last_revision = event.revision
             execution = await stream_db.get(Execution, execution_id)
-            if execution is None:
+            if execution is None or (execution.state in TERMINAL_STATES and not events):
-                return
-            event_id = execution_event_id(execution)
-            if event_id != last_sent:
-                payload = public_event(execution)
-                yield (f"id: {event_id}\\nevent: execution\\ndata: {json.dumps(payload)}\\n\\n")
-                last_sent = event_id
-            if execution.state in TERMINAL_STATES:
                 return
         await asyncio.sleep(0.1)
 

backend/src/backup_tool/db/models.py

@@ -165,6 +165,7 @@ class Execution(IdentityMixin, TimestampMixin, Base):
     lease_expires_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
     heartbeat_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
     progress: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
+    revision: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
     reason_code: Mapped[str | None] = mapped_column(String(64))
     operator_message: Mapped[str | None] = mapped_column(Text)
     started_at: Mapped[datetime | None] = mapped_column(UTCDateTime())
@@ -189,6 +190,19 @@ class Execution(IdentityMixin, TimestampMixin, Base):
     )
 
 
+class ExecutionEvent(Base):
+    __tablename__ = "execution_events"
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    execution_id: Mapped[str] = mapped_column(
+        ForeignKey("executions.id", ondelete="CASCADE"), nullable=False
+    )
+    revision: Mapped[int] = mapped_column(Integer, nullable=False)
+    payload: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False)
+    __table_args__ = (
+        UniqueConstraint("execution_id", "revision", name="uq_execution_event_revision"),
+    )
+
+
 class Backup(IdentityMixin, Base):
     __tablename__ = "backups"
     execution_id: Mapped[str] = mapped_column(

backend/src/backup_tool/execution.py

@@ -7,7 +7,7 @@ from sqlalchemy import select, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from backup_tool.db.models import Execution, Job
+from backup_tool.db.models import Execution, ExecutionEvent, Job
 from backup_tool.security.redaction import redact
 
 ACTIVE_STATES = frozenset({"queued", "preparing", "running", "verifying", "cancelling"})
@@ -67,6 +67,8 @@ async def enqueue(db: AsyncSession, job_id: str, trigger: str = "manual") -> Exe
             "execution_active", "Job already has an active execution.", existing
         ) from error
     await db.refresh(execution)
+    await record_event(db, execution)
+    await db.commit()
     return execution
 
 
@@ -89,7 +91,11 @@ async def claim(
         await db.rollback()
         return None
     await db.commit()
-    return await db.get(Execution, execution_id)
+    execution = await db.get(Execution, execution_id)
+    if execution is not None:
+        await record_event(db, execution)
+        await db.commit()
+    return execution
 
 
 async def heartbeat(
@@ -128,7 +134,11 @@ async def request_cancellation(db: AsyncSession, execution_id: str) -> Execution
         await db.rollback()
         return None
     await db.commit()
-    return await db.get(Execution, execution_id)
+    execution = await db.get(Execution, execution_id)
+    if execution is not None:
+        await record_event(db, execution)
+        await db.commit()
+    return execution
 
 
 async def complete_cancellation(db: AsyncSession, execution_id: str, owner: str) -> bool:
@@ -177,6 +187,8 @@ async def retry(db: AsyncSession, execution_id: str) -> Execution | None:
             "execution_active", "Job already has an active execution.", active_id
         ) from error
     await db.refresh(execution)
+    await record_event(db, execution)
+    await db.commit()
     return execution
 
 
@@ -196,6 +208,19 @@ def _redact_progress(value: object) -> object:
     return redact(value)
 
 
+async def record_event(db: AsyncSession, execution: Execution) -> ExecutionEvent:
+    """Append the public representation as the next durable event revision."""
+    execution.revision += 1
+    event = ExecutionEvent(
+        execution_id=execution.id,
+        revision=execution.revision,
+        payload=public_event(execution),
+    )
+    db.add(event)
+    await db.flush()
+    return event
+
+
 def public_event(execution: Execution) -> dict[str, object]:
     """Return redacted progress suitable for polling or SSE."""
     progress = _redact_progress(execution.progress)
@@ -203,6 +228,7 @@ def public_event(execution: Execution) -> dict[str, object]:
         "id": execution.id,
         "state": execution.state,
         "attempt": execution.attempt,
+        "revision": execution.revision,
         "reason_code": execution.reason_code,
         "progress": progress,
     }

tests/integration/test_migrations.py

@@ -20,6 +20,7 @@ EXPECTED_TABLES = {
     "audit_events",
     "backups",
     "executions",
+    "execution_events",
     "idempotency_records",
     "jobs",
     "notification_deliveries",