diff --git a/.github/actions/atelier/action.yaml b/.github/actions/atelier/action.yaml
index 55c8411..6900472 100644
--- a/.github/actions/atelier/action.yaml
+++ b/.github/actions/atelier/action.yaml
@@ -23,13 +23,30 @@ inputs:
   rules:
     description: Atelier rule file whose substituters and trusted-public-keys are injected into nix.conf
     default: atelier.toml
+  post-build-hook:
+    description: Install the spooling post-build hook and name it in nix.conf
+    default: "false"
 
 runs:
   using: composite
   steps:
+    - name: Export Atelier Root
+      # publish the action checkout's repo root so later workflow steps can
+      # invoke files shipped with atelier (the runner downloads the whole
+      # repository for a subdirectory action)
+      shell: bash
+      run: | # zizmor: ignore[github-env] value is the runner-provided action path, not consumer input
+        # a failed strip must collapse to the empty value the push step guards
+        root="${GITHUB_ACTION_PATH%/.github/actions/atelier}"
+        if [ ! -f "${root}/src/atelier/stream.py" ]; then
+          echo "::warning::Atelier root not found under ${GITHUB_ACTION_PATH}, streaming disabled"
+          root=""
+        fi
+        printf 'ATELIER_ROOT=%s\n' "${root}" >> "$GITHUB_ENV"
+
     - name: Pre Install Hook
       # consumer-supplied command, passed via env (not interpolated into the
-      # script body) so it cannot break out of the run. it is the first step and
+      # script body) so it cannot break out of the run. it is the first consumer-visible step and
       # carries no install-command guard, so it fires before the built-in or a
       # custom installer alike, on a fresh runner before disk reclaim
       if: ${{ inputs.pre-install != '' }}
@@ -81,6 +98,7 @@ runs:
         SANDBOX: ${{ runner.os == 'macOS' && 'relaxed' || 'true' }}
         EXTRA_CONF: ${{ inputs.extra-conf }}
         RULES: ${{ inputs.rules }}
+        POST_BUILD_HOOK: ${{ inputs.post-build-hook }}
       run: |
         set -euo pipefail
         # the nix.conf lines that apply whichever installer ran, so a custom
@@ -113,6 +131,10 @@ runs:
           if [ -n "${keys}" ]; then
             echo "extra-trusted-public-keys = ${keys}"
           fi
+          # named before extra-conf so a consumer override wins (last wins)
+          if [ "${POST_BUILD_HOOK}" = "true" ]; then
+            echo "post-build-hook = /nix/var/atelier/hook.sh"
+          fi
           # explicit if, not `[ -n ] && printf`, so an empty value does not trip
           # set -e on the trailing conditional
           if [ -n "${EXTRA_CONF}" ]; then
@@ -178,6 +200,24 @@ runs:
       shell: bash
       run: sudo mkdir -p /nix/build
 
+    - name: Install Post Build Hook
+      # after the installer because /nix does not exist on macos before it
+      # creates the volume. no build can run between daemon start and this
+      # step (the steps in between invoke no nix commands and the consumer
+      # post-install hook runs later), so the hook path named in nix.conf
+      # always resolves by the time a build finishes
+      if: ${{ inputs.post-build-hook == 'true' }}
+      shell: bash
+      run: |
+        set -euo pipefail
+        # explicit modes, sudo inherits the caller's umask
+        # install -m /dev/null also truncates a spool surviving on a
+        # persistent /nix so old jobs' paths are not replayed
+        # the paths must match hook.sh and stream.py's ATELIER_SPOOL default
+        sudo mkdir -p -m 0755 /nix/var/atelier
+        sudo install -m 0755 "${GITHUB_ACTION_PATH}/hook.sh" /nix/var/atelier/hook.sh
+        sudo install -m 0644 /dev/null /nix/var/atelier/spool
+
     - name: Report Disk Space (Darwin)
       if: ${{ inputs.reclaim == 'true' && runner.os == 'macOS' }}
       uses: srz-zumix/post-run-action@42756f7452b9439d0365b7e087b2c364f54209c6 # v3.0.2
diff --git a/.github/actions/atelier/hook.sh b/.github/actions/atelier/hook.sh
new file mode 100755
index 0000000..52016e4
--- /dev/null
+++ b/.github/actions/atelier/hook.sh
@@ -0,0 +1,9 @@
+#!/bin/sh
+# post-build hook, runs as root under the nix daemon
+# append every built output path to the spool and never fail the build loop
+# the env override exists for tests, the daemon never sets ATELIER_SPOOL
+# OUT_PATHS is deliberately unquoted so word splitting yields one path per line
+# set -f disables globbing so a metacharacter in a path never expands
+# an empty OUT_PATHS writes a blank line, the spool reader skips blank lines
+set -f
+{ printf '%s\n' $OUT_PATHS >> "${ATELIER_SPOOL:-/nix/var/atelier/spool}"; } 2>/dev/null || true
diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml
index 55cc7f9..5e71707 100644
--- a/.github/workflows/build.yaml
+++ b/.github/workflows/build.yaml
@@ -117,8 +117,41 @@ jobs:
           rules: ${{ inputs.rules }}
           pre-install: ${{ inputs.pre-install }}
           post-install: ${{ inputs.post-install }}
+          post-build-hook: ${{ inputs.push && inputs.push-command == '' && github.event.pull_request.head.repo.fork != true && ((vars.ATTIC_SERVER != '' && vars.ATTIC_CACHE != '') || vars.CACHIX_CACHE != '' || vars.NIKS3_SERVER != '') }}
+
+      # the streamer runs as the runner user with the step env, which is what
+      # keeps the secrets out of the root-run hook. it drains the spool the
+      # hook fills and pushes while the build runs. its stdout lands in a log
+      # file the final drain replays into the step log
+      - name: Start Cache Streamer
+        if: ${{ inputs.push && inputs.push-command == '' && matrix.installable != '' && github.event.pull_request.head.repo.fork != true && ((vars.ATTIC_SERVER != '' && vars.ATTIC_CACHE != '') || vars.CACHIX_CACHE != '' || vars.NIKS3_SERVER != '') }}
+        shell: bash
+        env:
+          ATTIC_SERVER: ${{ vars.ATTIC_SERVER }}
+          ATTIC_CACHE: ${{ vars.ATTIC_CACHE }}
+          ATTIC_TOKEN: ${{ secrets.ATTIC_TOKEN }}
+          CACHIX_CACHE: ${{ vars.CACHIX_CACHE }}
+          CACHIX_AUTH_TOKEN: ${{ secrets.CACHIX_AUTH_TOKEN }}
+          CACHIX_SIGNING_KEY: ${{ secrets.CACHIX_SIGNING_KEY }}
+          NIKS3_SERVER: ${{ vars.NIKS3_SERVER }}
+          NIKS3_TOKEN: ${{ secrets.NIKS3_TOKEN }}
+        run: |
+          set -euo pipefail
+          # a missing root means setup could not find the atelier tree
+          if [ -z "${ATELIER_ROOT:-}" ]; then
+            echo "::warning::Skipping cache streamer: setup did not complete"
+            exit 0
+          fi
+          # a stale sentinel or pidfile from a prior job on a persistent
+          # runner must not stop or stall this job's streamer
+          rm -f "${RUNNER_TEMP}/atelier-stream.done" "${RUNNER_TEMP}/atelier-stream.pid"
+          python3 "${ATELIER_ROOT}/src/atelier/stream.py" --mode stream \
+            > "${RUNNER_TEMP}/atelier-stream.log" 2>&1 &
+          printf '%s' "$!" > "${RUNNER_TEMP}/atelier-stream.pid" \
+            || echo "::warning::Could not record the streamer pid"
 
       - name: Build
+        id: build
         if: ${{ matrix.installable != '' }}
         shell: bash
         run: |
@@ -126,13 +159,17 @@ jobs:
           nix build "${INSTALLABLE}^*" --no-link --print-build-logs 2>&1 | tee build.log
 
       # the push phase brackets pre-push, the push (built-in or custom), and
-      # post-push under one guard: a push was requested for a real, non-fork
-      # build. it deliberately omits the "a backend is configured" check the
-      # built-in push carries, so push-command and the hooks can target a cache
-      # atelier has no native support for. each step is env-passed and not
-      # interpolated into the run body so it cannot break out of the run
+      # post-push. with built-in backends the streamer has already been
+      # pushing during the build, and the final drain below re-pushes the
+      # full spool plus the outputs' closure as the backstop, also on a
+      # failed build (never on a cancelled or setup-failed cell) so partial
+      # results land on the cache. the pre/post hooks keep their success-only
+      # timing unless the built-in drain runs, and only for a real build
+      # failure (not a setup failure, where no push happens). each step is
+      # env-passed and not interpolated into the run body so it cannot break
+      # out of the run
       - name: Pre Push Hook
-        if: ${{ inputs.push && matrix.installable != '' && github.event.pull_request.head.repo.fork != true && inputs.pre-push != '' }}
+        if: ${{ inputs.push && matrix.installable != '' && github.event.pull_request.head.repo.fork != true && inputs.pre-push != '' && (success() || (!cancelled() && steps.build.outcome == 'failure' && inputs.push-command == '' && ((vars.ATTIC_SERVER != '' && vars.ATTIC_CACHE != '') || vars.CACHIX_CACHE != '' || vars.NIKS3_SERVER != ''))) }}
         shell: bash
         env:
           PRE_PUSH: ${{ inputs.pre-push }}
@@ -149,9 +186,10 @@ jobs:
           bash -c "${PRE_PUSH}"
 
       - name: Push To Cache
-        if: ${{ inputs.push && inputs.push-command == '' && matrix.installable != '' && github.event.pull_request.head.repo.fork != true && ((vars.ATTIC_SERVER != '' && vars.ATTIC_CACHE != '') || vars.CACHIX_CACHE != '' || vars.NIKS3_SERVER != '') }}
+        if: ${{ (success() || (!cancelled() && steps.build.outcome == 'failure')) && inputs.push && inputs.push-command == '' && matrix.installable != '' && github.event.pull_request.head.repo.fork != true && ((vars.ATTIC_SERVER != '' && vars.ATTIC_CACHE != '') || vars.CACHIX_CACHE != '' || vars.NIKS3_SERVER != '') }}
         shell: bash
         env:
+          BUILD_OUTCOME: ${{ steps.build.outcome }}
           ATTIC_SERVER: ${{ vars.ATTIC_SERVER }}
           ATTIC_CACHE: ${{ vars.ATTIC_CACHE }}
           ATTIC_TOKEN: ${{ secrets.ATTIC_TOKEN }}
@@ -161,90 +199,13 @@ jobs:
           NIKS3_SERVER: ${{ vars.NIKS3_SERVER }}
           NIKS3_TOKEN: ${{ secrets.NIKS3_TOKEN }}
         run: |
-          set -uo pipefail
-
-          # only the outputs we just built
-          paths="$(nix build "${INSTALLABLE}^*" --no-link --print-out-paths)"
-          [ -n "$paths" ] || exit 0
-
-          # push to every configured backend, best-effort: each runs in its own
-          # subshell so one cannot abort the others, and a failed push only warns
-          # (the artifact is already built, so a cache upload hiccup must not fail
-          # the build). drop -e here so a failing backend falls through to the
-          # next; each subshell still runs set -euo pipefail, effective only
-          # because it runs as a plain statement whose status we read with $? -
-          # a "( set -e ... ) || warn" would disable that inner set -e (bash
-          # ignores -e set inside a command on the left of ||)
-          set +e
-
-          if [ -n "${ATTIC_SERVER}" ] && [ -n "${ATTIC_CACHE}" ]; then
-            (
-              set -euo pipefail
-              # lix only has 'install', on cppnix it is a deprecated alias for
-              # 'add'. revert to 'add' once lix supports 'add'
-              nix profile install nixpkgs#attic-client
-              attic login default "${ATTIC_SERVER}" "${ATTIC_TOKEN}"
-              # a missing or unreachable cache is a soft skip, not a build failure
-              attic cache info "${ATTIC_CACHE}" || { echo "::warning::Attic cache unavailable, skipping"; exit 0; }
-              # shellcheck disable=SC2086
-              attic push "${ATTIC_CACHE}" $paths
-            )
-            # shellcheck disable=SC2181
-            [ $? -eq 0 ] || echo "::warning::Attic push failed"
-          fi
-
-          if [ -n "${CACHIX_CACHE}" ]; then
-            (
-              set -euo pipefail
-              # lix only has 'install', on cppnix it is a deprecated alias for
-              # 'add'. revert to 'add' once lix supports 'add'
-              nix profile install nixpkgs#cachix
-              [ -n "${CACHIX_SIGNING_KEY:-}" ] || unset CACHIX_SIGNING_KEY
-              # shellcheck disable=SC2086
-              printf '%s\n' $paths | cachix push "${CACHIX_CACHE}"
-            )
-            # shellcheck disable=SC2181
-            [ $? -eq 0 ] || echo "::warning::Cachix push failed"
-          fi
-
-          if [ -n "${NIKS3_SERVER}" ]; then
-            (
-              set -euo pipefail
-              # nixpkgs niks3 lags at 1.4.0 (no --auth-token-{path,script})
-              # use my pinned niks3
-              # lix only has 'install', on cppnix it is a deprecated alias for
-              # 'add'. revert to 'add' once lix supports 'add'
-              nix profile install github:stepbrobd/inc#niks3
-              if [ -n "${NIKS3_TOKEN:-}" ]; then
-                # token auth: write the secret to a private file, pass its path
-                tok="$(mktemp)"
-                (umask 077; printf '%s' "${NIKS3_TOKEN}" > "$tok")
-                # shellcheck disable=SC2086
-                niks3 push --server-url "${NIKS3_SERVER}" --auth-token-path "$tok" $paths
-              elif [ -n "${ACTIONS_ID_TOKEN_REQUEST_URL:-}" ]; then
-                # oidc auth: the audience must match what the server validates the jwt against
-                aud="$(curl -sf "${NIKS3_SERVER}/api/cache-config?issuer=https://token.actions.githubusercontent.com" | jq -r '.oidc_audience // empty')"
-                if [ -z "$aud" ]; then
-                  echo "::warning::Skipping niks3 push: server advertises no oidc_audience for the GitHub issuer; configure a GitHub OIDC provider on the server"
-                else
-                  # niks3 reruns this script to refresh the token; leave the github
-                  # oidc env vars unexpanded so they are read at each mint, not now
-                  s="$(mktemp)"
-                  printf '%s\n' \
-                    '#!/bin/sh' \
-                    "exec curl -sf -H \"Authorization: Bearer \$ACTIONS_ID_TOKEN_REQUEST_TOKEN\" \"\$ACTIONS_ID_TOKEN_REQUEST_URL&audience=${aud}\" | jq '{token: .value, expires_at: ((now + 240) | todateiso8601)}'" \
-                    > "$s"
-                  chmod 700 "$s"
-                  # shellcheck disable=SC2086
-                  niks3 push --server-url "${NIKS3_SERVER}" --auth-token-script "$s" $paths
-                fi
-              else
-                echo "::warning::Skipping niks3 push: OIDC needs 'id-token: write' in the caller workflow"
-              fi
-            )
-            # shellcheck disable=SC2181
-            [ $? -eq 0 ] || echo "::warning::niks3 push failed"
+          set -euo pipefail
+          # a failed setup leaves no atelier checkout, skip rather than error
+          if [ -z "${ATELIER_ROOT:-}" ]; then
+            echo "::warning::Skipping cache push: setup did not complete"
+            exit 0
           fi
+          python3 "${ATELIER_ROOT}/src/atelier/stream.py" --mode final || echo "::warning::Cache push failed"
 
       # custom push replaces the built-in one (like install-command replaces the
       # installer): when set, the native Push To Cache above is skipped. the
@@ -269,7 +230,7 @@ jobs:
           bash -c "${PUSH_COMMAND}"
 
       - name: Post Push Hook
-        if: ${{ inputs.push && matrix.installable != '' && github.event.pull_request.head.repo.fork != true && inputs.post-push != '' }}
+        if: ${{ inputs.push && matrix.installable != '' && github.event.pull_request.head.repo.fork != true && inputs.post-push != '' && (success() || (!cancelled() && steps.build.outcome == 'failure' && inputs.push-command == '' && ((vars.ATTIC_SERVER != '' && vars.ATTIC_CACHE != '') || vars.CACHIX_CACHE != '' || vars.NIKS3_SERVER != ''))) }}
         shell: bash
         env:
           POST_PUSH: ${{ inputs.post-push }}
diff --git a/flake.nix b/flake.nix
index cb8d879..c664e46 100644
--- a/flake.nix
+++ b/flake.nix
@@ -27,6 +27,7 @@
             # code
             ./src
             ./tests
+            ./.github/actions/atelier/hook.sh
             # meta
             ./license.txt
             ./pyproject.toml
diff --git a/pyproject.toml b/pyproject.toml
index ec611c5..f217f1b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
 [project]
 name = "atelier"
-version = "2026.816.0"
+version = "2026.819.0"
 description = "nix atelier ;)"
 readme = "readme.md"
 requires-python = ">=3.14"
diff --git a/readme.md b/readme.md
index a716e34..2300ccb 100644
--- a/readme.md
+++ b/readme.md
@@ -169,6 +169,15 @@ Pushes happen on a push to your repository's default branch (or `master`) and on
 a run with `push: true`. Forked-PR runs never push. Caching is best-effort: a
 failed push to any backend is logged as a warning and never fails the build.
 
+Uploads start while the build runs: a post-build hook records every locally
+built store path, and a background uploader pushes them to every configured
+backend as they appear. After the build, a final drain re-pushes the full set
+plus the finished outputs' closure, so substituted dependencies missing from
+your cache still land there. A failed build pushes the paths it built before
+failing, and a cancelled or timed-out run keeps whatever was uploaded before the
+kill. On a self-hosted runner a cancelled cell can leave the background uploader
+running until the runner reaps job processes.
+
 ## Use it in your repo
 
 Atelier runs against whatever repository calls it. `actions/checkout` inside the
@@ -249,10 +258,13 @@ optional inputs change that without forking:
 Atelier separates installing Nix from configuring it. Whichever installer runs,
 Atelier applies its own required `nix.conf` afterwards (the GitHub access token,
 `experimental-features = nix-command flakes`, the build directory, the target
-`system`, and the sandbox mode), then appends your `extra-conf` last. So a
-custom installer still ends up with a correctly configured daemon, and adding
-settings is independent of the installer choice. Use Nix's `extra-` prefixes to
-add to a list setting rather than replace it.
+`system`, the sandbox mode, and the post-build hook), then appends your
+`extra-conf` last. So a custom installer still ends up with a correctly
+configured daemon, and adding settings is independent of the installer choice.
+Use Nix's `extra-` prefixes to add to a list setting rather than replace it.
+With a binary cache configured it also sets `post-build-hook` to record built
+paths for streaming. A `post-build-hook` of your own in `extra-conf` wins and
+replaces streaming.
 
 The inputs apply to every job, so discovery and every build cell use the same
 Nix. Install Lix instead of upstream Nix and enable the pipe operator:
@@ -323,6 +335,13 @@ jobs:
 `pre-push`, `post-push`, and `push-command` apply only to the build cells (the
 discovery job never pushes). They share the `push: true` guard.
 
+With built-in backends the push begins during the build, so `pre-push` runs
+before the final drain rather than before all push activity. Streamed batches
+that need a `pre-push` side effect fail soft and are re-pushed by the final
+drain after the hook has run. When a build fails and a built-in backend is
+configured, `pre-push` and `post-push` also run around the failure-path drain.
+With `push-command` set, hook timing is unchanged.
+
 `push-command` replaces the built-in push exactly like `install-command`
 replaces the installer (set it and the native Attic/Cachix/niks3 push is
 skipped). The attribute being built is exposed as `INSTALLABLE`, so a command
diff --git a/src/atelier/stream.py b/src/atelier/stream.py
new file mode 100644
index 0000000..8449f4a
--- /dev/null
+++ b/src/atelier/stream.py
@@ -0,0 +1,383 @@
+"""Streaming cache push for build cells.
+
+Run by file path with the runner's system python3, never as part of the
+packaged tool, so this module stays stdlib-only and keeps Python 3.12
+syntax (ubuntu runners ship 3.12, the package floor of 3.14 applies only
+to the tool run via nix).
+
+Stream mode tails the spool the post-build hook appends to and pushes new
+store paths to every configured backend while the build runs. Final mode
+stops the streamer, then pushes every spooled path plus, when the build
+succeeded, the final outputs' closure. Both modes exit 0 unconditionally
+because a cache push never fails a build.
+"""
+
+import argparse
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Protocol
+
+# paths per tool invocation, keeps argv far under ARG_MAX on both platforms
+BATCH = 1000
+# seconds between spool and pid polls when idle
+POLL = 2.0
+# seconds final mode waits for the streamer to drain before proceeding
+WAIT = 900.0
+
+# github's oidc issuer, the audience is discovered from the niks3 server
+ISSUER = "https://token.actions.githubusercontent.com"
+
+# the mint script niks3 re-runs to refresh its token
+# expires_at of now+240 keeps the refresh inside github's ~5 minute token life
+_MINT = """#!/usr/bin/env python3
+import json, os, time, urllib.request
+req = urllib.request.Request(
+    os.environ["ACTIONS_ID_TOKEN_REQUEST_URL"] + "&audience=@AUDIENCE@",
+    headers={"Authorization": "Bearer " + os.environ["ACTIONS_ID_TOKEN_REQUEST_TOKEN"]},
+)
+with urllib.request.urlopen(req) as resp:
+    token = json.load(resp)["value"]
+exp = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() + 240))
+print(json.dumps({"token": token, "expires_at": exp}))
+"""
+
+
+def _warn(msg: str) -> None:
+    # single write so concurrent threads cannot interleave annotations
+    # stdout on purpose, the streamer's stdout is the log final mode replays
+    print(f"::warning::{msg}\n", end="", flush=True)
+
+
+def _run(
+    argv: list[str], stdin: str | None = None, env: dict[str, str] | None = None
+) -> None:
+    # module level so tests can monkeypatch process execution
+    # argv list only, nothing is interpreted by a shell
+    # env is a complete environment, not an overlay over os.environ
+    subprocess.run(argv, input=stdin, env=env, check=True, text=True)
+
+
+def read_new(spool: Path, offset: int) -> tuple[list[str], int]:
+    """Complete lines past a byte offset and the new byte offset.
+
+    A partial trailing line stays for the next read, blank lines are
+    skipped, and a missing spool reads as empty. The spool is append
+    only; the offset never rewinds. Undecodable bytes are replaced so
+    a corrupt line degrades to a failed push instead of a dead reader.
+    """
+    try:
+        data = spool.read_bytes()[offset:]
+    except FileNotFoundError:
+        return [], offset
+    end = data.rfind(b"\n")
+    if end < 0:
+        return [], offset
+    chunk = data[: end + 1]
+    lines = [ln.strip() for ln in chunk.decode(errors="replace").split("\n")]
+    return [ln for ln in lines if ln], offset + len(chunk)
+
+
+class Backend(Protocol):
+    """One binary cache target, structurally satisfied by the classes below."""
+
+    name: str
+
+    def setup(self) -> None: ...
+
+    def push(self, paths: list[str]) -> None: ...
+
+
+def _ensure(binary: str, installable: str) -> None:
+    # lix only has 'install', on cppnix it is a deprecated alias for 'add'
+    # revert to 'add' once lix supports 'add'
+    if shutil.which(binary) is None:
+        _run(["nix", "profile", "install", installable])
+
+
+def _secret(token: str) -> str:
+    # mkstemp creates 0600, the token never lands in argv or the log
+    fd, path = tempfile.mkstemp()
+    with os.fdopen(fd, "w") as f:
+        f.write(token)
+    return path
+
+
+def _audience(server: str) -> str:
+    # https only, this fetch decides whether to trust the server with a token
+    if not server.startswith("https://"):
+        raise RuntimeError("server URL must be https")
+    query = urllib.parse.urlencode({"issuer": ISSUER})
+    with urllib.request.urlopen(
+        f"{server.rstrip('/')}/api/cache-config?{query}", timeout=30
+    ) as resp:
+        return str(json.load(resp).get("oidc_audience") or "")
+
+
+def _mint(audience: str) -> str:
+    fd, path = tempfile.mkstemp()
+    with os.fdopen(fd, "w") as f:
+        # percent-encode so a hostile audience cannot escape the string
+        # literal in the generated script or smuggle extra query parameters
+        f.write(_MINT.replace("@AUDIENCE@", urllib.parse.quote(audience, safe="")))
+    os.chmod(path, 0o700)
+    return path
+
+
+@dataclass
+class Attic:
+    server: str
+    cache: str
+    token: str = field(repr=False)
+    name: str = "Attic"
+
+    def setup(self) -> None:
+        if not self.token:
+            raise RuntimeError("ATTIC_TOKEN is not set")
+        _ensure("attic", "nixpkgs#attic-client")
+        try:
+            _run(["attic", "login", "default", self.server, self.token])
+        except subprocess.CalledProcessError as e:
+            # the token is in argv, re-raise without echoing it into the log
+            raise RuntimeError(
+                f"attic login failed with exit code {e.returncode}"
+            ) from None
+        # an unreachable cache disables the backend, never the build
+        _run(["attic", "cache", "info", self.cache])
+
+    def push(self, paths: list[str]) -> None:
+        _run(["attic", "push", self.cache, *paths])
+
+
+@dataclass
+class Cachix:
+    cache: str
+    name: str = "Cachix"
+
+    def setup(self) -> None:
+        # pushing needs credentials, fail once here instead of every batch
+        if not os.environ.get("CACHIX_AUTH_TOKEN") and not os.environ.get(
+            "CACHIX_SIGNING_KEY"
+        ):
+            raise RuntimeError("CACHIX_AUTH_TOKEN is not set")
+        _ensure("cachix", "nixpkgs#cachix")
+
+    def push(self, paths: list[str]) -> None:
+        env = dict(os.environ)
+        # cachix rejects an empty signing key, absent means unsigned
+        if not env.get("CACHIX_SIGNING_KEY"):
+            env.pop("CACHIX_SIGNING_KEY", None)
+        _run(["cachix", "push", self.cache], stdin="\n".join(paths) + "\n", env=env)
+
+
+@dataclass
+class Niks3:
+    server: str
+    token: str = field(repr=False)
+    name: str = "niks3"
+    auth: tuple[str, str] | None = None
+
+    def setup(self) -> None:
+        # nixpkgs still lags at 1.4.0 without the auth flags, use the pin
+        _ensure("niks3", "github:stepbrobd/inc#niks3")
+        if self.token:
+            self.auth = ("--auth-token-path", _secret(self.token))
+        elif os.environ.get("ACTIONS_ID_TOKEN_REQUEST_URL"):
+            audience = _audience(self.server)
+            if not audience:
+                raise RuntimeError(
+                    "server advertises no oidc_audience for the GitHub issuer"
+                )
+            self.auth = ("--auth-token-script", _mint(audience))
+        else:
+            raise RuntimeError("OIDC needs 'id-token: write' in the caller workflow")
+
+    def push(self, paths: list[str]) -> None:
+        if self.auth is None:
+            raise RuntimeError("niks3 setup did not run")
+        _run(["niks3", "push", "--server-url", self.server, *self.auth, *paths])
+
+
+def backends() -> list[Backend]:
+    env = os.environ
+    out: list[Backend] = []
+    if env.get("ATTIC_SERVER") and env.get("ATTIC_CACHE"):
+        out.append(
+            Attic(env["ATTIC_SERVER"], env["ATTIC_CACHE"], env.get("ATTIC_TOKEN", ""))
+        )
+    if env.get("CACHIX_CACHE"):
+        out.append(Cachix(env["CACHIX_CACHE"]))
+    if env.get("NIKS3_SERVER"):
+        out.append(Niks3(env["NIKS3_SERVER"], env.get("NIKS3_TOKEN", "")))
+    return out
+
+
+def ready(backend: Backend) -> bool:
+    # one time setup, a failure disables the backend for this run
+    try:
+        backend.setup()
+    except Exception as e:  # noqa: BLE001
+        _warn(f"{backend.name} disabled: {e}")
+        return False
+    return True
+
+
+def push(backend: Backend, paths: list[str]) -> None:
+    # per batch so one failure cannot abandon the rest, this same function
+    # is the final drain's last line of defense
+    for i in range(0, len(paths), BATCH):
+        try:
+            backend.push(paths[i : i + BATCH])
+        except Exception as e:  # noqa: BLE001
+            # a CalledProcessError str embeds the whole argv, keep it short
+            reason: object = e
+            if isinstance(e, subprocess.CalledProcessError):
+                reason = f"exit {e.returncode}"
+            _warn(f"{backend.name} push failed: {reason}")
+
+
+# tool installs share one nix profile, serialize the setups
+_SETUP = threading.Lock()
+
+
+def stream_backend(backend: Backend, spool: Path, done: Path) -> None:
+    # own offset per backend, exits once the sentinel exists and every
+    # complete line has been read, or immediately when setup fails
+    offset = 0
+    ok: bool | None = None
+    try:
+        while True:
+            lines, offset = read_new(spool, offset)
+            if lines:
+                if ok is None:
+                    with _SETUP:
+                        ok = ready(backend)
+                if not ok:
+                    return
+                push(backend, lines)
+            elif done.exists():
+                return
+            else:
+                time.sleep(POLL)
+    except Exception as e:  # noqa: BLE001
+        # a dead reader is invisible otherwise, final mode re-pushes the spool
+        _warn(f"{backend.name} streamer stopped: {e}")
+
+
+def mode_stream(spool: Path, done: Path) -> None:
+    bs = backends()
+    if not bs:
+        _warn("No cache backend configured, streamer exiting")
+        return
+    threads = [
+        threading.Thread(target=stream_backend, args=(b, spool, done)) for b in bs
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+
+def _wait(pidfile: Path) -> None:
+    # bounded wait, a hung streamer only costs duplicate bandwidth
+    try:
+        pid = int(pidfile.read_text().strip())
+    except (FileNotFoundError, ValueError):
+        return
+    deadline = time.monotonic() + WAIT
+    while time.monotonic() < deadline:
+        try:
+            os.kill(pid, 0)
+        except (ProcessLookupError, PermissionError):
+            return
+        time.sleep(POLL)
+    _warn("Streamer still running after wait timeout, pushing anyway")
+
+
+def _replay(log: Path) -> None:
+    if not log.exists():
+        return
+    text = log.read_text(errors="replace")
+    print("::group::Streamer log")
+    sys.stdout.write(text)
+    # github only parses commands at line start, keep the group closable
+    if text and not text.endswith("\n"):
+        sys.stdout.write("\n")
+    print("::endgroup::", flush=True)
+
+
+def _outputs(installable: str) -> list[str]:
+    # the final outputs, resolvable only after a successful build
+    if not installable:
+        return []
+    try:
+        proc = subprocess.run(
+            ["nix", "build", f"{installable}^*", "--no-link", "--print-out-paths"],
+            check=True,
+            text=True,
+            capture_output=True,
+            timeout=WAIT,
+        )
+    except Exception as e:  # noqa: BLE001
+        # str(CalledProcessError) omits the captured stderr, print it too
+        stderr = str(getattr(e, "stderr", "") or "").strip()
+        if stderr:
+            print(stderr[-2000:], flush=True)
+        _warn(f"Final output resolution failed: {e}")
+        return []
+    return [ln for ln in proc.stdout.splitlines() if ln.strip()]
+
+
+def mode_final(spool: Path, done: Path, pidfile: Path, log: Path) -> None:
+    done.touch()
+    try:
+        _wait(pidfile)
+        _replay(log)
+    except Exception as e:  # noqa: BLE001
+        # a preamble failure must never cost the drain itself
+        _warn(f"Streamer wait or log replay failed: {e}")
+    paths = read_new(spool, 0)[0]
+    if os.environ.get("BUILD_OUTCOME") == "success":
+        installable = os.environ.get("INSTALLABLE", "")
+        if not installable:
+            _warn("INSTALLABLE is not set, skipping the final closure push")
+        # the closure sweep also covers substituted deps missing on the cache
+        paths += _outputs(installable)
+    paths = list(dict.fromkeys(paths))
+    if not paths:
+        return
+    for b in backends():
+        if ready(b):
+            push(b, paths)
+
+
+def main() -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--mode", choices=("stream", "final"), required=True)
+    mode = ap.parse_args().mode
+    spool = Path(os.environ.get("ATELIER_SPOOL", "/nix/var/atelier/spool"))
+    tmp = Path(os.environ["RUNNER_TEMP"])
+    done = tmp / "atelier-stream.done"
+    if mode == "stream":
+        mode_stream(spool, done)
+    else:
+        mode_final(spool, done, tmp / "atelier-stream.pid", tmp / "atelier-stream.log")
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(main())
+    except Exception as e:  # noqa: BLE001
+        # a cache push must never fail the build, even on an internal bug
+        _warn(f"Cache push failed: {e!r}")
+        sys.exit(0)
diff --git a/tests/test_hook.py b/tests/test_hook.py
new file mode 100644
index 0000000..4e11200
--- /dev/null
+++ b/tests/test_hook.py
@@ -0,0 +1,40 @@
+import stat
+import subprocess
+from pathlib import Path
+
+_HOOK = Path(__file__).parent.parent / ".github" / "actions" / "atelier" / "hook.sh"
+
+
+def _hook(spool: Path, out_paths: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [str(_HOOK)],
+        env={"ATELIER_SPOOL": str(spool), "OUT_PATHS": out_paths},
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+def test_hook_appends_one_path_per_line(tmp_path: Path) -> None:
+    spool = tmp_path / "spool"
+    proc = _hook(spool, "/nix/store/aaa-x /nix/store/bbb-y")
+    assert proc.returncode == 0
+    assert spool.read_text() == "/nix/store/aaa-x\n/nix/store/bbb-y\n"
+
+
+def test_hook_appends_across_invocations(tmp_path: Path) -> None:
+    spool = tmp_path / "spool"
+    _hook(spool, "/nix/store/aaa-x")
+    _hook(spool, "/nix/store/bbb-y")
+    assert spool.read_text() == "/nix/store/aaa-x\n/nix/store/bbb-y\n"
+
+
+def test_hook_never_fails(tmp_path: Path) -> None:
+    # an unwritable spool must not abort the build loop
+    proc = _hook(tmp_path / "no" / "such" / "dir" / "spool", "/nix/store/aaa-x")
+    assert proc.returncode == 0
+    assert proc.stderr == ""
+
+
+def test_hook_is_executable() -> None:
+    assert _HOOK.stat().st_mode & stat.S_IXUSR
diff --git a/tests/test_stream.py b/tests/test_stream.py
new file mode 100644
index 0000000..3cdc4fb
--- /dev/null
+++ b/tests/test_stream.py
@@ -0,0 +1,686 @@
+import ast
+import io
+import os
+import stat
+import subprocess
+import sys
+import threading
+from pathlib import Path
+from typing import ClassVar
+
+import pytest
+
+from atelier import stream
+from atelier.stream import (
+    Attic,
+    Cachix,
+    Niks3,
+    backends,
+    mode_final,
+    mode_stream,
+    push,
+    read_new,
+    ready,
+    stream_backend,
+)
+
+
+def test_read_new_returns_complete_lines_and_offset(tmp_path: Path) -> None:
+    spool = tmp_path / "spool"
+    spool.write_text("/nix/store/aaa-x\n/nix/store/bbb-y\n")
+    lines, offset = read_new(spool, 0)
+    assert lines == ["/nix/store/aaa-x", "/nix/store/bbb-y"]
+    assert offset == len(spool.read_bytes())
+
+
+def test_read_new_resumes_from_offset(tmp_path: Path) -> None:
+    spool = tmp_path / "spool"
+    spool.write_text("/nix/store/aaa-x\n")
+    _, offset = read_new(spool, 0)
+    spool.write_text("/nix/store/aaa-x\n/nix/store/bbb-y\n")
+    lines, _ = read_new(spool, offset)
+    assert lines == ["/nix/store/bbb-y"]
+
+
+def test_read_new_holds_back_partial_trailing_line(tmp_path: Path) -> None:
+    spool = tmp_path / "spool"
+    spool.write_text("/nix/store/aaa-x\n/nix/store/bb")
+    lines, offset = read_new(spool, 0)
+    assert lines == ["/nix/store/aaa-x"]
+    spool.write_text("/nix/store/aaa-x\n/nix/store/bbb-y\n")
+    lines, _ = read_new(spool, offset)
+    assert lines == ["/nix/store/bbb-y"]
+
+
+def test_read_new_skips_blank_lines(tmp_path: Path) -> None:
+    spool = tmp_path / "spool"
+    spool.write_text("\n/nix/store/aaa-x\n\n")
+    lines, _ = read_new(spool, 0)
+    assert lines == ["/nix/store/aaa-x"]
+
+
+def test_read_new_missing_spool_reads_as_empty(tmp_path: Path) -> None:
+    assert read_new(tmp_path / "spool", 0) == ([], 0)
+
+
+def test_read_new_idle_returns_same_offset(tmp_path: Path) -> None:
+    # the poll loop's hottest path, an unchanged spool must not re-yield
+    spool = tmp_path / "spool"
+    spool.write_text("/nix/store/aaa-x\n")
+    _, offset = read_new(spool, 0)
+    assert read_new(spool, offset) == ([], offset)
+
+
+def test_read_new_no_newline_yet(tmp_path: Path) -> None:
+    spool = tmp_path / "spool"
+    spool.write_text("/nix/store/aa")
+    assert read_new(spool, 0) == ([], 0)
+
+
+def test_read_new_consumes_blank_lines(tmp_path: Path) -> None:
+    spool = tmp_path / "spool"
+    spool.write_text("\n/nix/store/aaa-x\n\n")
+    _, offset = read_new(spool, 0)
+    assert offset == len(spool.read_bytes())
+
+
+def test_module_keeps_python_312_floor() -> None:
+    # build cells run this file with the runner's system python3, 3.12 on ubuntu
+    src = Path(stream.__file__).read_text()
+    ast.parse(src, feature_version=(3, 12))
+    tree = ast.parse(src)
+    imported = {
+        n.name.split(".")[0]
+        for node in ast.walk(tree)
+        if isinstance(node, ast.Import)
+        for n in node.names
+    } | {
+        node.module.split(".")[0]
+        for node in ast.walk(tree)
+        if isinstance(node, ast.ImportFrom) and node.module
+    }
+    assert imported <= sys.stdlib_module_names
+
+
+def _record(monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
+    calls: list[list[str]] = []
+
+    def fake(
+        argv: list[str], stdin: str | None = None, env: dict[str, str] | None = None
+    ) -> None:
+        calls.append(list(argv))
+
+    monkeypatch.setattr(stream, "_run", fake)
+    monkeypatch.setattr(stream.shutil, "which", lambda _: "/bin/true")
+    return calls
+
+
+def test_backends_from_env(monkeypatch: pytest.MonkeyPatch) -> None:
+    for var in (
+        "ATTIC_SERVER",
+        "ATTIC_CACHE",
+        "ATTIC_TOKEN",
+        "CACHIX_CACHE",
+        "NIKS3_SERVER",
+        "NIKS3_TOKEN",
+    ):
+        monkeypatch.delenv(var, raising=False)
+    assert backends() == []
+    monkeypatch.setenv("ATTIC_SERVER", "https://a.example")
+    # attic needs both server and cache
+    assert backends() == []
+    monkeypatch.setenv("ATTIC_CACHE", "c")
+    monkeypatch.setenv("CACHIX_CACHE", "d")
+    monkeypatch.setenv("NIKS3_SERVER", "https://n.example")
+    assert backends() == [
+        Attic("https://a.example", "c", ""),
+        Cachix("d"),
+        Niks3("https://n.example", ""),
+    ]
+
+
+def test_attic_setup_and_push(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls = _record(monkeypatch)
+    b = Attic("https://a.example", "c", "tok")
+    b.setup()
+    b.push(["/nix/store/aaa-x"])
+    assert calls == [
+        ["attic", "login", "default", "https://a.example", "tok"],
+        ["attic", "cache", "info", "c"],
+        ["attic", "push", "c", "/nix/store/aaa-x"],
+    ]
+
+
+def test_ensure_installs_missing_tool(monkeypatch: pytest.MonkeyPatch) -> None:
+    calls: list[list[str]] = []
+    monkeypatch.setattr(
+        stream, "_run", lambda argv, stdin=None, env=None: calls.append(list(argv))
+    )
+    monkeypatch.setattr(stream.shutil, "which", lambda _: None)
+    monkeypatch.setenv("CACHIX_AUTH_TOKEN", "t")
+    Cachix("d").setup()
+    assert calls == [["nix", "profile", "install", "nixpkgs#cachix"]]
+
+
+def test_cachix_push_paths_on_stdin_and_drops_empty_signing_key(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    seen: dict[str, object] = {}
+
+    def fake(
+        argv: list[str], stdin: str | None = None, env: dict[str, str] | None = None
+    ) -> None:
+        seen["argv"], seen["stdin"], seen["env"] = argv, stdin, env
+
+    monkeypatch.setattr(stream, "_run", fake)
+    monkeypatch.setenv("CACHIX_SIGNING_KEY", "")
+    monkeypatch.setenv("ATELIER_CANARY", "1")
+    Cachix("d").push(["/nix/store/aaa-x", "/nix/store/bbb-y"])
+    assert seen["argv"] == ["cachix", "push", "d"]
+    assert seen["stdin"] == "/nix/store/aaa-x\n/nix/store/bbb-y\n"
+    env = seen["env"]
+    assert isinstance(env, dict)
+    assert "CACHIX_SIGNING_KEY" not in env
+    assert env.get("ATELIER_CANARY") == "1"
+
+
+def test_cachix_preserves_nonempty_signing_key(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    seen: dict[str, object] = {}
+
+    def fake(
+        argv: list[str], stdin: str | None = None, env: dict[str, str] | None = None
+    ) -> None:
+        seen["env"] = env
+
+    monkeypatch.setattr(stream, "_run", fake)
+    monkeypatch.setenv("CACHIX_SIGNING_KEY", "key")
+    Cachix("d").push(["/nix/store/aaa-x"])
+    env = seen["env"]
+    assert isinstance(env, dict)
+    assert env["CACHIX_SIGNING_KEY"] == "key"
+
+
+def test_niks3_token_auth(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    calls = _record(monkeypatch)
+    monkeypatch.setattr(stream.tempfile, "tempdir", str(tmp_path))
+    b = Niks3("https://n.example", "tok")
+    b.setup()
+    assert b.auth is not None
+    flag, path = b.auth
+    assert flag == "--auth-token-path"
+    assert Path(path).read_text() == "tok"
+    # mkstemp creates the token file private to the runner user
+    assert stat.S_IMODE(Path(path).stat().st_mode) == 0o600
+    b.push(["/nix/store/aaa-x"])
+    assert calls[-1] == [
+        "niks3",
+        "push",
+        "--server-url",
+        "https://n.example",
+        "--auth-token-path",
+        path,
+        "/nix/store/aaa-x",
+    ]
+
+
+def test_niks3_oidc_auth_writes_mint_script(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    _record(monkeypatch)
+    monkeypatch.setattr(stream.tempfile, "tempdir", str(tmp_path))
+    monkeypatch.setenv("ACTIONS_ID_TOKEN_REQUEST_URL", "https://gh.example/token")
+    monkeypatch.setattr(stream, "_audience", lambda _: "https://n.example")
+    b = Niks3("https://n.example", "")
+    b.setup()
+    assert b.auth is not None
+    flag, path = b.auth
+    assert flag == "--auth-token-script"
+    body = Path(path).read_text()
+    assert "&audience=https%3A%2F%2Fn.example" in body
+    assert "ACTIONS_ID_TOKEN_REQUEST_TOKEN" in body
+    assert stat.S_IMODE(Path(path).stat().st_mode) == 0o700
+
+
+def test_audience_parses_cache_config(monkeypatch: pytest.MonkeyPatch) -> None:
+    seen: list[str] = []
+
+    def fake(url: str, timeout: float = 0) -> io.BytesIO:
+        seen.append(url)
+        return io.BytesIO(b'{"oidc_audience": "https://n.example"}')
+
+    monkeypatch.setattr(stream.urllib.request, "urlopen", fake)
+    assert stream._audience("https://n.example") == "https://n.example"
+    assert seen == [
+        "https://n.example/api/cache-config?issuer=https%3A%2F%2Ftoken.actions.githubusercontent.com"
+    ]
+
+    def empty(url: str, timeout: float = 0) -> io.BytesIO:
+        return io.BytesIO(b"{}")
+
+    monkeypatch.setattr(stream.urllib.request, "urlopen", empty)
+    assert stream._audience("https://n.example") == ""
+
+
+def test_niks3_oidc_without_audience_disables(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    _record(monkeypatch)
+    monkeypatch.setenv("ACTIONS_ID_TOKEN_REQUEST_URL", "https://gh.example/token")
+    monkeypatch.setattr(stream, "_audience", lambda _: "")
+    assert ready(Niks3("https://n.example", "")) is False
+    assert "::warning::niks3 disabled" in capsys.readouterr().out
+
+
+def test_niks3_without_token_or_oidc_disables(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    _record(monkeypatch)
+    monkeypatch.delenv("ACTIONS_ID_TOKEN_REQUEST_URL", raising=False)
+    assert ready(Niks3("https://n.example", "")) is False
+    assert "id-token" in capsys.readouterr().out
+
+
+def test_push_batches_and_isolates_failure(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    class Fake:
+        name = "Fake"
+        batches: ClassVar[list[int]] = []
+
+        def setup(self) -> None:
+            pass
+
+        def push(self, paths: list[str]) -> None:
+            self.batches.append(len(paths))
+
+    fake = Fake()
+    push(fake, [f"/nix/store/{i}" for i in range(2500)])
+    assert fake.batches == [1000, 1000, 500]
+
+    class Boom:
+        name = "Boom"
+
+        def setup(self) -> None:
+            pass
+
+        def push(self, paths: list[str]) -> None:
+            raise RuntimeError("nope")
+
+    push(Boom(), ["/nix/store/aaa-x"])
+    assert "::warning::Boom push failed" in capsys.readouterr().out
+
+
+def test_audience_requires_https() -> None:
+    with pytest.raises(RuntimeError):
+        stream._audience("http://n.example")
+
+
+def test_mint_script_neutralizes_hostile_audience(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(stream.tempfile, "tempdir", str(tmp_path))
+    hostile = 'x" + 1 + "'
+    body = Path(stream._mint(hostile)).read_text()
+    ast.parse(body)
+    assert hostile not in body
+    assert "x%22%20%2B%201%20%2B%20%22" in body
+
+
+def test_push_continues_after_failed_batch(
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    class Flaky:
+        def __init__(self) -> None:
+            self.name = "Flaky"
+            self.attempts = 0
+
+        def setup(self) -> None:
+            pass
+
+        def push(self, paths: list[str]) -> None:
+            self.attempts += 1
+            if self.attempts == 1:
+                raise RuntimeError("transient")
+
+    flaky = Flaky()
+    push(flaky, [f"/nix/store/{i}" for i in range(1500)])
+    assert flaky.attempts == 2
+    assert capsys.readouterr().out.count("::warning::Flaky push failed") == 1
+
+
+def test_ready_reports_working_backend(monkeypatch: pytest.MonkeyPatch) -> None:
+    _record(monkeypatch)
+    monkeypatch.setenv("CACHIX_AUTH_TOKEN", "t")
+    assert ready(Cachix("d")) is True
+
+
+class _Sink:
+    def __init__(self, fail_setup: bool = False) -> None:
+        self.name = "Sink"
+        self.fail_setup = fail_setup
+        self.setups = 0
+        self.pushed: list[str] = []
+        self.idents: list[int] = []
+
+    def setup(self) -> None:
+        self.setups += 1
+        if self.fail_setup:
+            raise RuntimeError("no")
+
+    def push(self, paths: list[str]) -> None:
+        self.idents.append(threading.get_ident())
+        self.pushed.extend(paths)
+
+
+def test_stream_backend_drains_then_exits_on_sentinel(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    spool = tmp_path / "spool"
+    done = tmp_path / "done"
+    spool.write_text("/nix/store/aaa-x\n/nix/store/bbb-y\n")
+    done.touch()
+    sink = _Sink()
+    monkeypatch.setattr(stream.time, "sleep", lambda _: pytest.fail("loop did not exit"))
+    stream_backend(sink, spool, done)
+    assert sink.pushed == ["/nix/store/aaa-x", "/nix/store/bbb-y"]
+
+
+def test_stream_backend_exits_when_setup_fails(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    spool = tmp_path / "spool"
+    done = tmp_path / "done"
+    spool.write_text("/nix/store/aaa-x\n")
+    sink = _Sink(fail_setup=True)
+    # no sentinel, the disabled backend must still return
+    monkeypatch.setattr(stream.time, "sleep", lambda _: pytest.fail("loop did not exit"))
+    stream_backend(sink, spool, done)
+    assert sink.pushed == []
+    assert "Sink disabled" in capsys.readouterr().out
+
+
+def test_mode_stream_runs_one_thread_per_backend(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    spool = tmp_path / "spool"
+    done = tmp_path / "done"
+    spool.write_text("/nix/store/aaa-x\n")
+    done.touch()
+    sinks = [_Sink(), _Sink()]
+    monkeypatch.setattr(stream, "backends", lambda: list(sinks))
+    mode_stream(spool, done)
+    assert sinks[0].pushed == ["/nix/store/aaa-x"]
+    assert sinks[1].pushed == ["/nix/store/aaa-x"]
+    # one real thread per backend, neither on the caller's thread
+    idents = {sinks[0].idents[0], sinks[1].idents[0]}
+    assert len(idents) == 2
+    assert threading.get_ident() not in idents
+
+
+def test_mode_stream_without_backends_warns_and_returns(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    monkeypatch.setattr(stream, "backends", list)
+    mode_stream(tmp_path / "spool", tmp_path / "done")
+    assert "No cache backend configured" in capsys.readouterr().out
+
+
+def test_stream_backend_streams_across_polls(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    spool = tmp_path / "spool"
+    done = tmp_path / "done"
+    spool.write_text("")
+    sink = _Sink()
+    ticks: list[int] = []
+
+    def tick(_: float) -> None:
+        ticks.append(1)
+        if len(ticks) == 1:
+            spool.write_text("/nix/store/aaa-x\n")
+        elif len(ticks) == 2:
+            spool.write_text("/nix/store/aaa-x\n/nix/store/bbb-y\n")
+            done.touch()
+        elif len(ticks) > 5:
+            pytest.fail("loop did not terminate")
+
+    monkeypatch.setattr(stream.time, "sleep", tick)
+    stream_backend(sink, spool, done)
+    assert sink.pushed == ["/nix/store/aaa-x", "/nix/store/bbb-y"]
+    assert sink.setups == 1
+
+
+def test_stream_backend_warns_when_reader_dies(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    def boom(spool: Path, offset: int) -> tuple[list[str], int]:
+        raise PermissionError("spool unreadable")
+
+    monkeypatch.setattr(stream, "read_new", boom)
+    stream_backend(_Sink(), tmp_path / "spool", tmp_path / "done")
+    assert "::warning::Sink streamer stopped" in capsys.readouterr().out
+
+
+def test_mode_final_dedups_and_pushes_spool(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    spool = tmp_path / "spool"
+    spool.write_text("/nix/store/aaa-x\n/nix/store/aaa-x\n/nix/store/bbb-y\n")
+    monkeypatch.delenv("BUILD_OUTCOME", raising=False)
+    sink = _Sink()
+    monkeypatch.setattr(stream, "backends", lambda: [sink])
+    mode_final(spool, tmp_path / "done", tmp_path / "pid", tmp_path / "log")
+    assert sink.pushed == ["/nix/store/aaa-x", "/nix/store/bbb-y"]
+    assert (tmp_path / "done").exists()
+
+
+def test_mode_final_appends_outputs_only_on_success(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    spool = tmp_path / "spool"
+    spool.write_text("/nix/store/aaa-x\n")
+    monkeypatch.setenv("BUILD_OUTCOME", "success")
+    monkeypatch.setenv("INSTALLABLE", ".#pkg")
+    seen: list[str] = []
+
+    def outputs(installable: str) -> list[str]:
+        seen.append(installable)
+        return ["/nix/store/aaa-x", "/nix/store/fff-out"]
+
+    monkeypatch.setattr(stream, "_outputs", outputs)
+    sink = _Sink()
+    monkeypatch.setattr(stream, "backends", lambda: [sink])
+    mode_final(spool, tmp_path / "done", tmp_path / "pid", tmp_path / "log")
+    assert seen == [".#pkg"]
+    # finals overlap the spool when the top drv was built locally, dedup holds
+    assert sink.pushed == ["/nix/store/aaa-x", "/nix/store/fff-out"]
+
+
+def test_mode_final_failure_outcome_skips_outputs(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    spool = tmp_path / "spool"
+    spool.write_text("/nix/store/aaa-x\n")
+    monkeypatch.setenv("BUILD_OUTCOME", "failure")
+
+    def boom(installable: str) -> list[str]:
+        raise AssertionError("must not resolve outputs on failure")
+
+    monkeypatch.setattr(stream, "_outputs", boom)
+    sink = _Sink()
+    monkeypatch.setattr(stream, "backends", lambda: [sink])
+    mode_final(spool, tmp_path / "done", tmp_path / "pid", tmp_path / "log")
+    assert sink.pushed == ["/nix/store/aaa-x"]
+
+
+def test_mode_final_empty_set_pushes_nothing(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.delenv("BUILD_OUTCOME", raising=False)
+    called = False
+
+    def probe() -> list[stream.Backend]:
+        nonlocal called
+        called = True
+        return []
+
+    monkeypatch.setattr(stream, "backends", probe)
+    mode_final(
+        tmp_path / "spool", tmp_path / "done", tmp_path / "pid", tmp_path / "log"
+    )
+    assert called is False
+
+
+def test_mode_final_replays_streamer_log(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    (tmp_path / "log").write_text("hello from the streamer\n")
+    monkeypatch.delenv("BUILD_OUTCOME", raising=False)
+    monkeypatch.setattr(stream, "backends", list)
+    mode_final(
+        tmp_path / "spool", tmp_path / "done", tmp_path / "pid", tmp_path / "log"
+    )
+    out = capsys.readouterr().out
+    assert "::group::Streamer log" in out
+    assert "hello from the streamer" in out
+    assert "::endgroup::" in out
+
+
+def test_mode_final_waits_for_dead_pid_instantly(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    # a pid that cannot exist, kill(pid, 0) raises and the wait returns
+    (tmp_path / "pid").write_text("99999999")
+    monkeypatch.delenv("BUILD_OUTCOME", raising=False)
+    monkeypatch.setattr(stream, "backends", list)
+    mode_final(
+        tmp_path / "spool", tmp_path / "done", tmp_path / "pid", tmp_path / "log"
+    )
+
+
+def test_cli_exits_zero_without_backends(tmp_path: Path) -> None:
+    # run by file path exactly as the workflow does
+    env = {"RUNNER_TEMP": str(tmp_path), "ATELIER_SPOOL": str(tmp_path / "spool")}
+    for mode in ("stream", "final"):
+        proc = subprocess.run(
+            [sys.executable, stream.__file__, "--mode", mode],
+            env=env,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        assert proc.returncode == 0, proc.stderr
+
+
+def test_outputs_resolves_and_surfaces_stderr(
+    monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    seen: list[list[str]] = []
+
+    def ok(argv: list[str], **kw: object) -> subprocess.CompletedProcess[str]:
+        seen.append(list(argv))
+        return subprocess.CompletedProcess(
+            argv, 0, stdout="/nix/store/fff-out\n\n", stderr=""
+        )
+
+    monkeypatch.setattr(stream.subprocess, "run", ok)
+    assert stream._outputs(".#pkg") == ["/nix/store/fff-out"]
+    assert seen == [["nix", "build", ".#pkg^*", "--no-link", "--print-out-paths"]]
+
+    def boom(argv: list[str], **kw: object) -> subprocess.CompletedProcess[str]:
+        raise subprocess.CalledProcessError(1, argv, stderr="error: attribute missing")
+
+    monkeypatch.setattr(stream.subprocess, "run", boom)
+    assert stream._outputs(".#pkg") == []
+    out = capsys.readouterr().out
+    assert "error: attribute missing" in out
+    assert "::warning::Final output resolution failed" in out
+
+
+def test_wait_warns_on_timeout(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    # a live pid with a zero budget takes the timeout path immediately
+    (tmp_path / "pid").write_text(str(os.getpid()))
+    monkeypatch.setattr(stream, "WAIT", 0.0)
+    stream._wait(tmp_path / "pid")
+    assert "Streamer still running" in capsys.readouterr().out
+
+
+def test_replay_tolerates_bad_bytes_and_missing_newline(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    log = tmp_path / "log"
+    log.write_bytes(b"partial \xff line without newline")
+    stream._replay(log)
+    out = capsys.readouterr().out
+    assert out.startswith("::group::Streamer log\n")
+    assert out.endswith("\n::endgroup::\n")
+
+
+def test_cli_missing_runner_temp_warns_and_exits_zero() -> None:
+    proc = subprocess.run(
+        [sys.executable, stream.__file__, "--mode", "final"],
+        env={},
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert proc.returncode == 0
+    assert "::warning::Cache push failed" in proc.stdout
+
+
+def test_cli_final_replays_log_and_touches_sentinel(tmp_path: Path) -> None:
+    # pins the RUNNER_TEMP file names build.yaml writes
+    (tmp_path / "atelier-stream.log").write_text("streamer said hi\n")
+    proc = subprocess.run(
+        [sys.executable, stream.__file__, "--mode", "final"],
+        env={"RUNNER_TEMP": str(tmp_path), "ATELIER_SPOOL": str(tmp_path / "spool")},
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert proc.returncode == 0
+    assert "streamer said hi" in proc.stdout
+    assert (tmp_path / "atelier-stream.done").exists()
+
+
+def test_main_uses_the_runner_temp_names_build_yaml_writes(
+    monkeypatch: pytest.MonkeyPatch, tmp_path: Path
+) -> None:
+    seen: list[Path] = []
+    monkeypatch.setenv("RUNNER_TEMP", str(tmp_path))
+    monkeypatch.setattr(stream, "mode_final", lambda *a: seen.extend(a))
+    monkeypatch.setattr(sys, "argv", ["stream.py", "--mode", "final"])
+    assert stream.main() == 0
+    assert [p.name for p in seen[1:]] == [
+        "atelier-stream.done",
+        "atelier-stream.pid",
+        "atelier-stream.log",
+    ]
+
+
+def test_push_warning_omits_argv(capsys: pytest.CaptureFixture[str]) -> None:
+    class Argv:
+        name = "Argv"
+
+        def setup(self) -> None:
+            pass
+
+        def push(self, paths: list[str]) -> None:
+            raise subprocess.CalledProcessError(2, ["attic", "push", *paths])
+
+    push(Argv(), ["/nix/store/aaa-x"])
+    out = capsys.readouterr().out
+    assert "::warning::Argv push failed: exit 2" in out
+    assert "/nix/store/aaa-x" not in out
diff --git a/uv.lock b/uv.lock
index 7a715ee..e3c4e87 100644
--- a/uv.lock
+++ b/uv.lock
@@ -4,7 +4,7 @@ requires-python = ">=3.14"
 
 [[package]]
 name = "atelier"
-version = "2026.816.0"
+version = "2026.819.0"
 source = { editable = "." }
 dependencies = [
     { name = "click" },