README.md

@@ -49,3 +49,30 @@ work_item/scripts/python -m pytest tests/e2e tests/integration -q
 - `tests/story_traceability.json` maps every delivered story to its tests;
   `tests/e2e/test_traceability.py` fails if a Phase A story loses coverage or a test
   file is left unexercised.
+
+### Phase B acceptance gate
+
+Phase B (Epic E02: durable jobs, workflow shell, safety/analysis views) is proven
+through the real process and browser boundaries. One command runs the Phase B API,
+worker-recovery, and browser (Playwright) suites:
+
+```bash
+work_item/scripts/python -m pytest tests/e2e -m phase_b -q
+```
+
+- `tests/e2e/test_phase_b_pipeline.py` launches the real server and durable worker as
+  child processes and drives them only over HTTP/SSE: analysis start/progress,
+  resumable SSE reconnect, the polling fallback, cancellation, per-asset error
+  inspection, one-mutating-job rejection during read-only browsing, the NSFW→vision
+  privacy gate, and durability across a full restart.
+- `tests/e2e/test_worker_kill.py` kills a worker mid-item and proves a fresh worker
+  resumes the fenced job (the "resume" journey).
+- `tests/e2e/test_analysis_browser.py` starts a job from the Analyze view and watches
+  live progress arrive over the browser's real SSE adapter.
+
+The full Phase B regression, including the unchanged Phase A gate, is the whole
+end-to-end suite:
+
+```bash
+work_item/scripts/python -m pytest tests/e2e -q
+```

photo_pipeline/__main__.py

@@ -27,6 +27,11 @@ def main(argv: Sequence[str] | None = None) -> int:
 
     if args.command == "worker":
         from photo_pipeline.db import create_db_engine, create_session_factory
+
+        # Importing this registers the safety/analysis job handlers into the shared
+        # REGISTRY the Worker defaults to; without it a standalone worker process
+        # claims nothing because it knows no job types.
+        import photo_pipeline.jobs.domain_handlers  # noqa: F401
         from photo_pipeline.jobs.worker import Worker
 
         run_migrations(config.database_url)

photo_pipeline/services/analysis.py

@@ -17,6 +17,7 @@ ponytail: port the donor's retry+RPD throttling when analysis runs at real volum
 from __future__ import annotations
 
 import json
+import os
 import uuid
 from datetime import datetime, timezone
 from typing import Protocol
@@ -216,9 +217,45 @@ def _result_dict(row: AnalysisResult) -> dict:
 
 
 def _default_provider() -> VisionProvider:
+    # Test seam (concept §18: deterministic fakes replace the vision edge, enabled
+    # only by test configuration). When this env var names a writable log file, the
+    # worker/API use a fake that records every analyzed path — so the SFW-only gate
+    # can be asserted end-to-end through the real integration layer — instead of the
+    # real OpenAI-compatible call. Never set in production.
+    log_path = os.environ.get("PHOTO_PIPELINE_FAKE_VISION_LOG")
+    if log_path:
+        return _RecordingFakeVision(log_path)
     return OpenAIVisionProvider()
 
 
+class _RecordingFakeVision:
+    """Deterministic vision fake for end-to-end tests. Appends every analyzed file
+    path to its log so a test can prove NSFW assets never reach the provider, and
+    raises for a path whose stem contains ``boom`` to exercise per-asset error
+    handling. Constructed only when ``PHOTO_PIPELINE_FAKE_VISION_LOG`` is set."""
+
+    def __init__(self, log_path: str) -> None:
+        self._log_path = log_path
+
+    def analyze(self, path: str, *, album_hint: str) -> dict:
+        with open(self._log_path, "a", encoding="utf-8") as handle:
+            handle.write(path + "\n")
+        if "boom" in os.path.splitext(os.path.basename(path))[0]:
+            raise AnalysisError("fake vision failure")
+        return {
+            "description": f"a deterministic scene in {album_hint}",
+            "tags": ["fixture", "deterministic"],
+            "people_count": 1,
+            "setting": "outdoor",
+            "time_of_day": "day",
+            "season": "summer",
+            "mood": "calm",
+            "location_hint": None,
+            "approx_year": None,
+            "_tokens": 7,
+        }
+
+
 class OpenAIVisionProvider:
     """The real provider: an OpenAI-compatible vision call (Gemini by default).
 

pyproject.toml

@@ -27,3 +27,6 @@ line-length = 100
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
+markers = [
+    "phase_b: Phase B end-to-end acceptance (US02-07) — API, worker-recovery, and browser journeys",
+]

tests/e2e/_pipeline_harness.py

@@ -0,0 +1,161 @@
+"""Reusable Phase B end-to-end harness: seed an isolated library, then launch the
+real API server and durable worker as child processes so tests talk to them only
+over HTTP/SSE — the same boundaries used in production (concept §18).
+
+External vision is replaced at its integration edge by the in-repo recording fake
+(``PHOTO_PIPELINE_FAKE_VISION_LOG``); it is invoked through the real AnalysisService
+and worker, never mocked inside a test.
+"""
+
+from __future__ import annotations
+
+import os
+import socket
+import subprocess
+import sys
+import time
+from dataclasses import dataclass
+from pathlib import Path
+
+import httpx
+import numpy as np
+from PIL import Image
+
+REPO = Path(__file__).resolve().parents[2]
+
+
+def image(path: Path, seed: int) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    rng = np.random.default_rng(seed)
+    Image.fromarray(rng.integers(0, 256, (96, 128, 3), dtype=np.uint8)).save(path, quality=90)
+
+
+def free_port() -> int:
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+@dataclass
+class Seeded:
+    data: Path
+    lib: Path
+    asset_ids: dict[str, str]  # filename stem -> asset id
+
+
+def seed_library(tmp_path: Path, files: dict[str, int], decisions: dict[str, str]) -> Seeded:
+    """Scan ``files`` (stem -> seed) into a fresh DB and apply safety ``decisions``
+    (stem -> ``sfw``/``nsfw``) in-process before any server starts, so the seed is
+    deterministic and independent of the API under test."""
+    data = tmp_path / "data"
+    data.mkdir()
+    lib = tmp_path / "lib"
+    lib.mkdir()
+    for stem, seed in files.items():
+        image(lib / f"{stem}.jpg", seed)
+
+    from photo_pipeline.config import Config
+    from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
+    from photo_pipeline.models import Asset
+    from photo_pipeline.services.inventory import InventoryService
+    from photo_pipeline.services.safety import SafetyService
+    from sqlalchemy import select
+
+    config = Config.from_env(
+        {"PHOTO_PIPELINE_DATA_DIR": str(data), "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib)}
+    )
+    run_migrations(config.database_url)
+    engine = create_db_engine(config.database_url)
+    sf = create_session_factory(engine)
+    InventoryService(sf).scan(lib)
+    with sf() as session:
+        rows = list(session.execute(select(Asset.id, Asset.current_path)).all())
+    ids = {Path(path).stem: aid for aid, path in rows}
+    safety = SafetyService(sf)
+    for stem, decision in decisions.items():
+        safety.decide(ids[stem], decision, write_exif=False)
+    engine.dispose()
+    return Seeded(data=data, lib=lib, asset_ids=ids)
+
+
+def _env(seeded: Seeded, port: int, *, extra: dict[str, str] | None = None) -> dict[str, str]:
+    env = {
+        "PATH": os.environ.get("PATH", ""),
+        "PHOTO_PIPELINE_DATA_DIR": str(seeded.data),
+        "PHOTO_PIPELINE_LIBRARY_ROOTS": str(seeded.lib),
+        "PHOTO_PIPELINE_HOST": "127.0.0.1",
+        "PHOTO_PIPELINE_PORT": str(port),
+    }
+    if extra:
+        env.update(extra)
+    return env
+
+
+class Server:
+    """A real ``photo_pipeline serve`` child process, ready when /health/ready is 200."""
+
+    def __init__(self, seeded: Seeded, *, extra_env: dict[str, str] | None = None) -> None:
+        self._seeded = seeded
+        self._extra = extra_env
+        self.port = free_port()
+        self.base = f"http://127.0.0.1:{self.port}"
+        self.proc: subprocess.Popen | None = None
+
+    def start(self) -> "Server":
+        self.proc = subprocess.Popen(
+            [sys.executable, "-m", "photo_pipeline", "serve"],
+            cwd=str(REPO),
+            env=_env(self._seeded, self.port, extra=self._extra),
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        deadline = time.monotonic() + 30
+        while time.monotonic() < deadline:
+            if self.proc.poll() is not None:
+                _, err = self.proc.communicate()
+                raise RuntimeError(f"server exited: {err.decode(errors='replace')}")
+            try:
+                if httpx.get(f"{self.base}/api/v1/health/ready", timeout=1).status_code == 200:
+                    return self
+            except httpx.HTTPError:
+                time.sleep(0.2)
+        self.stop()
+        raise RuntimeError("server never became ready")
+
+    def stop(self) -> None:
+        if self.proc is None:
+            return
+        self.proc.terminate()
+        try:
+            self.proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            self.proc.kill()
+        self.proc = None
+
+
+def start_worker(seeded: Seeded, *, fake_vision_log: Path) -> subprocess.Popen:
+    """Launch a real durable worker wired to the recording vision fake."""
+    return subprocess.Popen(
+        [sys.executable, "-m", "photo_pipeline", "worker", "--id", "e2e-worker"],
+        cwd=str(REPO),
+        env=_env(
+            seeded,
+            free_port(),  # unused by the worker, but keeps the env shape uniform
+            extra={"PHOTO_PIPELINE_FAKE_VISION_LOG": str(fake_vision_log)},
+        ),
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+
+
+def wait_until(predicate, *, timeout: float = 20, interval: float = 0.1):
+    """Poll ``predicate`` until it returns a truthy value or the timeout elapses.
+    Returns the truthy value; raises on timeout. Synchronizes on observable state
+    instead of sleeping for a fixed duration (concept §18)."""
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        value = predicate()
+        if value:
+            return value
+        time.sleep(interval)
+    raise AssertionError("condition not met before timeout")

tests/e2e/test_analysis_browser.py

@@ -0,0 +1,58 @@
+"""Phase B browser journey (US02-07): start a content-analysis job from the Analyze
+view and watch live progress arrive over the real SSE adapter, against a real server
+and durable worker. The frontend's SSE→polling event adapter and the worker are
+exercised end to end; the vision edge is the recording fake.
+"""
+
+from __future__ import annotations
+
+import pytest
+from playwright.sync_api import expect
+
+from tests.e2e._pipeline_harness import Server, seed_library, start_worker
+
+pytestmark = pytest.mark.phase_b
+
+
+@pytest.fixture
+def running_stack(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1}, {"beach": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        yield server
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+def test_analyze_start_shows_live_progress_and_completes(page, running_stack):
+    errors = []
+    page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None)
+
+    page.goto(f"{running_stack.base}/app/#/analyze")
+    run = page.get_by_test_id("run-analysis")
+    run.wait_for()
+    assert run.is_enabled()  # one confirmed-SFW asset is eligible
+
+    run.click()
+
+    # Live progress streams into the activity log via the browser's SSE adapter.
+    log = page.get_by_test_id("analyze-log")
+    log.get_by_text("Started analysis job").wait_for()
+
+    # On completion the view re-renders with the updated counts: the SFW asset is
+    # now analysed. Waiting on this observable state (not a sleep) proves the whole
+    # start → worker → SSE → done round trip worked.
+    analyzed_value = (
+        page.get_by_test_id("analyze-counts")
+        .locator(".stat")
+        .filter(has_text="Analyzed")
+        .locator(".stat-value")
+    )
+    # ``expect`` re-queries the locator each poll, so it survives the re-render that
+    # replaces the counts DOM when the job completes.
+    expect(analyzed_value).to_have_text("1", timeout=15000)
+
+    assert errors == [], f"console errors: {errors}"

tests/e2e/test_phase_b_pipeline.py

@@ -0,0 +1,254 @@
+"""Phase B black-box end-to-end acceptance (US02-07).
+
+Launches the real FastAPI server and durable worker as child processes and drives
+them only over HTTP/SSE — no service/repository imports for the behaviour under
+test. Covers the Phase B journeys the browser suite does not exhaustively assert:
+analysis start/progress, resumable SSE reconnect, the polling fallback,
+cancellation, per-asset error inspection, one-mutating-job rejection during
+read-only browsing, the NSFW→vision privacy gate, and durability across restart.
+
+External vision is the in-repo recording fake, invoked through the real worker.
+"""
+
+from __future__ import annotations
+
+import json
+
+import httpx
+import pytest
+
+from tests.e2e._pipeline_harness import Server, seed_library, start_worker, wait_until
+
+pytestmark = pytest.mark.phase_b
+
+TIMEOUT = 10
+
+
+def _job_state(base: str, job_id: str) -> str:
+    return httpx.get(f"{base}/api/v1/jobs/{job_id}", timeout=TIMEOUT).json()["state"]
+
+
+def _sse_events(
+    base: str,
+    job_id: str,
+    *,
+    after: int = 0,
+    last_event_id: int | None = None,
+    stop_after: int | None = None,
+) -> list[dict]:
+    """Read the resumable SSE stream, returning ``[{seq, type, message}, ...]``. When
+    ``stop_after`` is set, close the stream after that many events to model a client
+    that drops mid-stream. ``last_event_id`` sends the reconnect header."""
+    headers = {}
+    if last_event_id is not None:
+        headers["Last-Event-ID"] = str(last_event_id)
+    url = f"{base}/api/v1/jobs/{job_id}/events/stream?after={after}"
+    events: list[dict] = []
+    seq: int | None = None
+    with httpx.stream("GET", url, headers=headers, timeout=TIMEOUT) as response:
+        response.raise_for_status()
+        for line in response.iter_lines():
+            if line.startswith("id:"):
+                seq = int(line[3:].strip())
+            elif line.startswith("data:"):
+                payload = json.loads(line[5:].strip())
+                if seq is not None:  # skip the terminal `event: done` (empty {} data)
+                    events.append({"seq": seq, **payload})
+                    seq = None
+                    if stop_after is not None and len(events) >= stop_after:
+                        return events
+            elif line.startswith("event: done"):
+                break
+    return events
+
+
+# ── analysis job: start, progress, and completion ────────────────────────────
+
+
+def test_analysis_job_runs_to_completion_over_real_http(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1}, {"beach": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        counts = httpx.get(f"{server.base}/api/v1/analysis/counts", timeout=TIMEOUT).json()
+        assert counts["analyzed"] == 1
+        # The fake actually ran through the real worker for the one SFW asset.
+        assert log.read_text().strip().endswith("beach.jpg")
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+# ── resumable SSE reconnect and the polling fallback ─────────────────────────
+
+
+def test_sse_reconnect_resumes_without_gap_or_duplicate(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1, "meadow": 2}, {"beach": "sfw", "meadow": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        full = _sse_events(server.base, job["id"])
+        assert [e["type"] for e in full][:2] == ["queued", "claimed"]
+        assert full[-1]["type"] == "state:succeeded"
+        assert len(full) >= 3
+
+        # A client that reads only the first event, drops, then reconnects with the
+        # last id it saw must receive exactly the remaining events — no gap, no dup.
+        first = _sse_events(server.base, job["id"], stop_after=1)
+        assert len(first) == 1
+        rest = _sse_events(server.base, job["id"], last_event_id=first[0]["seq"])
+        assert first + rest == full
+        seqs = [e["seq"] for e in full]
+        assert seqs == sorted(set(seqs))  # strictly increasing, unique
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+def test_polling_fallback_returns_the_same_durable_events(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1}, {"beach": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        sse = _sse_events(server.base, job["id"])
+        polled = httpx.get(
+            f"{server.base}/api/v1/jobs/{job['id']}/events?after=0", timeout=TIMEOUT
+        ).json()
+        assert polled["state"] == "succeeded"
+        assert [(e["seq"], e["type"]) for e in polled["events"]] == [
+            (e["seq"], e["type"]) for e in sse
+        ]
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+# ── cancellation ─────────────────────────────────────────────────────────────
+
+
+def test_queued_job_cancels_and_runs_no_items(tmp_path):
+    # No worker: the safety job stays queued so cancellation is deterministic.
+    seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {})
+    server = Server(seeded).start()
+    try:
+        job = httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=TIMEOUT).json()
+        cancelled = httpx.post(
+            f"{server.base}/api/v1/jobs/{job['id']}/cancel", timeout=TIMEOUT
+        ).json()
+        assert cancelled["state"] == "cancelled"
+
+        snapshot = httpx.get(f"{server.base}/api/v1/jobs/{job['id']}", timeout=TIMEOUT).json()
+        assert snapshot["progress"]["by_state"].get("succeeded", 0) == 0
+    finally:
+        server.stop()
+
+
+# ── per-asset error inspection without losing completed work ──────────────────
+
+
+def test_one_analysis_error_is_inspectable_and_others_still_complete(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1, "boom": 2}, {"beach": "sfw", "boom": "sfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        counts = httpx.get(f"{server.base}/api/v1/analysis/counts", timeout=TIMEOUT).json()
+        assert counts["analyzed"] == 1 and counts["error"] == 1  # boom failed, beach kept
+
+        boom_id = seeded.asset_ids["boom"]
+        result = httpx.get(
+            f"{server.base}/api/v1/analysis/results/{boom_id}", timeout=TIMEOUT
+        ).json()
+        assert result["status"] == "error"
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+# ── one mutating job at a time, read-only browsing stays available ───────────
+
+
+def test_second_mutation_rejected_while_reads_continue(tmp_path):
+    # No worker: the first safety job holds the library_write lock for the test.
+    seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {})
+    server = Server(seeded).start()
+    try:
+        first = httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=TIMEOUT)
+        assert first.status_code == 200
+
+        second = httpx.post(f"{server.base}/api/v1/safety/jobs", timeout=TIMEOUT)
+        assert second.status_code == 409
+        assert second.json()["error"]["code"] == "lock_held"
+
+        # Read-only browsing is unaffected by the held mutation lock.
+        for path in ("/api/v1/workflow", "/api/v1/library/assets", "/api/v1/safety/queue"):
+            assert httpx.get(f"{server.base}{path}", timeout=TIMEOUT).status_code == 200
+    finally:
+        server.stop()
+
+
+# ── privacy gate: NSFW assets never reach the vision provider ────────────────
+
+
+def test_nsfw_asset_never_reaches_the_vision_provider(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {"beach": "sfw", "city": "nsfw"})
+    log = tmp_path / "vision.log"
+    server = Server(seeded).start()
+    worker = start_worker(seeded, fake_vision_log=log)
+    try:
+        job = httpx.post(f"{server.base}/api/v1/analysis/jobs", timeout=TIMEOUT).json()
+        wait_until(lambda: _job_state(server.base, job["id"]) == "succeeded", timeout=TIMEOUT)
+
+        logged = log.read_text()
+        assert "beach.jpg" in logged  # the SFW asset was analysed
+        assert "city.jpg" not in logged  # the NSFW asset never reached the provider
+
+        # And the NSFW asset was never even enqueued: no analysis row exists for it.
+        city_id = seeded.asset_ids["city"]
+        assert (
+            httpx.get(
+                f"{server.base}/api/v1/analysis/results/{city_id}", timeout=TIMEOUT
+            ).status_code
+            == 404
+        )
+    finally:
+        worker.terminate()
+        server.stop()
+
+
+# ── durability: decisions survive a full server restart ──────────────────────
+
+
+def test_safety_decisions_survive_a_restart(tmp_path):
+    seeded = seed_library(tmp_path, {"beach": 1, "city": 2}, {"beach": "sfw", "city": "nsfw"})
+    server = Server(seeded).start()
+    try:
+        before = httpx.get(f"{server.base}/api/v1/safety/queue?state=nsfw", timeout=TIMEOUT).json()
+        assert any(row["asset_id"] == seeded.asset_ids["city"] for row in before["items"])
+    finally:
+        server.stop()
+
+    # Fresh process against the same data dir: the durable decision is still there.
+    restarted = Server(seeded).start()
+    try:
+        after = httpx.get(
+            f"{restarted.base}/api/v1/safety/queue?state=nsfw", timeout=TIMEOUT
+        ).json()
+        assert any(row["asset_id"] == seeded.asset_ids["city"] for row in after["items"])
+    finally:
+        restarted.stop()

tests/e2e/test_worker_kill.py

@@ -14,6 +14,10 @@ from photo_pipeline.services.jobs import ItemState, JobService, JobState
 
 REPO = Path(__file__).resolve().parents[2]
 
+# Worker-recovery is also part of the Phase B acceptance command (US02-07: the
+# "resume" journey); it stays mapped to US02-03 for traceability.
+pytestmark = pytest.mark.phase_b
+
 # A worker that claims the job, marks its item running, signals it started, then
 # blocks — so the test can kill it mid-item.
 WORKER_SCRIPT = """

tests/story_traceability.json

@@ -64,6 +64,10 @@
     "US02-06": [
       "tests/integration/test_safety_analysis.py",
       "tests/e2e/test_workflow_views.py"
+    ],
+    "US02-07": [
+      "tests/e2e/test_phase_b_pipeline.py",
+      "tests/e2e/test_analysis_browser.py"
     ]
   }
 }

work_item/src/work_item/core.py

@@ -446,19 +446,24 @@ class Gitea:
         self.tea(*args)
 
     def claim(self, story: Story) -> None:
-        self.tea(
-            "issues",
-            "edit",
-            str(story.number),
-            "--repo",
-            self.config.repo_slug,
-            "--add-assignees",
-            self.config.assignee,
-            "--add-labels",
-            "status/in-progress",
-            "--remove-labels",
-            "status/backlog,status/ready,status/blocked,status/review,status/done",
+        self.edit_labels(
+            story.number,
+            add=("status/in-progress",),
+            remove=(
+                "status/backlog",
+                "status/ready",
+                "status/blocked",
+                "status/review",
+                "status/done",
+            ),
         )
+        self.set_assignee(story.number, self.config.assignee)
+
+    def set_assignee(self, issue: int, assignee: str) -> None:
+        # Some Gitea/Forgejo deployments 404 on the issues/{n}/assignees
+        # sub-route that `tea --add-assignees` uses; the issue PATCH endpoint
+        # accepts the full assignees list and works across those versions.
+        self.api(f"{self.base}/issues/{issue}", method="PATCH", data={"assignees": [assignee]})
 
     def comment(self, issue: int, body: str) -> None:
         self.tea("comments", "add", str(issue), body, "--repo", self.config.repo_slug)

work_item/tests/test_cli_e2e.py

@@ -50,6 +50,13 @@ if args[0] == "api":
         out(state["prs"][str(number)])
     elif "/commits/" in endpoint and endpoint.endswith("/status"):
         out({"state": state.get("ci_state", "success")})
+    elif "--method" in args and args[args.index("--method") + 1] == "PATCH" and "/issues/" in endpoint:
+        number = int(endpoint.rsplit("/", 1)[1])
+        issue = next(x for x in state["issues"] if x["number"] == number)
+        payload = json.loads(args[args.index("--data") + 1])
+        if "assignees" in payload:
+            issue["assignees"] = [{"login": login} for login in payload["assignees"]]
+        save(); out(issue)
     else:
         raise SystemExit(f"unsupported api: {endpoint}")
 elif args[:2] == ["issues", "edit"]: