feat(v2): stream execution revisions over sse

backend/src/backup_tool/api/app.py

@@ -2,7 +2,7 @@ import asyncio
 import base64
 import hashlib
 import json
-from collections.abc import AsyncIterator
+from collections.abc import AsyncGenerator, AsyncIterator
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Annotated, Any, cast
@@ -32,6 +32,7 @@ from backup_tool.db.models import (
     User,
 )
 from backup_tool.execution import (
+    TERMINAL_STATES,
     EnqueueError,
     enqueue,
     public_event,
@@ -57,8 +58,30 @@ from backup_tool.security.redaction import redact
 from backup_tool.security.secrets import EnvelopeCipher
 
 
-async def single_execution_event(event: dict[str, object]) -> AsyncIterator[str]:
+def execution_event_id(execution: Execution) -> str:
-    yield f"event: execution\\ndata: {json.dumps(event)}\\n\\n"
+    """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."""
+    last_sent = last_event_id
+    while True:
+        async with sessions() as stream_db:
+            execution = await stream_db.get(Execution, execution_id)
+            if execution is None:
+                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)
 
 
 class Problem(Exception):
@@ -838,14 +861,15 @@ def create_app(settings: Settings) -> FastAPI:
         execution_id: str,
         db: Annotated[AsyncSession, Depends(session)],
         identity: Annotated[tuple[User, set[str], bool], Depends(actor)],
+        last_event_id: Annotated[str | None, Header(alias="Last-Event-ID")] = None,
     ) -> StreamingResponse:
         enforce_scope(identity[1], "execution:read")
-        execution = await db.get(Execution, execution_id)
+        if await db.get(Execution, execution_id) is None:
-        if execution is None:
             raise Problem(404, "resource_not_found", "Execution was not found.")
-
         return StreamingResponse(
-            single_execution_event(public_event(execution)), media_type="text/event-stream"
+            execution_events_stream(app.state.sessions, execution_id, last_event_id),
+            media_type="text/event-stream",
+            headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
         )
 
     @app.get("/api/v2/audit")

tests/integration/test_sources_jobs.py

@@ -5,12 +5,13 @@ from pathlib import Path
 
 import httpx
 import pytest
-from backup_tool.api.app import create_app
+from backup_tool.api.app import create_app, execution_events_stream
 from backup_tool.config import Settings
-from backup_tool.db.models import Execution
+from backup_tool.db.models import Execution, Repository, Source
 from backup_tool.execution import (
     claim,
     complete_cancellation,
+    enqueue,
     heartbeat,
     recover_stale,
     request_cancellation,
@@ -157,6 +158,10 @@ async def test_local_source_probe_archive_and_repository_targeted_job(
         assert retried.status_code == 202
         assert retried.json()["id"] == execution.json()["id"]
         assert retried.json()["attempt"] == 2
+        cancelled = await client.post(
+            f"/api/v2/executions/{execution.json()['id']}/cancel", headers=headers
+        )
+        assert cancelled.status_code == 202
         stream = await client.get(
             f"/api/v2/executions/{execution.json()['id']}/events", headers=headers
         )
@@ -172,6 +177,77 @@ async def test_local_source_probe_archive_and_repository_targeted_job(
         ).status_code == 409
 
 
+@pytest.mark.asyncio
+async def test_execution_sse_replays_later_redacted_revision(tmp_path: Path) -> None:
+    source_root = tmp_path / "sources"
+    source_root.mkdir()
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+    key = tmp_path / "master.key"
+    key.write_bytes(b"x" * 32)
+    key.chmod(0o600)
+    repositories = tmp_path / "repositories"
+    restore = tmp_path / "restore"
+    repositories.mkdir()
+    restore.mkdir()
+    settings = Settings(
+        data_dir=data_dir,
+        database_url=f"sqlite+aiosqlite:///{data_dir / 'db.sqlite'}",
+        repository_roots=(repositories,),
+        local_source_roots=(source_root,),
+        restore_roots=(restore,),
+        master_key_file=key,
+    )
+    app = create_app(settings)
+    from backup_tool.db.models import Base, Job
+
+    async with app.state.engine.begin() as connection:
+        await connection.run_sync(Base.metadata.create_all)
+    async with app.state.sessions() as db:
+        repository = Repository(
+            name="stream-repository",
+            root=str(repositories / "stream"),
+            format_version=1,
+            compression="none",
+            encryption="none",
+        )
+        source = Source(
+            name="stream-source",
+            kind="local",
+            public_config={"root": str(source_root)},
+            secret_refs=[],
+        )
+        db.add_all([repository, source])
+        await db.flush()
+        job = Job(
+            name="stream-job",
+            source_id=source.id,
+            repository_id=repository.id,
+            requested_mode="full",
+            exclusions=[],
+            retention={},
+        )
+        db.add(job)
+        await db.commit()
+        execution = await enqueue(db, job.id)
+        execution_id = execution.id
+
+    stream = execution_events_stream(app.state.sessions, execution_id, None)
+    first = await anext(stream)
+    first_id = first.split("\\n", 1)[0].removeprefix("id: ")
+    assert "queued" in first
+    async with app.state.sessions() as db:
+        assert await request_cancellation(db, execution_id) is not None
+    second = await anext(stream)
+    assert second.split("\\n", 1)[0].removeprefix("id: ") != first_id
+    assert "cancelled" in second
+
+    replay = execution_events_stream(app.state.sessions, execution_id, first_id)
+    assert "cancelled" in await anext(replay)
+    await stream.aclose()
+    await replay.aclose()
+
+
 @pytest.mark.asyncio
 async def test_local_source_rejects_unallowlisted_root(tmp_path: Path) -> None:
     allowed = tmp_path / "allowed"