feat(v2): expose execution control and event APIs

backend/src/backup_tool/api/app.py

@@ -1,6 +1,7 @@
 import asyncio
 import base64
 import hashlib
+import json
 from collections.abc import AsyncIterator
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
@@ -8,7 +9,7 @@ from typing import Annotated, Any, cast
 
 from fastapi import Depends, FastAPI, Header, Request, Response
 from fastapi.exceptions import RequestValidationError
-from fastapi.responses import JSONResponse
+from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel, Field
 from sqlalchemy import desc, select
 from sqlalchemy.exc import IntegrityError
@@ -21,6 +22,7 @@ from backup_tool.db.engine import SchemaNotCurrentError, assert_schema_current,
 from backup_tool.db.models import (
     ApiToken,
     AuditEvent,
+    Execution,
     IdempotencyRecord,
     Job,
     Repository,
@@ -29,7 +31,13 @@ from backup_tool.db.models import (
     Source,
     User,
 )
-from backup_tool.execution import EnqueueError, enqueue
+from backup_tool.execution import (
+    EnqueueError,
+    enqueue,
+    public_event,
+    request_cancellation,
+    retry,
+)
 from backup_tool.repository import (
     RepositoryError,
     initialize,
@@ -49,6 +57,10 @@ 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]:
+    yield f"event: execution\\ndata: {json.dumps(event)}\\n\\n"
+
+
 class Problem(Exception):
     def __init__(self, status: int, code: str, detail: str):
         self.status = status
@@ -777,6 +789,56 @@ def create_app(settings: Settings) -> FastAPI:
             "attempt": execution.attempt,
         }
 
+    @app.get("/api/v2/executions/{execution_id}")
+    async def get_execution(
+        execution_id: str,
+        db: Annotated[AsyncSession, Depends(session)],
+        _: Annotated[tuple[User, set[str], bool], Depends(actor)],
+    ) -> dict[str, object]:
+        execution = await db.get(Execution, execution_id)
+        if execution is None:
+            raise Problem(404, "resource_not_found", "Execution was not found.")
+        return public_event(execution)
+
+    @app.post("/api/v2/executions/{execution_id}/cancel", status_code=202)
+    async def cancel_execution(
+        execution_id: str,
+        db: Annotated[AsyncSession, Depends(session)],
+        _: Annotated[tuple[User, set[str], bool], Depends(require)],
+    ) -> dict[str, object]:
+        execution = await request_cancellation(db, execution_id)
+        if execution is None:
+            raise Problem(409, "cancellation_not_allowed", "Execution cannot be cancelled.")
+        return public_event(execution)
+
+    @app.post("/api/v2/executions/{execution_id}/retry", status_code=202)
+    async def retry_execution(
+        execution_id: str,
+        db: Annotated[AsyncSession, Depends(session)],
+        _: Annotated[tuple[User, set[str], bool], Depends(require)],
+    ) -> dict[str, object]:
+        try:
+            execution = await retry(db, execution_id)
+        except EnqueueError as error:
+            raise Problem(409, error.code, str(error)) from error
+        if execution is None:
+            raise Problem(404, "resource_not_found", "Execution was not found.")
+        return public_event(execution)
+
+    @app.get("/api/v2/executions/{execution_id}/events")
+    async def execution_events(
+        execution_id: str,
+        db: Annotated[AsyncSession, Depends(session)],
+        _: Annotated[tuple[User, set[str], bool], Depends(actor)],
+    ) -> StreamingResponse:
+        execution = await db.get(Execution, execution_id)
+        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"
+        )
+
     @app.get("/api/v2/audit")
     async def list_audit(
         db: Annotated[AsyncSession, Depends(session)],

backend/src/backup_tool/execution.py

@@ -11,6 +11,7 @@ from backup_tool.db.models import Execution, Job
 
 ACTIVE_STATES = frozenset({"queued", "preparing", "running", "verifying", "cancelling"})
 TERMINAL_STATES = frozenset({"committed", "cancelled", "failed"})
+TRANSIENT_REASONS = frozenset({"worker_lost", "timeout", "unavailable", "transient_io"})
 
 _ALLOWED: Mapping[str, frozenset[str]] = {
     "queued": frozenset({"preparing", "cancelled", "failed"}),
@@ -107,6 +108,71 @@ async def heartbeat(
     return getattr(result, "rowcount", 0) == 1
 
 
+async def request_cancellation(db: AsyncSession, execution_id: str) -> Execution | None:
+    result = await db.execute(
+        update(Execution)
+        .where(
+            Execution.id == execution_id,
+            Execution.state.in_({"queued", "preparing", "running"}),
+        )
+        .values(state="cancelling", reason_code="cancellation_requested")
+    )
+    if getattr(result, "rowcount", 0) != 1:
+        await db.rollback()
+        return None
+    await db.commit()
+    return await db.get(Execution, execution_id)
+
+
+async def complete_cancellation(db: AsyncSession, execution_id: str, owner: str) -> bool:
+    result = await db.execute(
+        update(Execution)
+        .where(
+            Execution.id == execution_id,
+            Execution.lease_owner == owner,
+            Execution.state == "cancelling",
+        )
+        .values(
+            state="cancelled",
+            completed_at=datetime.now(UTC),
+            lease_owner=None,
+            lease_expires_at=None,
+        )
+    )
+    await db.commit()
+    return getattr(result, "rowcount", 0) == 1
+
+
+async def retry(db: AsyncSession, execution_id: str) -> Execution | None:
+    execution = await db.get(Execution, execution_id)
+    if execution is None:
+        return None
+    if execution.state != "failed" or execution.reason_code not in TRANSIENT_REASONS:
+        raise EnqueueError("retry_not_allowed", "Execution failure is not retryable.")
+    execution.state = "queued"
+    execution.attempt += 1
+    execution.lease_owner = None
+    execution.lease_expires_at = None
+    execution.heartbeat_at = None
+    execution.reason_code = None
+    execution.operator_message = None
+    await db.commit()
+    await db.refresh(execution)
+    return execution
+
+
+def public_event(execution: Execution) -> dict[str, object]:
+    """Return redacted progress suitable for polling or SSE."""
+    progress = {k: v for k, v in execution.progress.items() if k not in {"path", "secret", "token"}}
+    return {
+        "id": execution.id,
+        "state": execution.state,
+        "attempt": execution.attempt,
+        "reason_code": execution.reason_code,
+        "progress": progress,
+    }
+
+
 async def recover_stale(db: AsyncSession) -> int:
     now = datetime.now(UTC)
     result = await db.execute(

tests/integration/test_sources_jobs.py

@@ -6,12 +6,15 @@ import httpx
 import pytest
 from backup_tool.api.app import create_app
 from backup_tool.config import Settings
+from backup_tool.db.models import Execution
 
 PASSWORD = "correct-horse-battery-staple"
 
 
 async def login(client: httpx.AsyncClient) -> dict[str, str]:
-    response = await client.post("/api/v2/setup", json={"username": "admin", "password": PASSWORD})
+    response = await client.post(
+        "/api/v2/setup", json={"username": "admin", "password": PASSWORD}
+    )
     assert response.status_code == 201
     return {"X-CSRF-Token": client.cookies["backup_tool_csrf"]}
 
@@ -46,7 +49,9 @@ async def test_local_source_probe_archive_and_repository_targeted_job(
     async with app.state.engine.begin() as connection:
         await connection.run_sync(Base.metadata.create_all)
     transport = httpx.ASGITransport(app=app)
-    async with httpx.AsyncClient(transport=transport, base_url="https://test") as client:
+    async with httpx.AsyncClient(
+        transport=transport, base_url="https://test"
+    ) as client:
         headers = await login(client)
         repository = await client.post(
             "/api/v2/repositories",
@@ -94,6 +99,31 @@ async def test_local_source_probe_archive_and_repository_targeted_job(
         )
         assert duplicate.status_code == 409
         assert duplicate.json()["code"] == "execution_active"
+        polled = await client.get(f"/api/v2/executions/{execution.json()['id']}", headers=headers)
+        assert polled.status_code == 200
+        assert polled.json()["state"] == "queued"
+        cancellation = await client.post(
+            f"/api/v2/executions/{execution.json()['id']}/cancel", headers=headers
+        )
+        assert cancellation.status_code == 202
+        assert cancellation.json()["state"] == "cancelling"
+        async with app.state.sessions() as db:
+            stored = await db.get(Execution, execution.json()["id"])
+            assert stored is not None
+            stored.state = "failed"
+            stored.reason_code = "timeout"
+            await db.commit()
+        retried = await client.post(
+            f"/api/v2/executions/{execution.json()['id']}/retry", headers=headers
+        )
+        assert retried.status_code == 202
+        assert retried.json()["id"] == execution.json()["id"]
+        assert retried.json()["attempt"] == 2
+        stream = await client.get(
+            f"/api/v2/executions/{execution.json()['id']}/events", headers=headers
+        )
+        assert stream.status_code == 200
+        assert "event: execution" in stream.text
         archived = await client.delete(f"/api/v2/sources/{source_id}", headers=headers)
         assert archived.status_code == 204
         assert (