US02-04: Expose Job APIs and Activity Events (#57)

photo_pipeline/api/app.py

@@ -14,7 +14,7 @@ from pathlib import Path
 from fastapi import FastAPI
 from fastapi.staticfiles import StaticFiles
 
-from photo_pipeline.api.routes import duplicates, health, inventory, thumbnails
+from photo_pipeline.api.routes import duplicates, health, inventory, jobs, thumbnails
 from photo_pipeline.config import Config
 from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
 from photo_pipeline.logging import configure_logging
@@ -44,6 +44,7 @@ def create_app(config: Config | None = None) -> FastAPI:
     app.include_router(health.router, prefix="/api/v1")
     app.include_router(inventory.router, prefix="/api/v1")
     app.include_router(duplicates.router, prefix="/api/v1")
+    app.include_router(jobs.router, prefix="/api/v1")
     app.include_router(thumbnails.router, prefix="/api/v1")
     # Static single-page app (hash-routed). Mounted last so /api/v1 wins.
     if FRONTEND_DIR.is_dir():

photo_pipeline/api/routes/jobs.py

@@ -0,0 +1,116 @@
+"""Job command, status, blocker, and activity-event API.
+
+Operational endpoints return JSON with the shared error envelope; the event stream
+returns SSE with resumable ``id:`` cursors (SQLite rowid). A client that drops the
+stream reconnects with ``Last-Event-ID`` and receives exactly the durable events it
+missed — no duplicates, none lost. The same events are available as plain JSON via
+``?after=`` for a polling fallback.
+"""
+
+from __future__ import annotations
+
+import json
+
+import anyio
+from fastapi import APIRouter, Query, Request
+from fastapi.responses import JSONResponse, StreamingResponse
+
+from photo_pipeline.schemas import JobStartRequest
+from photo_pipeline.services.jobs import (
+    ACTIVE_STATES,
+    InvalidTransition,
+    JobBlocked,
+    JobError,
+    JobService,
+)
+
+router = APIRouter(tags=["jobs"])
+
+
+def _service(request: Request) -> JobService:
+    return JobService(request.app.state.session_factory)
+
+
+def _error(status: int, code: str, message: str) -> JSONResponse:
+    return JSONResponse(status_code=status, content={"error": {"code": code, "message": message}})
+
+
+@router.post("/jobs")
+def start_job(body: JobStartRequest, request: Request):
+    service = _service(request)
+    try:
+        return service.enqueue(
+            body.job_type,
+            lock=body.lock,
+            idempotency_key=body.idempotency_key,
+            items=body.items,
+        )
+    except JobBlocked as error:
+        return _error(409, error.code, str(error))
+
+
+@router.get("/jobs/{job_id}")
+def get_job(job_id: str, request: Request):
+    service = _service(request)
+    snapshot = service.get(job_id)
+    if snapshot is None:
+        return _error(404, "not_found", f"unknown job {job_id}")
+    return {**snapshot, "progress": service.progress(job_id)}
+
+
+@router.post("/jobs/{job_id}/cancel")
+def cancel_job(job_id: str, request: Request):
+    service = _service(request)
+    if service.get(job_id) is None:
+        return _error(404, "not_found", f"unknown job {job_id}")
+    try:
+        return service.cancel(job_id)
+    except InvalidTransition as error:
+        return _error(409, "invalid_transition", str(error))
+    except JobError as error:
+        return _error(409, "job_error", str(error))
+
+
+@router.get("/jobs/{job_id}/blockers")
+def job_blockers(job_id: str, request: Request, lock: str = Query(...)):
+    return {"lock": lock, "blockers": _service(request).blockers(lock)}
+
+
+@router.get("/jobs/{job_id}/events")
+def list_events(job_id: str, request: Request, after: int = 0):
+    service = _service(request)
+    snapshot = service.get(job_id)
+    if snapshot is None:
+        return _error(404, "not_found", f"unknown job {job_id}")
+    return {"state": snapshot["state"], "events": service.events_after(job_id, after)}
+
+
+@router.get("/jobs/{job_id}/events/stream")
+async def stream_events(job_id: str, request: Request, after: int = 0):
+    service = _service(request)
+    if service.get(job_id) is None:
+        return _error(404, "not_found", f"unknown job {job_id}")
+
+    header = request.headers.get("last-event-id")
+    cursor = int(header) if header and header.isdigit() else after
+
+    async def generate():
+        nonlocal cursor
+        # Bounded so a stream can't run forever; a terminal job closes immediately.
+        # ponytail: sync DB reads inside the async loop are fine for a local
+        # single-user app; move to a threadpool/pubsub if it ever multiplexes.
+        for _ in range(600):
+            for event in service.events_after(job_id, cursor):
+                cursor = event["seq"]
+                yield (
+                    f"id: {event['seq']}\n"
+                    f"event: {event['type']}\n"
+                    f"data: {json.dumps({'message': event['message']})}\n\n"
+                )
+            snapshot = service.get(job_id)
+            if snapshot is None or snapshot["state"] not in ACTIVE_STATES:
+                yield "event: done\ndata: {}\n\n"
+                return
+            await anyio.sleep(0.1)
+
+    return StreamingResponse(generate(), media_type="text/event-stream")

photo_pipeline/schemas/__init__.py

@@ -1,5 +1,6 @@
 """Pydantic API request/response contracts."""
 
 from photo_pipeline.schemas.duplicates import DecisionRequest
+from photo_pipeline.schemas.jobs import JobStartRequest
 
-__all__ = ["DecisionRequest"]
+__all__ = ["DecisionRequest", "JobStartRequest"]

photo_pipeline/schemas/jobs.py

@@ -0,0 +1,12 @@
+"""Request contracts for job commands."""
+
+from __future__ import annotations
+
+from pydantic import BaseModel
+
+
+class JobStartRequest(BaseModel):
+    job_type: str
+    lock: str | None = None
+    idempotency_key: str | None = None
+    items: list[str] = []

photo_pipeline/services/jobs.py

@@ -23,7 +23,7 @@ from collections import Counter
 from datetime import datetime, timedelta, timezone
 from collections.abc import Iterable, Sequence
 
-from sqlalchemy import select, update
+from sqlalchemy import select, text, update
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy.orm import sessionmaker
 
@@ -357,6 +357,20 @@ class JobService:
             ).scalars()
             return [{"type": e.event_type, "message": e.message} for e in rows]
 
+    def events_after(self, job_id: str, after: int = 0) -> list[dict]:
+        """Durable events with ``seq`` > ``after``, ordered — the resumable cursor for
+        SSE and polling. SQLite's monotonic rowid is the sequence (job_events is a
+        rowid table), so no extra column is needed."""
+        with self._session_factory() as session:
+            rows = session.execute(
+                text(
+                    "SELECT rowid AS seq, event_type, message FROM job_events "
+                    "WHERE job_id = :jid AND rowid > :after ORDER BY rowid"
+                ),
+                {"jid": job_id, "after": after},
+            ).mappings().all()
+        return [{"seq": r["seq"], "type": r["event_type"], "message": r["message"]} for r in rows]
+
     # ── helpers ──────────────────────────────────────────────────────────────────
     @staticmethod
     def _event(session, job_id: str, event_type: str, message: str | None = None) -> None:

tests/integration/test_jobs_api.py

@@ -0,0 +1,108 @@
+"""Job API contract: typed commands, status codes, error envelope, idempotency,
+cancellation, blockers, and OpenAPI schema."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import run_migrations
+from photo_pipeline.services.jobs import ItemState, JobService, JobState
+
+
+@pytest.fixture
+def client(tmp_path):
+    (tmp_path / "data").mkdir()
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
+    run_migrations(config.database_url)
+    app = create_app(config)
+    with TestClient(app) as test_client:
+        test_client.app = app
+        yield test_client
+
+
+def _drive_to_success(app, job_id, items):
+    service = JobService(app.state.session_factory)
+    service.claim(["scan"], "w1")
+    for item in items:
+        service.set_item(job_id, item, ItemState.RUNNING)
+        service.set_item(job_id, item, ItemState.SUCCEEDED)
+    service.transition(job_id, JobState.SUCCEEDED, worker_id="w1")
+
+
+def test_start_status_and_progress(client):
+    start = client.post("/api/v1/jobs", json={"job_type": "scan", "items": ["a", "b"]})
+    assert start.status_code == 200
+    job = start.json()
+    assert job["state"] == "queued"
+
+    status = client.get(f"/api/v1/jobs/{job['id']}")
+    assert status.status_code == 200
+    body = status.json()
+    assert body["progress"] == {"total": 2, "done": 0, "by_state": {"queued": 2}}
+
+
+def test_idempotent_start_returns_same_job(client):
+    first = client.post("/api/v1/jobs", json={"job_type": "scan", "idempotency_key": "k"}).json()
+    second = client.post("/api/v1/jobs", json={"job_type": "scan", "idempotency_key": "k"}).json()
+    assert first["id"] == second["id"]
+
+
+def test_lock_conflict_returns_409(client):
+    client.post("/api/v1/jobs", json={"job_type": "scan", "lock": "library_write"})
+    conflict = client.post("/api/v1/jobs", json={"job_type": "rename", "lock": "library_write"})
+    assert conflict.status_code == 409
+    assert conflict.json()["error"]["code"] == "lock_held"
+
+
+def test_cancel_queued_then_terminal(client):
+    job = client.post("/api/v1/jobs", json={"job_type": "scan"}).json()
+    cancelled = client.post(f"/api/v1/jobs/{job['id']}/cancel")
+    assert cancelled.status_code == 200
+    assert cancelled.json()["state"] == "cancelled"
+    # Cancelling a terminal job conflicts.
+    again = client.post(f"/api/v1/jobs/{job['id']}/cancel")
+    assert again.status_code == 409
+
+
+def test_blockers_endpoint(client):
+    client.post("/api/v1/jobs", json={"job_type": "scan", "lock": "library_write"})
+    job = client.post("/api/v1/jobs", json={"job_type": "scan"}).json()
+    blockers = client.get(f"/api/v1/jobs/{job['id']}/blockers?lock=library_write").json()
+    assert len(blockers["blockers"]) == 1
+
+
+def test_unknown_job_uses_error_envelope(client):
+    for resp in (
+        client.get("/api/v1/jobs/ghost"),
+        client.post("/api/v1/jobs/ghost/cancel"),
+        client.get("/api/v1/jobs/ghost/events"),
+    ):
+        assert resp.status_code == 404
+        assert resp.json()["error"]["code"] == "not_found"
+
+
+def test_polling_events_with_cursor(client):
+    job = client.post("/api/v1/jobs", json={"job_type": "scan", "items": ["a"]}).json()
+    _drive_to_success(client.app, job["id"], ["a"])
+    all_events = client.get(f"/api/v1/jobs/{job['id']}/events").json()["events"]
+    seqs = [e["seq"] for e in all_events]
+    assert seqs == sorted(seqs) and len(seqs) >= 3
+    after_first = client.get(f"/api/v1/jobs/{job['id']}/events?after={seqs[0]}").json()["events"]
+    assert [e["seq"] for e in after_first] == seqs[1:]
+
+
+def test_openapi_exposes_job_contracts(client):
+    paths = client.get("/openapi.json").json()["paths"]
+    assert "/api/v1/jobs" in paths
+    assert "post" in paths["/api/v1/jobs"]
+    assert "/api/v1/jobs/{job_id}" in paths
+    assert "/api/v1/jobs/{job_id}/cancel" in paths
+    assert "/api/v1/jobs/{job_id}/events/stream" in paths
+
+
+def test_status_content_type_is_json(client):
+    job = client.post("/api/v1/jobs", json={"job_type": "scan"}).json()
+    assert client.get(f"/api/v1/jobs/{job['id']}").headers["content-type"].startswith(
+        "application/json"
+    )

tests/integration/test_jobs_sse.py

@@ -0,0 +1,69 @@
+"""SSE activity stream: resumable event ids, reconnect without duplicate/lost
+events, and parity with the polling fallback."""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from photo_pipeline.api.app import create_app
+from photo_pipeline.config import Config
+from photo_pipeline.db import run_migrations
+from photo_pipeline.services.jobs import ItemState, JobService, JobState
+
+
+@pytest.fixture
+def client(tmp_path):
+    (tmp_path / "data").mkdir()
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
+    run_migrations(config.database_url)
+    app = create_app(config)
+    with TestClient(app) as test_client:
+        test_client.app = app
+        yield test_client
+
+
+def _finished_job(client):
+    job = client.post("/api/v1/jobs", json={"job_type": "scan", "items": ["a", "b"]}).json()
+    service = JobService(client.app.state.session_factory)
+    service.claim(["scan"], "w1")
+    for item in ("a", "b"):
+        service.set_item(job["id"], item, ItemState.RUNNING)
+        service.set_item(job["id"], item, ItemState.SUCCEEDED)
+    service.transition(job["id"], JobState.SUCCEEDED, worker_id="w1")
+    return job["id"]
+
+
+def _read_ids(client, job_id, headers=None):
+    ids, saw_done = [], False
+    with client.stream(
+        "GET", f"/api/v1/jobs/{job_id}/events/stream", headers=headers or {}
+    ) as stream:
+        assert stream.headers["content-type"].startswith("text/event-stream")
+        for line in stream.iter_lines():
+            if line.startswith("id:"):
+                ids.append(int(line.split(":", 1)[1]))
+            elif line.startswith("event: done"):
+                saw_done = True
+    return ids, saw_done
+
+
+def test_sse_streams_all_events_then_closes(client):
+    job_id = _finished_job(client)
+    ids, saw_done = _read_ids(client, job_id)
+    assert ids == sorted(ids) and len(ids) >= 3
+    assert saw_done  # terminal job closes the stream
+
+
+def test_sse_reconnect_has_no_duplicate_or_lost_events(client):
+    job_id = _finished_job(client)
+    full, _ = _read_ids(client, job_id)
+    resume_from = full[1]
+    resumed, saw_done = _read_ids(client, job_id, headers={"Last-Event-ID": str(resume_from)})
+    assert resumed == [i for i in full if i > resume_from]  # no dup, none lost
+    assert saw_done
+
+
+def test_sse_matches_polling(client):
+    job_id = _finished_job(client)
+    sse_ids, _ = _read_ids(client, job_id)
+    polled = client.get(f"/api/v1/jobs/{job_id}/events").json()["events"]
+    assert sse_ids == [e["seq"] for e in polled]

tests/story_traceability.json

@@ -53,6 +53,10 @@
       "tests/unit/test_lock_order.py",
       "tests/integration/test_worker.py",
       "tests/e2e/test_worker_kill.py"
+    ],
+    "US02-04": [
+      "tests/integration/test_jobs_api.py",
+      "tests/integration/test_jobs_sse.py"
     ]
   }
 }