US08-03: Compose the Runtime and Mount the Library Safely

2026-08-19 19:46:20 +02:00
12 changed files with 158 additions and 1499 deletions
--- a/.gitea/workflows/deploy.yml
+++ b/.gitea/workflows/deploy.yml
@@ -1,99 +0,0 @@
 # Publish and redeploy (US08-04). Adapted from the crowdsec-admin deploy workflow:
 # build, log in to the Gitea registry, push, trigger the Portainer webhook, prune.
 # The difference is the gate — this project has a required suite that must not be
 # skipped, so publishing happens only after `Test` (`.gitea/workflows/test.yml`)
 # succeeded on `main`, never on the push itself.
 #
 # The stack is managed by Portainer from git (`docker-compose.yml`), and the runtime
 # secrets it needs — vision key, Immich key, access secret — live in the Portainer
 # stack's environment. They are deliberately not repository secrets and are not in the
 # image: rotation stays in one place, and a repository read never discloses them.
 #
 # Repository secrets required:
 #   REGISTRY_USER          user with write:package on the registry
 #   REGISTRY_TOKEN         that user's token
 #   PORTAINER_WEBHOOK_URL  POST URL from the stack's auto-update setting
 name: Deploy
 on:
  workflow_run:
    workflows:
      - Test
    types:
      - completed
    branches:
      - main
  workflow_dispatch:
    inputs:
      dry_run:
        description: Build and push a scratch tag only — leave `latest` and the running stack alone
        type: boolean
        default: true
 concurrency:
  # Deliberately not keyed by commit: the point is that two deploys of *different*
  # commits cannot overlap. Queued, not cancelled — a half-pushed tag set is worse
  # than a late one.
  group: deploy
  cancel-in-progress: false
 env:
  IMAGE: git.domverse-berlin.eu/domverse/photoanalyzer
  # The commit that was tested, not whatever `main` points at by the time this starts.
  SHA: ${{ gitea.event.workflow_run.head_sha || gitea.sha }}
 jobs:
  publish:
    # A completed `Test` run is not a passing one.
    if: >-
      (gitea.event_name == 'workflow_run' && gitea.event.workflow_run.conclusion == 'success')
      || (gitea.event_name == 'workflow_dispatch' && inputs.dry_run == false)
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4
        with:
          ref: ${{ env.SHA }}
      - name: Log in to the Gitea registry
        run: echo "${{ secrets.REGISTRY_TOKEN }}" | docker login git.domverse-berlin.eu -u "${{ secrets.REGISTRY_USER }}" --password-stdin
      - name: Build and push
        # The commit tag is pushed first, so a `latest` that exists is always a tag
        # that also exists under its own commit — which is what makes a rollback a
        # tag change rather than a rebuild.
        run: |
          docker build -t "$IMAGE:$SHA" -t "$IMAGE:latest" .
          docker push "$IMAGE:$SHA"
          docker push "$IMAGE:latest"
      - name: Trigger the Portainer redeploy
        # --fail turns an HTTP error into a non-zero exit: a redeploy that did not
        # happen must not read as a green deploy.
        run: curl -sS --fail -X POST "${{ secrets.PORTAINER_WEBHOOK_URL }}"
      - name: Prune dangling images
        # Untagged layers only. Published tags are the rollback history; `-a` would
        # delete exactly the images this workflow exists to keep.
        run: docker image prune -f
  dry-run:
    # Manual only, and the default: prove the image still builds and the registry
    # still accepts it without moving `latest` or touching the running stack.
    if: gitea.event_name == 'workflow_dispatch' && inputs.dry_run
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4
      - name: Log in to the Gitea registry
        run: echo "${{ secrets.REGISTRY_TOKEN }}" | docker login git.domverse-berlin.eu -u "${{ secrets.REGISTRY_USER }}" --password-stdin
      - name: Build and push a scratch tag
        run: |
          docker build -t "$IMAGE:scratch-$SHA" .
          docker push "$IMAGE:scratch-$SHA"
      - name: Prune dangling images
        run: docker image prune -f
--- a/.gitea/workflows/test.yml
+++ b/.gitea/workflows/test.yml
@@ -1,103 +0,0 @@
 # The test gate. Deploy waits for this workflow by name (`.gitea/workflows/deploy.yml`
 # triggers on `workflow_run: [Test]`), so renaming it here without renaming it there
 # would leave `main` publishing without a suite. `tests/integration/test_deploy_workflows.py`
 # asserts both halves of that link, and that the commands below are still the ones
 # configured in `work_item/.work-item.yml`.
 name: Test
 on:
  pull_request:
  push:
    branches:
      - main
 concurrency:
  # One run per branch; a newer push makes the older run's answer irrelevant.
  group: test-${{ gitea.ref }}
  cancel-in-progress: true
 jobs:
  suites:
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        uses: actions/checkout@v4
      - name: Install exiftool
        # The EXIF checkpoints are the safety invariant of every metadata stage; a run
        # without exiftool would skip them for a reason the release gate accepts.
        run: |
          SUDO=
          if [ "$(id -u)" -ne 0 ]; then SUDO=sudo; fi
          $SUDO apt-get update
          $SUDO apt-get install -y --no-install-recommends libimage-exiftool-perl
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - name: Install the application and its test dependencies
        run: |
          python -m pip install --upgrade pip
          python -m pip install -e '.[test]'
          python -m playwright install --with-deps chromium
      - name: Helper suite
        run: work_item/scripts/python -m unittest discover -s work_item/tests -v
      - name: Application suite
        run: work_item/scripts/python -m pytest tests -q
  # The deployed container, verified the way the host application is (US08-05). It
  # runs on `main` only — a pull request has nothing published to upgrade *from*, and
  # building two images per push would pay for that on every commit. Deploy waits for
  # this whole workflow, so a red container gate is a deploy that does not happen.
  container:
    if: gitea.event_name == 'push'
    runs-on: ubuntu-latest
    steps:
      - name: Checkout
        # The upgrade journey builds the previous commit's tree when no published
        # image is named, so the history has to be there.
        uses: actions/checkout@v4
        with:
          fetch-depth: 2
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - name: Install the application and its test dependencies
        run: |
          python -m pip install --upgrade pip
          python -m pip install -e '.[test]'
          python -m playwright install --with-deps chromium
      - name: Container acceptance gate
        # One command: it builds the image, provisions the composition against a
        # temporary fixture library and an isolated volume, runs the phase_h journeys,
        # destroys the stack, and writes the evidence. Any skipped check fails it.
        env:
          # The fixture libraries are bind-mounted into the containers, so this path
          # has to be one the Docker daemon can see. On a runner that talks to a
          # sibling daemon, point it at a shared host path instead of the workspace —
          # an unshared path arrives as an empty mount and the journeys fail on the
          # scan, which is the symptom to recognise.
          PHOTO_PIPELINE_DATA_DIR: ${{ gitea.workspace }}/gate-data
          PHOTO_PIPELINE_LIBRARY_ROOTS: ${{ gitea.workspace }}/gate-library
          PHOTO_PIPELINE_TEST_MOUNT_BASE: ${{ gitea.workspace }}/gate-mounts
        run: |
          mkdir -p "$PHOTO_PIPELINE_LIBRARY_ROOTS" "$PHOTO_PIPELINE_TEST_MOUNT_BASE"
          python -m photo_pipeline container-gate --output gate-evidence
      - name: Keep the evidence
        # Retained per run, and retained on failure especially: the logs are the only
        # account of what the containers did.
        if: always()
        uses: actions/upload-artifact@v3
        with:
          name: container-gate-${{ gitea.sha }}
          path: gate-evidence
--- a/README.md
+++ b/README.md
@@ -192,88 +192,7 @@ service the command borrows the image and mounts from. `restore` is deliberately
 an API call: it replaces the state of an installation and belongs to a stopped one,
 so stop `api` and `worker` first and restart them against the restored directory.
-## Publishing and deploying (US08-04)
+Publishing and deploying the image is US08-04.
 Two workflows in `.gitea/workflows/` make `main` the only path to the running stack:
 | workflow | runs on | does |
 |---|---|---|
 | `test.yml` (**Test**) | every pull request, and every push to `main` | installs exiftool, the package, and Chromium, then runs the suites configured in `work_item/.work-item.yml` |
 | `deploy.yml` (**Deploy**) | a **successful** `Test` run on `main`, or manual dispatch | builds the image, pushes it, triggers the Portainer webhook, prunes dangling layers |
 Deploy waits for `Test` through `workflow_run`, so publishing is downstream of a green
 suite rather than of a push. `completed` is not `success`: the publish job runs only on
 `conclusion == 'success'`. Because an unmatched `workflow_run` filter does not fail but
 simply never fires, `tests/integration/test_deploy_workflows.py` asserts the link from
 both ends — renaming either workflow, or dropping one of the configured suites out of
 the test job, fails that test rather than silently unhooking the gate.
 The image is `git.domverse-berlin.eu/domverse/photoanalyzer`, tagged with the commit
 SHA and `latest`. The commit tag is pushed first, so a `latest` that exists always has
 a commit tag beside it and **a rollback is a tag change, not a rebuild**:
 ```bash
 docker pull  git.domverse-berlin.eu/domverse/photoanalyzer:<known-good-sha>
 docker tag   git.domverse-berlin.eu/domverse/photoanalyzer:<known-good-sha> \
             git.domverse-berlin.eu/domverse/photoanalyzer:latest
 docker push  git.domverse-berlin.eu/domverse/photoanalyzer:latest
 # then trigger the Portainer webhook, or redeploy the stack from Portainer
 ```
 Manual dispatch defaults to **dry run**: it builds and pushes `scratch-<sha>` only, and
 touches neither `latest` nor the running stack. Clear the `dry_run` input to publish by
 hand. A workflow-level `concurrency: deploy` group with `cancel-in-progress: false`
 queues deploys instead of overlapping them, so two commits can never race to `latest`.
 ### Secrets
 Repository secrets — used only by the pipeline:
 | secret | purpose |
 |---|---|
 | `REGISTRY_USER` | registry user with `write:package` |
 | `REGISTRY_TOKEN` | that user's token, passed on stdin, never as an argument |
 | `PORTAINER_WEBHOOK_URL` | the stack's auto-update POST URL; `curl --fail` makes a refused redeploy a failed workflow |
 Runtime secrets — `PHOTO_PIPELINE_VISION_API_KEY`, `PHOTO_PIPELINE_IMMICH_API_KEY`,
 `PHOTO_PIPELINE_ACCESS_SECRET` — are **not** repository secrets and are not in the
 image. The stack is managed by Portainer from git (`docker-compose.yml`), and those
 values live in the Portainer stack's environment, so rotation is one place and a
 repository read discloses nothing. A test asserts the workflows never name them.
 ## Container acceptance gate (US08-05)
 The deployed container is verified the way the host application is, by one command:
 ```bash
 work_item/scripts/python -m photo_pipeline container-gate --output gate-evidence
 ```
 It builds the image, provisions the composition against a **temporary fixture library
 on a bind mount and an isolated data volume**, runs the `phase_h` journeys against it,
 destroys every stack afterwards, and writes `release-report.json`, `logs/container.log`,
 and `CHECKSUMS.sha256` into the evidence directory. It exits non-zero when a journey
 fails, when the story matrix has a hole, **or when any check skipped at all** — unlike
 the release gate, this one accepts no environment excuse: a run that never reached the
 containers proved nothing about them.
 The journeys (`tests/e2e/test_phase_h_container.py`) are:
 | journey | what it proves |
 |---|---|
 | browser | discovery, duplicate review, analysis, album proposal, rename, upload preflight, and archive views, driven through the containerized frontend — including a rename that really moves the operator's folder on the bind mount |
 | upgrade | the previous version's image runs first, then this one against the same volume: schema at the new head, assets, analysis results, job progress, the unapplied rename plan, and the thumbnail cache all survive |
 | restart | `docker kill` on both containers mid-job; the job resumes, every photo ends with exactly one stored result, and only the in-flight item ever reaches the provider twice |
 | security | no session refused, a forged `X-Forwarded-Host` cannot smuggle an allowed hostname past the check, a symlink out of the mounted library is refused, an unmounted library root refuses startup, and no secret appears in `docker compose logs` |
 The upgrade journey builds the previous commit's tree when no published image is named;
 point it at the real one with `PHOTO_PIPELINE_PREVIOUS_IMAGE`. On a Docker VM (Colima,
 Docker Desktop) the fixture library must live on a shared path — it defaults to
 `~/.cache/photo-pipeline`, overridable with `PHOTO_PIPELINE_TEST_MOUNT_BASE`.
 CI runs this gate as the `container` job of **Test** on pushes to `main` and keeps its
 evidence as a run artefact. Deploy waits for the whole `Test` workflow, so a red
 container gate is a publish that does not happen.
 ## Testing
--- a/photo_pipeline/main.py
+++ b/photo_pipeline/main.py
@@ -1,5 +1,5 @@
 """Application management CLI:
-``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores,backup,verify-backup,restore,diagnostics,benchmark,release-gate,container-gate,dry-run,approve-dry-run}``.
+``python -m photo_pipeline {serve,migrate,worker,import-legacy-scores,backup,verify-backup,restore,diagnostics}``.
 ``serve`` and ``worker`` take the library process lock for their role (US07-05):
 two workers, or the frozen CLI running beside the app, would each be safe on their
@@ -70,14 +70,6 @@ def main(argv: Sequence[str] | None = None) -> int:
        "release-gate", help="Run every suite in an isolated stack and keep the evidence"
    )
    gate_cmd.add_argument("--output", help="Evidence directory (default: data/release/<stamp>)")
    container_cmd = commands.add_parser(
        "container-gate",
        help="Provision the composition from the built image, run the phase_h "
        "acceptance suite against it, destroy it, and keep the evidence (US08-05)",
    )
    container_cmd.add_argument(
        "--output", help="Evidence directory (default: data/container-gate/<stamp>)"
    )
    dry_cmd = commands.add_parser(
        "dry-run", help="Read-only reconciliation of the configured library (US07-07)"
    )
@@ -163,31 +155,6 @@ def main(argv: Sequence[str] | None = None) -> int:
        )
        return 0 if report["ok"] else 1
    if args.command == "container-gate":
        from datetime import datetime, timezone
        from photo_pipeline.services import release
        # The suite provisions and destroys the composition itself; what this command
        # adds is the single entry point and the retained evidence. No skip is an
        # environment limit here: a gate that did not reach the containers proved
        # nothing about them.
        output = args.output or Path(config.data_dir) / "container-gate" / datetime.now(
            timezone.utc
        ).strftime("%Y%m%dT%H%M%SZ")
        report = release.run_gate(
            config,
            output=output,
            stages=release.CONTAINER_STAGES,
            allowed_skips=release.CONTAINER_ALLOWED_SKIP_REASONS,
        )
        print(
            json.dumps(
                {k: v for k, v in report.items() if k not in ("stages", "matrix")}, indent=2
            )
        )
        return 0 if report["ok"] else 1
    if args.command == "dry-run":
        from photo_pipeline.services import release
--- a/photo_pipeline/services/release.py
+++ b/photo_pipeline/services/release.py
@@ -60,17 +60,6 @@ ALLOWED_SKIP_REASONS = (
    "bind-mount ownership is virtualised",
 )
 # The container acceptance gate (US08-05): the deployed application, verified the way
 # the host application is. Deliberately one stage selected by marker, so adding a
 # phase_h test is enough to put it in front of a deploy.
 CONTAINER_STAGES: tuple[tuple[str, tuple[str, ...]], ...] = (
    ("container", ("tests/e2e", "-m", "phase_h")),
 )
 # Nothing. This gate exists to prove the *deployed* runtime, and every reason a check
 # would skip here — no daemon, no compose, no browser — means it was not proven.
 CONTAINER_ALLOWED_SKIP_REASONS: tuple[str, ...] = ()
 class ReleaseError(RuntimeError):
    pass
@@ -186,15 +175,13 @@ def _run_stage(label: str, paths: tuple[str, ...], *, repo: Path, log_dir: Path)
    return StageResult(label, command, result.returncode, elapsed, summary, skipped)
-def unexpected_skips(
+def unexpected_skips(results: list[StageResult]) -> list[str]:
    results: list[StageResult], allowed: tuple[str, ...] = ALLOWED_SKIP_REASONS
 ) -> list[str]:
    """Skips the gate will not accept: everything but the documented environment ones."""
    return [
        line
        for result in results
        for line in result.skipped
-        if not any(reason in line for reason in allowed)
+        if not any(reason in line for reason in ALLOWED_SKIP_REASONS)
    ]
@@ -203,7 +190,6 @@ def run_gate(
    *,
    output: Path | str | None = None,
    stages: tuple[tuple[str, tuple[str, ...]], ...] = STAGES,
    allowed_skips: tuple[str, ...] = ALLOWED_SKIP_REASONS,
    repo: Path | None = None,
 ) -> dict:
    """Run every suite in an isolated stack and retain checksummed evidence.
@@ -220,14 +206,13 @@ def run_gate(
    logs = directory / "logs"
    logs.mkdir(parents=True, exist_ok=True)
    started_at = _now()
    matrix = story_matrix(repo)
    results = [_run_stage(label, paths, repo=repo, log_dir=logs) for label, paths in stages]
-    skips = unexpected_skips(results, allowed_skips)
+    skips = unexpected_skips(results)
    report = {
        "schema_version": SCHEMA_VERSION,
-        "started_at": started_at.isoformat(),
+        "started_at": _now().isoformat(),
        "revision": revision(),
        "python": sys.version.split()[0],
        "platform": os.uname().sysname,
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -24,9 +24,6 @@ test = [
    "httpx>=0.27",
    "playwright>=1.40",
    "pytest-playwright>=0.4",
    # The composition and the workflows are configuration, and the tests that hold
    # them to their promises read them (US08-03, US08-04).
    "pyyaml>=6",
 ]
 [build-system]
@@ -53,5 +50,4 @@ markers = [
    "phase_e: Phase E end-to-end acceptance (US05-06) — upload preflight, uploader, and browser journeys",
    "phase_f: Phase F end-to-end acceptance (US06-06) — archive destination, transfer, and restore journeys",
    "container: builds and runs the container image and its composition (US08-02, US08-03) — needs a Docker daemon, the compose plugin, and the network",
    "phase_h: Phase H container deployment acceptance (US08-05) — the browser, upgrade, restart, and security journeys against the composed stack",
 ]
--- a/tests/e2e/_container_harness.py
+++ b/tests/e2e/_container_harness.py
@@ -1,260 +0,0 @@
 """The composed stack, as a fixture: build, provision, drive, destroy (US08-03/US08-05).
 Extracted from ``tests/e2e/test_compose_stack.py`` when the container acceptance gate
 needed the same stack under a different image, project, and library. One
 implementation, because two would drift on exactly the details that make a container
 test worth anything — the mount, the volume, the ports, and the teardown.
 Nothing here fakes anything below the process boundary: Docker builds the image,
 Compose starts the real containers, and every helper talks to them over HTTP or the
 Docker CLI.
 """
 from __future__ import annotations
 import os
 import shutil
 import socket
 import subprocess
 import tempfile
 import time
 from pathlib import Path
 import httpx
 import pytest
 from PIL import Image
 REPO = Path(__file__).resolve().parents[2]
 COMPOSE_FILE = REPO / "docker-compose.yml"
 CONTAINER_LIBRARY = "/library"
 READY_TIMEOUT_SECONDS = 180
 JOB_TIMEOUT_SECONDS = 300
 UP_TIMEOUT_SECONDS = 30 * 60
 def compose_available() -> bool:
    try:
        return (
            subprocess.run(
                ["docker", "compose", "version"], capture_output=True, timeout=60
            ).returncode
            == 0
        )
    except (OSError, subprocess.SubprocessError):
        return False
 def free_port() -> int:
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]
 def mount_base() -> Path:
    """A directory a Docker VM shares with the host.
    Not pytest's ``tmp_path``: on macOS that is ``/var/folders/...``, which a Docker VM
    (Colima, Docker Desktop) does not share, so the bind mount would arrive empty and
    every assertion would be about nothing. ``$HOME`` is shared by every default
    configuration.
    """
    base = Path(
        os.environ.get("PHOTO_PIPELINE_TEST_MOUNT_BASE", Path.home() / ".cache" / "photo-pipeline")
    )
    base.mkdir(parents=True, exist_ok=True)
    return base
 def temporary_library(albums: dict[str, int], *, prefix: str = "library-") -> Path:
    """A fixture library on a shareable host path, with the exclusion sentinel in it."""
    root = Path(tempfile.mkdtemp(prefix=prefix, dir=mount_base()))
    for position, (album, count) in enumerate(albums.items()):
        (root / album).mkdir(parents=True, exist_ok=True)
        for index in range(count):
            # Distinct per album *and* index: two solid images of the same colour are
            # byte-identical, which would make them a duplicate cluster by accident.
            colour = (17 + 7 * index, 31 + 29 * position, 160 - 3 * index)
            Image.new("RGB", (64, 48), colour).save(root / album / f"{album}_{index}.jpg")
    # Never discovered, counted, analyzed, or uploaded — asserted from outside it.
    (root / "_IGNORE").mkdir(exist_ok=True)
    Image.new("RGB", (32, 32), (0, 0, 0)).save(root / "_IGNORE" / "sentinel.jpg")
    return root
 def remove_library(root: Path) -> None:
    shutil.rmtree(root, ignore_errors=True)
 def write_env_file(path: Path, secret: str, extra: dict[str, str] | None = None) -> Path:
    """Configuration and secrets come from the environment, so a test writes its own
    file rather than borrowing the operator's ``.env``."""
    lines = [
        f"PHOTO_PIPELINE_ACCESS_SECRET={secret}",
        "PHOTO_PIPELINE_LOG_FORMAT=text",
        # The deterministic vision seam, in the data volume so both roles and the test
        # can read it (concept §18).
        "PHOTO_PIPELINE_FAKE_VISION_LOG=/data/vision.log",
        *(f"{key}={value}" for key, value in (extra or {}).items()),
    ]
    path.write_text("\n".join(lines) + "\n")
    return path
 class Stack:
    """The composition under test, plus the environment it was started with."""
    def __init__(
        self,
        library: Path,
        env_file: Path,
        *,
        project: str,
        image: str,
        secret: str,
    ) -> None:
        self.library = library
        self.project = project
        self.secret = secret
        self.port = free_port()
        self.base = f"http://127.0.0.1:{self.port}"
        # Compose reads the repository's own .env for substitution; the process
        # environment wins over it, so the test's values are the ones that apply.
        self.env = {
            **os.environ,
            "PHOTO_PIPELINE_IMAGE": image,
            "PHOTO_PIPELINE_ENV_FILE": str(env_file),
            "PHOTO_PIPELINE_LIBRARY_HOST_PATH": str(library),
            "PHOTO_PIPELINE_LIBRARY_ROOTS": CONTAINER_LIBRARY,
            "PHOTO_PIPELINE_PORT": str(self.port),
            "PHOTO_PIPELINE_UID": str(os.getuid()),
            "PHOTO_PIPELINE_GID": str(os.getgid()),
        }
    # ── the compose lifecycle ────────────────────────────────────────────────
    def compose(self, *args: str, check: bool = True, timeout: int = 300):
        result = subprocess.run(
            ["docker", "compose", "-p", self.project, "-f", str(COMPOSE_FILE), *args],
            capture_output=True,
            text=True,
            env=self.env,
            cwd=REPO,
            timeout=timeout,
        )
        if check and result.returncode != 0:
            raise AssertionError(
                f"docker compose {' '.join(args)} failed:\n{result.stdout}\n{result.stderr}\n"
                f"{self.compose('logs', '--tail', '80', check=False).stdout}"
            )
        return result
    def up(self, *extra: str) -> None:
        self.compose("up", "--detach", *extra, timeout=UP_TIMEOUT_SECONDS)
    def down(self, *, volumes: bool = True) -> None:
        self.compose(
            "down",
            *(("--volumes",) if volumes else ()),
            "--remove-orphans",
            check=False,
            timeout=300,
        )
    def use_image(self, image: str) -> None:
        """Point the composition at another tag — the upgrade path (US08-05)."""
        self.env["PHOTO_PIPELINE_IMAGE"] = image
    def logs(self, *services: str) -> str:
        result = self.compose("logs", *services, check=False)
        return result.stdout + result.stderr
    def wait_until_ready(self) -> None:
        deadline = time.monotonic() + READY_TIMEOUT_SECONDS
        while time.monotonic() < deadline:
            try:
                if httpx.get(f"{self.base}/api/v1/health/ready", timeout=5).status_code == 200:
                    return
            except httpx.HTTPError:
                pass
            time.sleep(0.5)
        raise AssertionError(f"the stack never became ready:\n{self.logs()}")
    # ── talking to it ────────────────────────────────────────────────────────
    def client(self) -> httpx.Client:
        """A browser that has loaded the app: session cookie in the jar, token in a
        header. The session belongs to the API process, so it is re-bootstrapped after
        every restart."""
        client = httpx.Client(base_url=f"{self.base}/api/v1", timeout=60)
        bootstrap = client.get("/session", headers={"X-Access-Secret": self.secret})
        assert bootstrap.status_code == 200, bootstrap.text
        client.headers["X-CSRF-Token"] = bootstrap.json()["csrf_token"]
        return client
 def await_job(client: httpx.Client, job_id: str, states=("succeeded",)) -> dict:
    deadline = time.monotonic() + JOB_TIMEOUT_SECONDS
    snapshot: dict = {}
    while time.monotonic() < deadline:
        response = client.get(f"/jobs/{job_id}")
        if response.status_code == 200:
            snapshot = response.json()
            if snapshot["state"] in states:
                return snapshot
        time.sleep(0.5)
    raise AssertionError(f"job {job_id} never reached {states}: {snapshot}")
 def build_image(tag: str, *, revision: str | None = None) -> str:
    """Build the application image; from a git revision's tree when one is named.
    ``revision`` is how the upgrade journey gets the *previous* version without a
    registry: the tree of that commit is the build context, so what it produces is the
    image that commit would have published.
    """
    if revision is None:
        subprocess.run(
            [
                "docker",
                "build",
                "--build-arg",
                f"UID={os.getuid()}",
                "--build-arg",
                f"GID={os.getgid()}",
                "-t",
                tag,
                str(REPO),
            ],
            check=True,
            timeout=UP_TIMEOUT_SECONDS,
        )
        return tag
    archive = subprocess.run(
        ["git", "archive", "--format=tar", revision],
        cwd=REPO,
        capture_output=True,
        check=True,
        timeout=300,
    ).stdout
    subprocess.run(
        [
            "docker",
            "build",
            "--build-arg",
            f"UID={os.getuid()}",
            "--build-arg",
            f"GID={os.getgid()}",
            "-t",
            tag,
            "-",
        ],
        input=archive,
        check=True,
        timeout=UP_TIMEOUT_SECONDS,
    )
    return tag
 needs_compose = pytest.mark.skipif(
    not compose_available(), reason="no Docker daemon with the compose plugin"
 )
--- a/tests/e2e/test_compose_stack.py
+++ b/tests/e2e/test_compose_stack.py
@@ -10,63 +10,188 @@ what this story is about — the mount, the lock, the volume, and the restart ar
 The file contract (one API, one worker, migrations first, no committed values) is
 checked without a daemon in ``tests/integration/test_compose_runtime.py``; only the
 running proof needs Docker, and CI is where it runs unskipped (US08-04).
 The stack itself lives in ``tests/e2e/_container_harness.py``, shared with the
 container acceptance gate (US08-05).
 """
 from __future__ import annotations
 import json
 import os
 import shutil
 import socket
 import subprocess
 import sys
 import tempfile
 import time
 from pathlib import Path
 import httpx
 import pytest
 from PIL import Image
-from tests.e2e._container_harness import (
+REPO = Path(__file__).resolve().parents[2]
    CONTAINER_LIBRARY,
    Stack,
    await_job,
    compose_available,
    needs_compose,
    remove_library,
    temporary_library,
    write_env_file,
 )
 PROJECT = "photo-pipeline-us0803"
 IMAGE = "photo-pipeline-test:us08-03"
 SECRET = "compose-acceptance-secret"
 CONTAINER_LIBRARY = "/library"
 READY_TIMEOUT_SECONDS = 180
 JOB_TIMEOUT_SECONDS = 180
 UP_TIMEOUT_SECONDS = 30 * 60
 pytestmark = pytest.mark.container
 def compose_available() -> bool:
    try:
        return (
            subprocess.run(
                ["docker", "compose", "version"], capture_output=True, timeout=60
            ).returncode
            == 0
        )
    except (OSError, subprocess.SubprocessError):
        return False
 needs_compose = pytest.mark.skipif(
    not compose_available(), reason="no Docker daemon with the compose plugin"
 )
 def free_port() -> int:
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        return sock.getsockname()[1]
 class Stack:
    """The composition under test, plus the environment it was started with."""
    def __init__(self, library: Path, env_file: Path) -> None:
        self.library = library
        self.port = free_port()
        self.base = f"http://127.0.0.1:{self.port}"
        # Compose reads the repository's own .env for substitution; the process
        # environment wins over it, so the test's values are the ones that apply.
        self.env = {
            **os.environ,
            "PHOTO_PIPELINE_IMAGE": IMAGE,
            "PHOTO_PIPELINE_ENV_FILE": str(env_file),
            "PHOTO_PIPELINE_LIBRARY_HOST_PATH": str(library),
            "PHOTO_PIPELINE_LIBRARY_ROOTS": CONTAINER_LIBRARY,
            "PHOTO_PIPELINE_PORT": str(self.port),
            "PHOTO_PIPELINE_UID": str(os.getuid()),
            "PHOTO_PIPELINE_GID": str(os.getgid()),
        }
    def compose(self, *args: str, check: bool = True, timeout: int = 300):
        result = subprocess.run(
            ["docker", "compose", "-p", PROJECT, "-f", str(REPO / "docker-compose.yml"), *args],
            capture_output=True,
            text=True,
            env=self.env,
            cwd=REPO,
            timeout=timeout,
        )
        if check and result.returncode != 0:
            raise AssertionError(
                f"docker compose {' '.join(args)} failed:\n{result.stdout}\n{result.stderr}\n"
                f"{self.compose('logs', '--tail', '80', check=False).stdout}"
            )
        return result
    def wait_until_ready(self) -> None:
        deadline = time.monotonic() + READY_TIMEOUT_SECONDS
        while time.monotonic() < deadline:
            try:
                if httpx.get(f"{self.base}/api/v1/health/ready", timeout=5).status_code == 200:
                    return
            except httpx.HTTPError:
                pass
            time.sleep(0.5)
        logs = self.compose("logs", "--tail", "120", check=False)
        raise AssertionError(f"the stack never became ready:\n{logs.stdout}\n{logs.stderr}")
    def client(self) -> httpx.Client:
        client = httpx.Client(base_url=f"{self.base}/api/v1", timeout=60)
        bootstrap = client.get("/session", headers={"X-Access-Secret": SECRET})
        assert bootstrap.status_code == 200, bootstrap.text
        client.headers["X-CSRF-Token"] = bootstrap.json()["csrf_token"]
        return client
@pytest.fixture(scope="module")
 def library() -> Path:
-    root = temporary_library({"01_day": 2, "02_night": 1})
+    """A small fixture library on the host, mounted into both containers.
    Not under pytest's ``tmp_path``: on macOS that is ``/var/folders/...``, which a
    Docker VM (Colima, Docker Desktop) does not share, so the bind mount would arrive
    empty and every assertion below would be about nothing. ``$HOME`` is shared by
    every default configuration.
    """
    base = Path(
        os.environ.get("PHOTO_PIPELINE_TEST_MOUNT_BASE", Path.home() / ".cache" / "photo-pipeline")
    )
    base.mkdir(parents=True, exist_ok=True)
    root = Path(tempfile.mkdtemp(prefix="library-", dir=base))
    for album, count in (("01_day", 2), ("02_night", 1)):
        (root / album).mkdir()
        for index in range(count):
            colour = (40 * (index + 1), 90, 160)
            Image.new("RGB", (64, 48), colour).save(root / album / f"{album}_{index}.jpg")
    # The exclusion sentinel: it must never be discovered, counted, or analyzed.
    (root / "_IGNORE").mkdir()
    Image.new("RGB", (32, 32), (0, 0, 0)).save(root / "_IGNORE" / "sentinel.jpg")
    try:
        yield root
    finally:
-        remove_library(root)
+        shutil.rmtree(root, ignore_errors=True)
@pytest.fixture(scope="module")
 def env_file(tmp_path_factory) -> Path:
-    return write_env_file(tmp_path_factory.mktemp("config") / "compose.env", SECRET)
+    """Configuration and secrets come from the environment, so the test writes its
    own file rather than borrowing the operator's."""
    path = tmp_path_factory.mktemp("config") / "compose.env"
    path.write_text(
        "\n".join(
            [
                f"PHOTO_PIPELINE_ACCESS_SECRET={SECRET}",
                "PHOTO_PIPELINE_LOG_FORMAT=text",
                # The deterministic vision seam, in the data volume so both roles and
                # the test can see it (concept §18).
                "PHOTO_PIPELINE_FAKE_VISION_LOG=/data/vision.log",
            ]
        )
        + "\n"
    )
    return path
@pytest.fixture(scope="module")
 def stack(library, env_file):
    if not compose_available():
        pytest.skip("no Docker daemon with the compose plugin")
-    running = Stack(library, env_file, project=PROJECT, image=IMAGE, secret=SECRET)
+    running = Stack(library, env_file)
-    running.down()
+    running.compose("down", "--volumes", "--remove-orphans", check=False)
-    running.up("--build")
+    running.compose("up", "--detach", "--build", timeout=UP_TIMEOUT_SECONDS)
    try:
        running.wait_until_ready()
        yield running
    finally:
-        running.down()
+        running.compose("down", "--volumes", "--remove-orphans", check=False, timeout=300)
 def await_job(client: httpx.Client, job_id: str, states=("succeeded",)) -> dict:
    deadline = time.monotonic() + JOB_TIMEOUT_SECONDS
    snapshot: dict = {}
    while time.monotonic() < deadline:
        response = client.get(f"/jobs/{job_id}")
        if response.status_code == 200:
            snapshot = response.json()
            if snapshot["state"] in states:
                return snapshot
        time.sleep(0.5)
    raise AssertionError(f"job {job_id} never reached {states}: {snapshot}")
 # ── the mounted library ──────────────────────────────────────────────────────
@@ -171,8 +296,8 @@ def test_the_data_volume_survives_recreating_the_containers(stack):
    finally:
        client.close()
-    stack.down(volumes=False)
+    stack.compose("down", "--remove-orphans", timeout=300)
-    stack.up()
+    stack.compose("up", "--detach", timeout=UP_TIMEOUT_SECONDS)
    stack.wait_until_ready()
    client = stack.client()
--- a/tests/e2e/test_phase_h_container.py
+++ b/tests/e2e/test_phase_h_container.py
@@ -1,510 +0,0 @@
 """Phase H — the container acceptance gate (US08-05).
 The deployed application is verified the way the host application is: real image,
 real composition, real browser, real restarts. Everything below runs against
 containers that this suite provisions from the built image, against a temporary
 fixture library on a bind mount and an isolated data volume, and destroys afterwards.
 Four journeys, one per thing a deployment can get wrong:
 * **the browser journey** — discovery, duplicate review, analysis, album proposal,
  rename, upload preflight, and archive, driven through the containerized frontend;
 * **the upgrade journey** — the previous version's image runs first, then this one,
  and the database, its migrations, the rename journal, the job history, and the
  thumbnail cache have to still be there;
 * **the restart journey** — both containers are killed mid-job and the work resumes
  without doing anything twice;
 * **the security gates** — the refusals a loopback deployment made are still made
  behind a published port: no session, forged forwarded headers, a path that leaves
  the mounted library, and a secret in the logs.
 Run it as one command, with evidence: ``python -m photo_pipeline container-gate``.
 """
 from __future__ import annotations
 import collections
 from contextlib import closing
 import json
 import os
 import shutil
 import subprocess
 import sys
 import time
 from pathlib import Path
 import httpx
 import pytest
 from playwright.sync_api import expect
 from photo_pipeline.db import head_revision
 from tests.e2e._container_harness import (
    Stack,
    await_job,
    build_image,
    compose_available,
    needs_compose,
    remove_library,
    temporary_library,
    write_env_file,
 )
 pytestmark = [pytest.mark.phase_h, pytest.mark.container]
 PROJECT = "photo-pipeline-us0805"
 IMAGE = "photo-pipeline-test:us08-05"
 # The version being upgraded *from*. CI passes the tag it last published; without one,
 # the previous commit's tree is built, which is the same claim without a registry.
 PREVIOUS_IMAGE = os.environ.get("PHOTO_PIPELINE_PREVIOUS_IMAGE")
 PREVIOUS_TAG = "photo-pipeline-test:us08-05-previous"
 SECRET = "container-gate-access-secret"
 IMMICH_SENTINEL = "immich-sentinel-9f3a2b"
 HOSTNAME = "photos.test"
 ALBUM = "rome"
 RENAMED = "2019 Rome"
 BURST = 30
 # ── the image and the stacks ─────────────────────────────────────────────────
@pytest.fixture(scope="session")
 def image() -> str:
    if not compose_available():
        pytest.skip("no Docker daemon with the compose plugin")
    return build_image(IMAGE)
@pytest.fixture(scope="module")
 def journey(image, tmp_path_factory):
    """One album, one exact duplicate of a photo in it, and the exclusion sentinel."""
    library = temporary_library({ALBUM: 3}, prefix="us0805-journey-")
    shutil.copyfile(library / ALBUM / f"{ALBUM}_0.jpg", library / ALBUM / "copy.jpg")
    env_file = write_env_file(
        tmp_path_factory.mktemp("journey") / "gate.env",
        SECRET,
        {
            "PHOTO_PIPELINE_ALLOWED_HOSTS": HOSTNAME,
            # Configured but never reachable: the upload view's job here is to show the
            # preflight blockers, and the key's job is to be absent from every log.
            "PHOTO_PIPELINE_IMMICH_API_KEY": IMMICH_SENTINEL,
            "PHOTO_PIPELINE_IMMICH_SERVER_URL": "http://127.0.0.1:1",
        },
    )
    stack = Stack(library, env_file, project=f"{PROJECT}-journey", image=image, secret=SECRET)
    try:
        stack.down()
        stack.up()
        stack.wait_until_ready()
        _seed_journey(stack)
        yield stack
    finally:
        stack.down()
        remove_library(library)
 def _seed_journey(stack: Stack) -> None:
    """Everything the views need, established over the public API before they render."""
    with closing(stack.client()) as client:
        client.post("/inventory/scan").raise_for_status()
        client.post("/duplicates/detect").raise_for_status()
        queue = client.get("/safety/queue", params={"limit": 100}).json()["items"]
        for item in queue:
            decided = client.post(
                "/safety/decisions", json={"asset_id": item["asset_id"], "decision": "sfw"}
            )
            assert decided.status_code == 200, decided.text
        job = client.post("/analysis/jobs").json()
        await_job(client, job["id"])
        client.post("/albums/proposals", json={}).raise_for_status()
        # An archive destination inside the data volume: a second mount would prove
        # nothing more, and the view needs a location to have something to show.
        stack.compose("exec", "-T", "api", "mkdir", "-p", "/data/archive")
        client.post(
            "/archive-locations", json={"name": "external", "root": "/data/archive"}
        ).raise_for_status()
 # ── the browser journey ──────────────────────────────────────────────────────
@needs_compose
 def test_the_browser_journey_covers_every_stage_view_of_the_deployed_app(page, journey):
    """One pass through the deployed frontend, in workflow order.
    The access secret is supplied the way a person supplies it — the app asks, the
    answer is kept for the tab — so what is proven is the authenticated deployment,
    not a test-only bypass.
    """
    page.on("dialog", lambda dialog: dialog.accept(SECRET))
    errors: list[str] = []
    page.on("pageerror", lambda error: errors.append(str(error)))
    base = journey.base
    with closing(journey.client()) as client:
        cluster = client.get("/duplicates/clusters").json()["items"][0]
    # ── discovery ────────────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/workflow")
    page.get_by_test_id("stage-safety").wait_for()
    expect(page.get_by_test_id("stage-analysis")).to_be_visible()
    page.goto(f"{base}/app/#/inventory")
    rows = page.get_by_test_id("asset-row")
    rows.first.wait_for()
    assert rows.count() == 4, "three photos and the duplicate copy; never the sentinel"
    assert "sentinel" not in page.content() and "_IGNORE" not in page.content()
    # ── duplicate review ─────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/duplicates/{cluster['id']}")
    page.get_by_test_id("cluster-state").wait_for()
    # Exact bytes: the cluster arrives decided, and the review surface has to show
    # both members and the evidence the decision was made on.
    expect(page.get_by_test_id("cluster-state")).to_contain_text("decided")
    expect(page.get_by_test_id("member")).to_have_count(2)
    expect(page.get_by_test_id("member").first).to_contain_text("/library/")
    # ── analysis ─────────────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/analyze")
    page.get_by_test_id("analyze-counts").wait_for()
    expect(page.get_by_test_id("run-analysis")).to_be_visible()
    # ── album proposal ───────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/albums?album={ALBUM}")
    page.get_by_test_id("suggested-name").wait_for()
    page.get_by_test_id("final-name").fill(RENAMED)
    page.get_by_test_id("save-name").click()
    page.get_by_test_id("approve").click()
    expect(page.get_by_test_id("proposal-status")).to_contain_text("approved")
    # ── rename, applied against the bind mount ───────────────────────────────
    page.goto(f"{base}/app/#/renames")
    page.get_by_test_id("build-plan").click()
    page.get_by_test_id("operations").wait_for()
    expect(page.get_by_test_id("op-destination").first).to_contain_text(RENAMED)
    page.get_by_test_id("apply-plan").click()
    expect(page.get_by_test_id("apply-result")).to_contain_text("Applied 1, failed 0")
    # The mounted library is the host's directory: the container renamed the operator's
    # folder, not a copy inside its own layer.
    assert (journey.library / RENAMED).is_dir()
    assert not (journey.library / ALBUM).exists()
    # ── upload preflight ─────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/uploads")
    page.get_by_test_id("upload-scope").wait_for()
    expect(page.get_by_test_id("album-row").first).to_be_visible()
    assert IMMICH_SENTINEL not in page.content(), "the API key never reaches the browser"
    # ── archive ──────────────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/archive")
    page.get_by_test_id("archive-locations").wait_for()
    expect(page.get_by_test_id("location-row").first).to_contain_text("external")
    assert errors == [], f"the deployed frontend raised page errors: {errors}"
 # ── the upgrade journey ──────────────────────────────────────────────────────
@pytest.fixture(scope="module")
 def previous_image(image) -> str:
    """The image the deployment is upgrading *from*.
    The published tag when there is one. Before the first publish there is nothing to
    pull, and refusing then would mean the gate could never let the first deploy
    through — so the previous commit's tree is built instead, which is the same claim
    without a registry.
    """
    if PREVIOUS_IMAGE:
        pulled = subprocess.run(
            ["docker", "pull", PREVIOUS_IMAGE], capture_output=True, timeout=1800
        )
        if pulled.returncode == 0:
            return PREVIOUS_IMAGE
    return build_image(PREVIOUS_TAG, revision="HEAD~1")
@needs_compose
 def test_an_upgrade_keeps_the_database_the_journal_the_jobs_and_the_cache(
    image, previous_image, tmp_path_factory
 ):
    library = temporary_library({ALBUM: 2}, prefix="us0805-upgrade-")
    env_file = write_env_file(tmp_path_factory.mktemp("upgrade") / "gate.env", SECRET)
    stack = Stack(
        library, env_file, project=f"{PROJECT}-upgrade", image=previous_image, secret=SECRET
    )
    try:
        stack.down()
        stack.up()
        stack.wait_until_ready()
        # ── what the previous version leaves behind ──────────────────────────
        with closing(stack.client()) as client:
            client.post("/inventory/scan").raise_for_status()
            assets = client.get("/inventory/assets", params={"limit": 200}).json()["items"]
            for item in client.get("/safety/queue", params={"limit": 100}).json()["items"]:
                client.post(
                    "/safety/decisions", json={"asset_id": item["asset_id"], "decision": "sfw"}
                ).raise_for_status()
            job = client.post("/analysis/jobs").json()
            finished = await_job(client, job["id"])
            for asset in assets:  # populate the thumbnail cache
                thumbnail = client.get(f"/assets/{asset['id']}/thumbnail", params={"size": 256})
                assert thumbnail.status_code == 200, thumbnail.text
            # A rename plan, left unapplied: the journal has to survive the upgrade
            # exactly as it was, or a half-applied one could not be recovered.
            client.post("/albums/proposals", json={}).raise_for_status()
            _approve(client, RENAMED)
            plan = client.post("/rename-plans").json()
        before = {
            "assets": sorted(asset["id"] for asset in assets),
            "job": finished["progress"],
            "plan": (plan["id"], plan["checksum"]),
            "thumbnails": _cache_files(stack),
            "revision": _revision(stack),
        }
        assert before["thumbnails"], "no thumbnail was cached, so nothing would be proven"
        # ── the upgrade: same volume, new image ──────────────────────────────
        stack.down(volumes=False)
        stack.use_image(image)
        stack.up()
        stack.wait_until_ready()
        after_revision = _revision(stack)
        assert after_revision == head_revision(), "the new image did not migrate the volume"
        if after_revision != before["revision"]:
            # A schema change is snapshotted before it is applied (US07-05), so a
            # failed upgrade is restorable rather than a lost library.
            assert _ls(stack, "/data/backups"), "a migration ran without a backup"
        with closing(stack.client()) as client:
            assets = client.get("/inventory/assets", params={"limit": 200}).json()["items"]
            assert sorted(asset["id"] for asset in assets) == before["assets"]
            assert client.get(f"/jobs/{job['id']}").json()["progress"] == before["job"]
            plans = client.get("/rename-plans").json()["items"]
            assert (plans[0]["id"], plans[0]["checksum"]) == before["plan"]
            for asset in assets:
                assert (
                    client.get(f"/analysis/results/{asset['id']}").status_code == 200
                ), "an analysis result did not survive the upgrade"
        # The cache is keyed by pixel hash and thumbnail version, so an upgrade that
        # kept the volume must keep the files: regenerating them is work nobody asked
        # for, and losing them silently is how a cache stops being one.
        assert set(before["thumbnails"]) <= set(_cache_files(stack))
    finally:
        stack.down()
        remove_library(library)
 def _approve(client: httpx.Client, name: str, *, album: str = ALBUM) -> None:
    for payload, route in (({"name": name}, "edit"), ({}, "approve")):
        current = client.get(f"/albums/proposals/{album}").json()
        client.post(
            f"/albums/proposals/{album}/{route}",
            json={**payload, "expected_version": current["version"]},
        ).raise_for_status()
 def _exec(stack: Stack, *args: str) -> str:
    return stack.compose("exec", "-T", "api", *args).stdout
 def _ls(stack: Stack, directory: str) -> list[str]:
    listing = stack.compose("exec", "-T", "api", "ls", directory, check=False)
    return [line for line in listing.stdout.split() if line]
 def _cache_files(stack: Stack) -> list[str]:
    return sorted(
        _exec(stack, "find", "/data/cache", "-type", "f", "-name", "*.webp").split()
    )
 def _revision(stack: Stack) -> str:
    """The schema revision the volume's database is actually at."""
    return _exec(
        stack,
        "python",
        "-c",
        "import sqlite3;print(sqlite3.connect('/data/photo_pipeline.db')"
        ".execute('select version_num from alembic_version').fetchone()[0])",
    ).strip()
 # ── the restart journey ──────────────────────────────────────────────────────
@needs_compose
 def test_killing_both_containers_mid_job_resumes_without_doing_anything_twice(
    image, tmp_path_factory
 ):
    """`docker kill` is the honest restart: no grace period, no orderly stop, no
    chance for either process to write a tidy final state."""
    library = temporary_library({"burst": BURST}, prefix="us0805-restart-")
    env_file = write_env_file(tmp_path_factory.mktemp("restart") / "gate.env", SECRET)
    stack = Stack(library, env_file, project=f"{PROJECT}-restart", image=image, secret=SECRET)
    try:
        stack.down()
        stack.up()
        stack.wait_until_ready()
        with closing(stack.client()) as client:
            client.post("/inventory/scan").raise_for_status()
            assets = client.get("/inventory/assets", params={"limit": 200}).json()["items"]
            assert len(assets) == BURST
            for item in client.get("/safety/queue", params={"limit": 100}).json()["items"]:
                client.post(
                    "/safety/decisions", json={"asset_id": item["asset_id"], "decision": "sfw"}
                ).raise_for_status()
            job = client.post("/analysis/jobs").json()
            progress = _wait_for_progress(client, job["id"])
        assert 0 < progress["done"] < progress["total"], progress
        stack.compose("kill", "api", "worker")
        stack.up()
        stack.wait_until_ready()
        with closing(stack.client()) as client:
            finished = await_job(client, job["id"])
            assert finished["progress"]["done"] == BURST, finished
            results = [
                client.get(f"/analysis/results/{asset['id']}").json() for asset in assets
            ]
        # Exactly one stored result per asset: at-least-once execution, idempotent
        # recovery — a retried item overwrites its own attempt, it does not add one.
        assert len(results) == BURST
        assert all(result["description"] for result in results)
        # And the side effect nobody can take back — the call to the provider — happened
        # again only for whatever was in flight when the containers died.
        analysed = collections.Counter(_exec(stack, "cat", "/data/vision.log").split())
        assert len(analysed) == BURST, "every photo was analysed, and only the library's"
        assert max(analysed.values()) <= 2, dict(analysed)
        assert sum(1 for count in analysed.values() if count > 1) <= 1, dict(analysed)
    finally:
        stack.down()
        remove_library(library)
 def _wait_for_progress(client: httpx.Client, job_id: str, *, timeout: float = 120) -> dict:
    """Wait until the job is provably under way but provably unfinished."""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        snapshot = client.get(f"/jobs/{job_id}").json()
        progress = snapshot["progress"]
        if progress["done"] and progress["done"] < progress["total"]:
            return progress
        if snapshot["state"] in ("succeeded", "failed"):
            raise AssertionError(f"the job finished before it could be interrupted: {snapshot}")
        time.sleep(0.05)
    raise AssertionError(f"the job never started: {client.get(f'/jobs/{job_id}').json()}")
 # ── the security gates ───────────────────────────────────────────────────────
@needs_compose
 def test_the_deployed_instance_refuses_a_caller_without_a_session(journey):
    with httpx.Client(base_url=f"{journey.base}/api/v1", timeout=30) as client:
        for method, path in (("GET", "/workflow"), ("POST", "/inventory/scan")):
            response = client.request(method, path)
            assert response.status_code == 401, path
            assert response.json()["error"]["code"] == "unauthenticated"
        # A session still has to be paid for with the operator's secret.
        assert client.get("/session", headers={"X-Access-Secret": "guessed"}).status_code == 401
        # Readiness stays open: the orchestrator's health check holds no session.
        assert client.get("/health/ready").status_code == 200
@needs_compose
 def test_forged_forwarded_headers_cannot_smuggle_an_allowed_host_past_the_check(journey):
    """Behind a proxy the app believes ``X-Forwarded-*`` — but only from the proxy.
    Nothing in this composition is a trusted proxy, so the claim is the client's."""
    with httpx.Client(base_url=f"{journey.base}/api/v1", timeout=30) as client:
        client.headers["X-CSRF-Token"] = (
            client.get("/session", headers={"X-Access-Secret": SECRET}).json()["csrf_token"]
        )
        # The hostname this deployment is reached under is accepted.
        assert client.get("/workflow", headers={"Host": HOSTNAME}).status_code == 200
        forged = client.get(
            "/workflow",
            headers={"Host": "photos.evil.example", "X-Forwarded-Host": HOSTNAME},
        )
        assert forged.status_code == 403
        assert forged.json()["error"]["code"] == "host_not_allowed"
        # A forged protocol claim must not mark the session cookie as HTTPS-only
        # either — that would strand the operator's real, plain-HTTP session.
        bootstrap = httpx.get(
            f"{journey.base}/api/v1/session",
            headers={"X-Access-Secret": SECRET, "X-Forwarded-Proto": "https"},
            timeout=30,
        )
        assert "secure" not in bootstrap.headers["set-cookie"].lower()
@needs_compose
 def test_a_path_that_leaves_the_mounted_library_is_refused(journey):
    """The container's own filesystem is not the library. A symlink swapped under a
    known asset is the sharpest version of the question, because the database still
    points at a path inside the mount."""
    with closing(journey.client()) as client:
        asset = client.get("/inventory/assets", params={"limit": 200}).json()["items"][0]
        original = journey.library / Path(asset["current_path"]).relative_to("/library")
        kept = original.read_bytes()
        original.unlink()
        original.symlink_to("/etc/passwd")
        try:
            escaped = client.get(f"/assets/{asset['id']}/thumbnail", params={"size": 256})
        finally:
            original.unlink()
            original.write_bytes(kept)
        assert escaped.status_code == 403
        assert escaped.json()["error"]["code"] == "path_not_allowed"
        assert "root:" not in escaped.text
        # And a root that names nothing mounted is refused before the process serves.
        refused = journey.compose(
            "run",
            "--rm",
            "--no-deps",
            "--env",
            "PHOTO_PIPELINE_LIBRARY_ROOTS=/srv/photos",
            "api",
            "serve",
            check=False,
        )
        assert refused.returncode == 5, refused.stdout + refused.stderr
@needs_compose
 def test_no_secret_reaches_the_container_logs(journey):
    with httpx.Client(base_url=f"{journey.base}/api/v1", timeout=30) as client:
        client.get("/session", headers={"X-Access-Secret": "wrong-secret-attempt"})
        client.get("/session", headers={"X-Access-Secret": SECRET})
    logs = journey.logs()
    assert SECRET not in logs
    assert IMMICH_SENTINEL not in logs
    assert "wrong-secret-attempt" not in logs, "a rejected secret is still a secret"
    # The refusal itself is logged, so an operator can see the attempt.
    assert "access secret rejected" in logs
@needs_compose
 def test_the_evidence_of_this_run_names_the_stack_it_was_produced_from(journey):
    """A gate that cannot say what it ran against is an opinion. `docker compose ps`
    is the record: one API, one worker, one completed migration."""
    listing = [
        json.loads(line)
        for line in journey.compose("ps", "--all", "--format", "json").stdout.splitlines()
        if line.strip()
    ]
    services = {entry["Service"]: entry["State"] for entry in listing}
    assert services["api"] == "running" and services["worker"] == "running"
    assert services["migrate"] == "exited"
 if __name__ == "__main__":  # a quick way to run just this file
    raise SystemExit(pytest.main([__file__, "-v", *sys.argv[1:]]))
--- a/tests/integration/test_container_gate.py
+++ b/tests/integration/test_container_gate.py
@@ -1,147 +0,0 @@
 """US08-05: the container acceptance gate's own contract, checked without a daemon.
 The gate itself needs Docker, a browser, and several minutes; running it from inside
 the suite would be a fork bomb with better manners. What is checkable offline is what
 makes it a *gate* rather than a long test run:
 * it selects the container journeys by marker, so adding one is enough to put it in
  front of a deploy;
 * it accepts no skip at all — every reason a check would skip here (no daemon, no
  compose plugin, no browser) means the deployed runtime was not proven;
 * it retains checksummed evidence per run;
 * and CI runs it on `main`, which is what the publish step waits for.
 The running proof is ``tests/e2e/test_phase_h_container.py``.
 """
 from __future__ import annotations
 import json
 import re
 from pathlib import Path
 import yaml
 from photo_pipeline.config import Config
 from photo_pipeline.services import release
 REPO = Path(__file__).resolve().parents[2]
 SUITE = REPO / "tests" / "e2e" / "test_phase_h_container.py"
 TEST_WORKFLOW = yaml.safe_load((REPO / ".gitea" / "workflows" / "test.yml").read_text())
 def _config(tmp_path) -> Config:
    for name in ("data", "lib"):
        (tmp_path / name).mkdir(exist_ok=True)
    return Config.from_env(
        {
            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(tmp_path / "lib"),
        }
    )
 # ── what the gate runs ───────────────────────────────────────────────────────
 def test_the_gate_selects_the_container_journeys_by_their_marker():
    assert release.CONTAINER_STAGES == (("container", ("tests/e2e", "-m", "phase_h")),)
    declared = [line for line in (REPO / "pyproject.toml").read_text().splitlines()
                if line.strip().startswith('"phase_h:')]
    assert declared, "an unregistered marker selects nothing and fails no gate"
 def test_every_journey_in_the_suite_carries_the_marker():
    """A test in that file without the marker is a check the gate never runs."""
    source = SUITE.read_text()
    assert "pytestmark = [pytest.mark.phase_h, pytest.mark.container]" in source
    journeys = re.findall(r"^def (test_[a-z_]+)", source, re.MULTILINE)
    assert len(journeys) >= 4, journeys
    for required in ("browser", "upgrade", "kill", "secret"):
        assert any(required in name for name in journeys), (required, journeys)
 # ── no skip is an environment limit here ─────────────────────────────────────
 def test_the_gate_accepts_no_skipped_check_at_all():
    assert release.CONTAINER_ALLOWED_SKIP_REASONS == ()
    docker_missing = release.StageResult(
        "container", [], 0, 0.1, "1 skipped", ["SKIPPED [1] x.py:1: no Docker daemon available"]
    )
    # The release gate tolerates exactly this one, because the image definition is
    # still checked offline. The container gate cannot: it is the deployment's proof.
    assert release.unexpected_skips([docker_missing]) == []
    assert release.unexpected_skips(
        [docker_missing], release.CONTAINER_ALLOWED_SKIP_REASONS
    ) == ["SKIPPED [1] x.py:1: no Docker daemon available"]
 def test_a_skipped_check_fails_the_run_and_the_evidence_says_so(tmp_path):
    skipping = tmp_path / "test_skipping.py"
    skipping.write_text(
        "import pytest\n\n"
        "def test_x():\n"
        "    pytest.skip('no Docker daemon available')\n"
    )
    evidence = tmp_path / "evidence"
    report = release.run_gate(
        _config(tmp_path),
        output=evidence,
        stages=(("container", (str(skipping),)),),
        allowed_skips=release.CONTAINER_ALLOWED_SKIP_REASONS,
    )
    assert report["ok"] is False
    assert report["failures"] == [], "the stage passed; the skip is what fails the gate"
    assert report["unexpected_skips"], report
    written = json.loads((evidence / release.REPORT_NAME).read_text())
    assert written["ok"] is False
    assert (evidence / "logs" / "container.log").exists()
    for line in (evidence / release.CHECKSUMS_NAME).read_text().splitlines():
        digest, name = line.split("  ", 1)
        assert release.sha256_file(evidence / name) == digest
 def test_the_same_run_passes_when_nothing_skips(tmp_path):
    passing = tmp_path / "test_passing.py"
    passing.write_text("def test_x():\n    assert True\n")
    report = release.run_gate(
        _config(tmp_path),
        output=tmp_path / "evidence",
        stages=(("container", (str(passing),)),),
        allowed_skips=release.CONTAINER_ALLOWED_SKIP_REASONS,
    )
    assert report["ok"] is True and report["unexpected_skips"] == []
    assert report["revision"], "the evidence must say which commit it covers"
 # ── the command, and CI ──────────────────────────────────────────────────────
 def test_the_command_is_documented_and_wired():
    from photo_pipeline.__main__ import main  # noqa: F401  (import proves it loads)
    assert "container-gate" in (REPO / "photo_pipeline" / "__main__.py").read_text()
    assert "container-gate" in (REPO / "README.md").read_text()
 def test_ci_runs_the_gate_on_main_and_keeps_its_evidence():
    job = TEST_WORKFLOW["jobs"]["container"]
    assert job["if"] == "gitea.event_name == 'push'", "pull requests have nothing to upgrade from"
    script = "\n".join(step["run"] for step in job["steps"] if "run" in step)
    assert "photo_pipeline container-gate" in script
    evidence = next(step for step in job["steps"] if "upload-artifact" in str(step.get("uses")))
    assert evidence["if"] == "always()", "a failed gate's logs are the ones worth keeping"
    assert evidence["with"]["path"] == "gate-evidence"
 def test_the_upgrade_journey_can_still_reach_the_previous_version():
    """Without depth, `git archive HEAD~1` has nothing to build."""
    checkout = next(
        step for step in TEST_WORKFLOW["jobs"]["container"]["steps"] if "checkout" in str(step.get("uses"))
    )
    assert checkout["with"]["fetch-depth"] >= 2
--- a/tests/integration/test_deploy_workflows.py
+++ b/tests/integration/test_deploy_workflows.py
@@ -1,210 +0,0 @@
 """US08-04: the deployment pipeline's contract, read from the workflow files.
 A CI/CD pipeline is the one piece of this repository whose failure mode is silence.
 A renamed workflow, a renamed job, a trigger that quietly stopped matching, and the
 suite is no longer between `main` and the registry — with nothing red anywhere to say
 so. So the link is asserted from both ends here: the deploy workflow waits for the
 test workflow *by name*, and that name is read out of the test workflow itself.
 The same applies to the commands: `work_item/.work-item.yml` is where the required
 suites are configured, and the test workflow has to run those, not a copy of them that
 was true once.
 Nothing here talks to Gitea, the registry, or Portainer. What is checkable offline is
 what the files promise; whether the runner honours them is the deployment's own
 evidence (US08-05).
 """
 from __future__ import annotations
 import re
 from pathlib import Path
 import yaml
 REPO = Path(__file__).resolve().parents[2]
 WORKFLOWS = REPO / ".gitea" / "workflows"
 TEST_FILE = WORKFLOWS / "test.yml"
 DEPLOY_FILE = WORKFLOWS / "deploy.yml"
 TEST_WORKFLOW = yaml.safe_load(TEST_FILE.read_text())
 DEPLOY_WORKFLOW = yaml.safe_load(DEPLOY_FILE.read_text())
 WORK_ITEM = yaml.safe_load((REPO / "work_item" / ".work-item.yml").read_text())
 REGISTRY = "git.domverse-berlin.eu"
 def triggers(workflow: dict) -> dict:
    """The `on:` block. PyYAML resolves the bare key `on` to ``True``."""
    return workflow[True] if True in workflow else workflow["on"]
 def steps(workflow: dict, job: str) -> list[dict]:
    return workflow["jobs"][job]["steps"]
 def run_script(workflow: dict, job: str) -> str:
    """Every `run:` in a job, as one blob — what the runner would execute."""
    return "\n".join(step["run"] for step in steps(workflow, job) if "run" in step)
 # ── the gate ─────────────────────────────────────────────────────────────────
 def test_the_test_workflow_runs_on_pull_requests_and_on_main():
    on = triggers(TEST_WORKFLOW)
    assert "pull_request" in on
    assert on["push"]["branches"] == ["main"]
 def test_the_test_workflow_runs_the_configured_required_suites():
    required = WORK_ITEM["workflow"]["required_tests"]
    assert required, "the helper's required tests are what CI exists to run"
    script = run_script(TEST_WORKFLOW, "suites")
    for command in required:
        assert command in script, f"CI does not run the configured suite: {command}"
 def test_the_deploy_workflow_waits_for_the_test_workflow_by_its_real_name():
    # Renaming either side without the other breaks this, which is the point: an
    # unmatched `workflow_run` filter does not fail, it simply never fires.
    awaited = triggers(DEPLOY_WORKFLOW)["workflow_run"]
    assert awaited["workflows"] == [TEST_WORKFLOW["name"]]
    assert awaited["types"] == ["completed"]
    assert awaited["branches"] == ["main"], "only main is deployable"
 def test_publishing_requires_the_test_run_to_have_succeeded_and_not_merely_finished():
    condition = DEPLOY_WORKFLOW["jobs"]["publish"]["if"]
    assert "workflow_run.conclusion == 'success'" in condition
    # `completed` includes failure and cancellation; the conclusion check is the gate.
    assert "gitea.event_name == 'workflow_run'" in condition
 def test_the_deploy_workflow_can_also_be_dispatched_by_hand():
    assert "workflow_dispatch" in triggers(DEPLOY_WORKFLOW)
 # ── what is published ────────────────────────────────────────────────────────
 def test_the_image_is_published_under_this_project_s_own_path():
    slug = WORK_ITEM["repository"]["slug"]
    assert DEPLOY_WORKFLOW["env"]["IMAGE"] == f"{REGISTRY}/{slug}"
 def test_both_latest_and_the_commit_are_pushed_so_a_rollback_is_a_tag_change():
    script = run_script(DEPLOY_WORKFLOW, "publish")
    assert 'docker push "$IMAGE:$SHA"' in script
    assert 'docker push "$IMAGE:latest"' in script
    # A `latest` without its commit tag would be a version that cannot be rolled back to.
    assert script.index('docker push "$IMAGE:$SHA"') < script.index('docker push "$IMAGE:latest"')
 def test_the_published_commit_is_the_one_that_was_tested():
    assert DEPLOY_WORKFLOW["env"]["SHA"].startswith("${{ gitea.event.workflow_run.head_sha")
    checkout = steps(DEPLOY_WORKFLOW, "publish")[0]
    assert checkout["with"]["ref"] == "${{ env.SHA }}"
 # ── secrets ──────────────────────────────────────────────────────────────────
 def test_registry_credentials_and_the_webhook_come_from_repository_secrets():
    script = run_script(DEPLOY_WORKFLOW, "publish")
    for secret in ("REGISTRY_USER", "REGISTRY_TOKEN", "PORTAINER_WEBHOOK_URL"):
        assert f"secrets.{secret}" in script
 def test_no_runtime_secret_is_named_by_the_pipeline_at_all():
    # The vision key, the Immich key, and the access secret belong to the Portainer
    # stack. A workflow that mentions one is a workflow that could carry one.
    text = TEST_FILE.read_text() + DEPLOY_FILE.read_text()
    for runtime_secret in (
        "PHOTO_PIPELINE_VISION_API_KEY",
        "PHOTO_PIPELINE_IMMICH_API_KEY",
        "PHOTO_PIPELINE_ACCESS_SECRET",
    ):
        assert runtime_secret not in text
 def test_the_workflows_carry_no_credential_values():
    text = TEST_FILE.read_text() + DEPLOY_FILE.read_text()
    for line in text.splitlines():
        if re.search(r"(?i)(token|secret|password|api[_-]?key)\s*[:=]", line):
            assert "${{ secrets." in line or line.lstrip().startswith("#"), line
 def test_the_registry_password_is_never_an_argument():
    # An argument is visible in the runner's process list and in `docker login`'s own
    # warning; stdin is not.
    script = run_script(DEPLOY_WORKFLOW, "publish") + run_script(DEPLOY_WORKFLOW, "dry-run")
    assert "--password-stdin" in script
    assert "--password " not in script
 # ── redeploy, pruning, and overlap ───────────────────────────────────────────
 def test_a_failed_webhook_call_fails_the_workflow():
    webhook = next(
        step
        for step in steps(DEPLOY_WORKFLOW, "publish")
        if "PORTAINER_WEBHOOK_URL" in step.get("run", "")
    )
    assert "--fail" in webhook["run"], "curl exits 0 on an HTTP 500 without it"
    assert "-k" not in webhook["run"].split(), "the redeploy call verifies TLS"
 def test_pruning_removes_dangling_layers_and_never_published_tags():
    for job in ("publish", "dry-run"):
        script = run_script(DEPLOY_WORKFLOW, job)
        assert "docker image prune -f" in script
        assert "prune -a" not in script and "--all" not in script
 def test_two_deploys_of_different_commits_cannot_overlap():
    guard = DEPLOY_WORKFLOW["concurrency"]
    assert "sha" not in guard["group"].lower(), "a per-commit group guards nothing"
    assert guard["cancel-in-progress"] is False, "a cancelled push leaves half a tag set"
 # ── the dry run ──────────────────────────────────────────────────────────────
 def test_the_dry_run_is_manual_only_and_defaults_to_on():
    condition = DEPLOY_WORKFLOW["jobs"]["dry-run"]["if"]
    assert condition == "gitea.event_name == 'workflow_dispatch' && inputs.dry_run"
    dispatch = triggers(DEPLOY_WORKFLOW)["workflow_dispatch"]
    assert dispatch["inputs"]["dry_run"]["default"] is True
    assert dispatch["inputs"]["dry_run"]["type"] == "boolean"
 def test_the_dry_run_publishes_a_scratch_tag_and_nothing_else():
    script = run_script(DEPLOY_WORKFLOW, "dry-run")
    assert 'docker push "$IMAGE:scratch-$SHA"' in script
    assert ":latest" not in script, "a dry run that moves latest is a deploy"
    assert "PORTAINER_WEBHOOK_URL" not in script, "a dry run never redeploys the stack"
 def test_a_manual_deploy_is_the_deliberate_choice_not_the_default():
    publish = DEPLOY_WORKFLOW["jobs"]["publish"]["if"]
    assert "inputs.dry_run == false" in publish
 # ── the files themselves ─────────────────────────────────────────────────────
 def test_every_workflow_file_parses_and_declares_a_name_a_trigger_and_a_job():
    files = sorted(WORKFLOWS.glob("*.yml")) + sorted(WORKFLOWS.glob("*.yaml"))
    assert {path.name for path in files} == {"test.yml", "deploy.yml"}
    for path in files:
        workflow = yaml.safe_load(path.read_text())
        assert workflow["name"], path
        assert triggers(workflow), path
        assert workflow["jobs"], path
        for job_name, job in workflow["jobs"].items():
            assert job["runs-on"], f"{path}:{job_name}"
            assert job["steps"], f"{path}:{job_name}"
            for step in job["steps"]:
                assert step.get("name"), f"{path}:{job_name} has an unnamed step"
--- a/tests/story_traceability.json
+++ b/tests/story_traceability.json
@@ -186,15 +186,11 @@
    "US08-03": [
      "tests/integration/test_compose_runtime.py",
      "tests/e2e/test_compose_stack.py"
    ],
    "US08-04": [
      "tests/integration/test_deploy_workflows.py"
    ],
    "US08-05": [
      "tests/integration/test_container_gate.py",
      "tests/e2e/test_phase_h_container.py"
    ]
  },
-  "planned": [],
+  "planned": [
    "US08-04",
    "US08-05"
  ],
  "_planned_comment": "Accepted backlog stories that are not implemented yet. The release gate (US07-07) requires every story file to be either mapped to tests or listed here, so an unimplemented story is a visible decision rather than a hole in the matrix."
 }