Add a start/stop controller for the S2 storage sink - #401

Stacked on #401, which adds `events.S2StorageController`. This layer
replaces the ad-hoc `atomic.Pointer[events.S2StorageWriter]`,
`sync.WaitGroup`, and `startS2Writer` lifecycle in `cmd/api/main.go`
with that controller, leaving `main.go` net **−6 lines**. The sink still
starts at boot when the instance is not waiting for a fork identity or
has already applied one, and otherwise from the fork-identity hook, off
the handoff critical path per the contract documented on
`forkIdentityHandler`. `api.New` also takes the controller behind a new
`S2Storage` interface for the later telemetry-side integration; nothing
reads it yet.

Stream selection keeps both existing paths, and keeps the hook's payload
authoritative by construction rather than by disk state. Boot and
API-process restarts resolve the persisted applied payload through
`appliedS2Stream`. A successful fork hook stores its own payload's
`S2_STREAM` as an in-process override before starting the controller, so
resolution no longer depends on `appliedS2Stream` re-reading
`ReadyFile`, the applied marker, and the payload file. That matters
because the controller binds one stream for the life of the instance and
never retries: had that re-read fallen through to the boot
`config.S2Stream`, a fork would have written its telemetry into the
stream of the instance it forked from, permanently and silently. The
wrapper writes `ReadyFile` before starting the API whenever
fork-identity wait is armed, so this closes a latent failure mode rather
than an observed production failure. The override is atomic because
resolution and the asynchronous start run on different goroutines. A
missing `S2_STREAM` still falls back to the boot configuration.

Shutdown still drains the HTTP servers first, then S2, then OTLP. The
controller waits for an in-flight start and bounds the whole stop with
the existing 10-second context. The old 2-second abandonment guard is
gone: in `s2-sdk-go` v0.22.1, `AppendSession`
(`s2/append_session.go:54`) only starts a pump goroutine and returns,
and the transport session is created from `processInflightQueue` after
the first record submit, so `S2StorageWriter.Start` never holds `w.mu`
across a network dial. That guard was skipping the drain — losing the
shutdown window's events — to avoid a block that cannot occur.

Existing behavior is otherwise unchanged: missing credentials or a
missing stream keep the sink closed, concurrent starts open at most one
writer, a failed start stays retryable, a successful start emits exactly
one `S2 storage enabled` line, and a boot failure in the optional sink
does not crashloop the browser. `S2StorageWriter`, `StorageWriter`, the
telemetry handlers, and storage configuration are untouched. Deferring
startup to `PUT /telemetry` remains a later change.

Tests cover boot resolution, applied-identity restarts, the
missing-ready-file hook path, and the surrounding API packages. `go
build ./...` and `go vet ./...` are clean at `0fe7b21`. `go test $(go
list ./... | grep -v /e2e$) -count=1 -race` passes all **38** non-e2e
packages; an earlier run hit
`TestUpstreamManagerDetectsChromiumAndRestart` in `lib/devtoolsproxy` on
its own `t.TempDir()` teardown racing a Chromium profile directory — a
package this PR does not touch — and it has not recurred at `-count=3`.
`TestS2StreamResolverUsesHookPayloadWithoutReadyFile` fails under the
pre-fix mutation, returning `seed-stream` instead of `fork-stream`. A
headless image built from `0fe7b21` was booted in three modes, reading
the API log at `/var/log/supervisord/kernel-images-api` rather than
`docker logs`: with S2 unset, **0** `S2 storage enabled` lines; with
fake credentials, exactly **1**; and with
`KERNEL_FORK_IDENTITY_WAIT=true` plus a seed `S2_STREAM=a2-seed-stream`,
**0** at boot and exactly **1** after `POST /internal/fork-identity`
returned 204, bound to the payload's `a2-fork-stream` rather than the
seed. In each mode `GET /spec.yaml` and `GET /telemetry` returned 200,
`PUT /telemetry` 201, `POST /telemetry/events` 200 with the envelope
arriving on `GET /telemetry/stream`, and `docker stop -t 30` exited 0
with no drain warning. The full e2e suite passes against locally built
headless and headful images: `go test ./e2e/ -count=1 -timeout 110m`
with `E2E_CHROMIUM_HEADLESS_IMAGE` and `E2E_CHROMIUM_HEADFUL_IMAGE` set
gives **52 passed, 0 failed, 2 skipped** (34 subtests, all passing).
`TestOTLPExportForkIdentityRefresh` is the closest analogue to this
change: it drives the real `/internal/fork-identity` endpoint and
confirms the sibling sink retargets after the handoff. Of the two skips,
`TestReplayRecordingZombocomArchiveAudio` needs a network fixture, and
`TestS2StorageWriter` (`server/e2e/e2e_s2_storage_test.go`) self-skips
unless `S2_BASIN`, `S2_ACCESS_TOKEN`, and `S2_STREAM` are set — no S2
credentials are available here, so nothing has verified a record landing
in a *real* S2 stream.

That last gap was closed against a local mock instead. The SDK's append
session is a bidirectional protobuf stream, so a stand-in basin serving
h2c was pointed at via `S2_BASIN_ENDPOINT`, which required a throwaway
local patch (`newS2Storage` passes `nil` where `s2.LoadConfigFromEnv()`
would be needed) that was **not committed**. Against it: three published
events arrived as one batched append; three more published immediately
before `SIGTERM` all arrived during the drain, with the API log showing
`shutdown signal received` and no `stop failed`, `drain incomplete`, or
`ack failed`; and a fork container seeded with `S2_STREAM=seed-stream`
opened its session on `/v1/streams/fork-stream/records` after applying a
payload naming `fork-stream`. The mock also timestamps the dial: `S2
storage enabled` logged at `14:18:35` and the append session opened only
on first record submit, confirming the lazy-dial premise for removing
the 2-second guard.


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes telemetry persistence lifecycle and fork-time S2 stream
selection; a wrong bind is permanent, though behavior is covered by new
tests and matches prior boot/defer semantics.
> 
> **Overview**
> **Replaces the ad-hoc S2 writer lifecycle in `cmd/api/main.go`**
(`atomic.Pointer`, `WaitGroup`, `startS2Writer`) with
`events.S2StorageController`, wired through boot, fork-identity hook,
and shutdown (HTTP drain first, then bounded `Stop`).
> 
> **Adds `s2StreamResolver`** so stream name comes from persisted
`appliedS2Stream` on boot/restart, and from an **in-process atomic
override** when the fork hook runs—so the controller binds the payload’s
`S2_STREAM` even before `ReadyFile`/disk state would make
`appliedS2Stream` trustworthy (avoids silently appending to the parent’s
stream).
> 
> **Plumbs the controller into `api.New`** via a new `S2Storage`
interface on `ApiService` (parallel to OTLP); handlers do not use it
yet. Tests gain a hook-path case
(`TestS2StreamResolverUsesHookPayloadWithoutReadyFile`) and updated
`New(..., nil, nil)` call sites.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
68dabd8. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>