git:20261025.067b5df to git:20261102.8121c63

2 added, 1 removed. Audit A to A.

# langalpha server

FastAPI server for langalpha — REST + SSE APIs for PTC-agent workflow execution, conversation/thread management, and workspaces. Endpoints live under `/api/v1/threads/*` and `/api/v1/workspaces/*` (discover the current set from the routers in `app/`).

> Single source of truth for AI coding agents in `src/server/`. `CLAUDE.md` imports this via `@AGENTS.md`; Codex/Cursor read it directly. Edit here, not there.

## Layering (put code in the right tier)

`app/` = route definitions + request orchestration (no raw SQL — DB access goes through services) · `handlers/` = per-request logic (chat streaming, workflow control, the error funnel) · `services/` = business logic + process singletons (session/workspace managers, `persistence/` DB services) · `database/` = raw-SQL query + pool layer · `models/` = Pydantic request/response · `dependencies/` = FastAPI `Depends` gates (auth + request admission).

Inside those tiers, the run machinery groups into **domain namespaces**: `app/threads/` (thread routes), `handlers/chat/` (the run generator, request prep, stream readers), `services/runs/` (the run lifecycle: admission → coordinator → executor → finalization → recovery, plus SSE production/transport), `services/report_back/` (report-back hooks: `handoff/` for the Chief of Staff's hand-offs to an analyst, `subagent.py` for subagents), and `database/runs/` (run-ledger SQL: lifecycle, outbox, subagent runs). The namespaces live *inside* the horizontal contract — they never invert it (`services/**` must not import `handlers/**`; enforced by `tests/unit/server/test_import_layering.py`).

## Landmines (non-obvious)

- **Two separate async psycopg3 pools, opened/closed in the `app/setup.py` lifespan** — the LangGraph checkpointer (workflow state; the LangGraph Store shares it) and the conversation-history (app-data) pool. Don't reuse one for the other.

- **Mid-flight reconnection is Redis-only, not the checkpoint.** Live events are buffered in a per-run Redis **Stream** `workflow:stream:{thread_id}:{run_id}` (XADD + `MAXLEN` trim, 24h TTL, 150k-event cap). `GET /api/v1/threads/{id}/messages/stream` XREAD-BLOCK-replays the buffer then tails live; `last_event_id` is a resume cursor (resumes at seq N+1), not post-hoc dedup. An abandoned running workflow is reaped after ~6h; a completed workflow's reconnect returns `410` once its task-info key expires.

- **Historical replay is projected from the LangGraph checkpoint** (re-homed in #315). `GET /api/v1/threads/{id}/messages/replay` (`source=auto`) rebuilds the transcript from checkpoint state via `CheckpointHistoryReader` + a pure projector (`services/history/`); a turn the checkpoint cannot cover replays as its user message, the stream its row archived (`replay_facts.archive`) if it kept one, and its closing rows (`replay/stubs.py`); one that answered and archived nothing closes on a `replay_unavailable` error frame, logged as an error. A turn settled while its live stream was archived also places what only that archive kept (widget data, evicted results, reasoning durations, signal rows in position) from its row's `replay_facts.legacy` (`services/history/replay/legacy.py`). **So the durable replay contract is the checkpoint state schema: `messages` (a `DeltaChannel`) + the `ui` channel** (`langgraph.graph.ui`; an id-keyed upsert accumulator of compact, non-rederivable records — image-URL maps, `model_fallback` notices).

- **What silently breaks replay of already-stored checkpoints:** renaming a `ui`-record's `name`/`props`, or emitting one without a stable pre-stamped `id` (the reducer upserts by id — keep the projector's field whitelists in sync); putting rederivable or bulk data in graph state (the projector rebuilds artifacts from tool-call *args* + message `additional_kwargs`, and >32KB widget data must be a content-addressed `data_ref`, not inlined; the one exception is a failed upload, whose data stays inline up to 500 KB so the widget still reloads); or adding a chat-wire SSE type without a checkpoint- or table-sourced home — `tests/unit/server/services/history/test_event_ledger.py` fails CI until every type is classified, so build the replay source before you emit.

- **Error funnel (`handlers/chat/error_handling.py`, `classify_error()`):** recoverable errors (DB/connection drop, timeout, network, API 5xx/429) emit a `retry` SSE event (`auto_retry`; thread → `interrupted`, and the client/gateway re-submits) up to 3× (`get_max_workflow_retries`), then convert to a terminal `error`. Non-recoverable *code bugs* (`AttributeError`/`NameError`/`TypeError`/`ImportError`/`SyntaxError`/`KeyError`) fail immediately with a stable `error` payload. Each of these frames names its `recovery`: `retry` when a failed run is on the ledger (`POST .../retry` with its `run_id`; a message that never reached the graph runs again from the turn's query row, else the checkpoint replays) or `resend` when no run started (the client sends the same request again). A reader error after the executor holds the run emits neither: the stream ends and the client reconnects. Don't add ad-hoc try/except that swallows this path.

- **Cancelled ≠ disconnected:** an explicit cancel stops the workflow; a client disconnect lets it keep running in the background (`LocalRunExecutor` keeps executing; the run's `in_progress` ledger row is the durable record), and a reconnect resumes the live stream.

- **The file-mount endpoint (`app/livefs.py`) authenticates by the computer's mount token, never user auth.** The token names the user and computer; the endpoint reuses the file tools' store routes, so a write rule lives once, in the route. Every write carries `If-Match` (the version read) or `If-None-Match: *` (428 otherwise), because a blind write to a profile file would apply a diff against rows the writer never saw. A route with versions of its own (`DbJsonRoute`: profile, automations) serves and checks those instead of a content hash, so a run moving an automation's `state` never conflicts a save. Whether a mount serves, and which folders it is linked into, is recorded in Postgres (the token row's `served_*` columns, `livefs_links`) and read once at each turn's start; `MachineState.livefs` (a `MountTracker`) holds that read plus this worker's backoffs and tasks in flight, and commands never read the rows. `services/livefs/mount.py`'s `serve` and `link` are idempotent, and the daemon stays the truth: a command that meets it dead restarts it whatever the rows say. A restart leaves links that answer ENOENT, which a command cannot tell from a missing file, so one the server learns of (a reconnect that booted the sandbox or found no mount, the platform-secret scrub) is written down to the rows before anything reads them again, and a command that met a missing file under the mount asks the sandbox, at most once a minute on each worker, whether the mount itself is gone.

- - **A Host under `PREVIEW_PROXY_BASE_URL` never reaches an app route.** `preview_proxy/` claims it ahead of every middleware, tracing included: `install_preview_proxy` wraps the *built* stack, so it must stay the last thing `setup.py` does to the app's middleware, and a route or middleware added for previews belongs inside the proxy. A preview URL reaches a browser only through `preview_capabilities.publish`, which passes a provider URL made for browsers and exchanges any other (a private E2B port needs a header) for a capability in the hostname; the capability, its sandbox's traffic header and its revocation counter live in Redis, so any worker serves it. Cookies cross the proxy untouched because `check_preview_proxy_config` refuses a base on the app's registrable domain; the proxy cannot keep the app's cookies from preview code, only that check can.

+ - **A Host under `PREVIEW_PROXY_BASE_URL` never reaches an app route.** `preview_proxy/` claims it ahead of every middleware, tracing included: `install_preview_proxy` wraps the *built* stack, so it must stay the last thing `setup.py` does to the app's middleware, and a route or middleware added for previews belongs inside the proxy. A preview URL reaches a browser only through `preview_capabilities.publish`, which passes a provider URL made for browsers and exchanges any other (a private E2B port needs a header) for a capability in the hostname; the capability, its sandbox's traffic header and its revocation counter live in Redis, so any worker serves it. Cookies cross the proxy untouched because `check_preview_proxy_config` refuses a base on the app's registrable domain; the proxy cannot keep the app's cookies from preview code, only that check can. The base also serves published artifact versions, each artifact at `artifact-<artifact id>.<base>` from Postgres and object storage, never a sandbox (`preview_proxy/artifacts.py`, branched before capability resolution): the artifact id names the origin, so its storage outlasts a grant, and the credential is a grant in the path, never a cookie, since the frame is cross-site to the app. The owner's grant opens any version; a visitor's names one version of a shared link and is checked against the link's row on every fetch, so an Update or an unshare ends it on every worker. Without a base the same files are served sandboxed on the API origin (`/api/v1/artifacts/g/...`, `app/artifact_frames.py`), which answers 404 when a base is set.

+ - **A file comment reaches the agent only inside a chat send, and a send the server refuses gives it back.** `ChatRequest.artifact_comment_ids` (with a `request_key` and `reading`) names comments on files; the generator claims their unsent owner messages under the key right after the dedup probe, under the admission lock, because admission needs the rendered message (`CommentSendScope` in `services/artifacts/comments.py`, `database/artifacts.claim_comments`). The claim takes only comments whose thread rule allows the sending thread (the owner's, not archived, in the file's workspace or the owner's Home), so a send that starts a thread makes it first, untitled, and titles it once admitted. The message is the typed note, then one `<file-comments>` block, and is what the query row, a steer and `/retry` carry; `artifact_comments` beside it (query metadata, the steer payload, `steering_accepted`, `steering_delivered` entries) is what a reader is shown instead. A refusal before a run or a steer took the send (a 409, an error, a disconnect) releases what the key marked and deletes a thread the send made if no turn ever ran there (`delete_unused_thread`), each unless `request_key_has_run` or `steer_key_taken` finds the key taken, read under the comments' or the thread's row lock, so a copy of the send another worker admitted keeps both; an admitted send marks its messages again, in case such a release landed after its claim. A comment's `delivery` is `sent` exactly when none of its owner messages is unsent, and `sent_from` keeps whether it was a note or queued, which a release restores, queued for the thread it was just sent to. `reply_to_comment` answers only a comment whose last sent owner message went to the replying thread.

- **A wakeup's notice is delivered through the outbox alert set, at least once.** The fire and its `hook_outbox` row commit together, and a series keeps at most one unread fire waiting: later slots are counted as skipped on the one that arrives, and the last always arrives. A notice joins a running turn through its report inbox when that turn can steer, and otherwise starts a `query_type="wakeup"` turn; a retry finds a wake already admitted by its request key and stamps the notice to that run instead of starting another (`services/wakeups/delivery.py`). The SQL rules are pinned in `tests/integration/test_wakeups_db.py`.

- **A background command's notice commits with its `command_runs` finalize, and one held for the next turn is never due.** A Bash or ExecuteCode job handed off past the tool's wait is a `command_runs` row from then on; its one guarded finalize enqueues the notice in the same transaction, and when it reports follows from the end (`command_runs.Outcome.report`). One due now is a `command_report_back` on the thread's own chain, like a subagent's report (a live turn takes it in, else `report_back/command.py` posts a `system` turn). One for the next turn (a command the user stopped, one the after-turn hour stopped, or one whose computer went away) is a `command_notice`, an alert registered in `outbox._ALERTS` with `wakes=False`: it waits on `command_notice:<thread>` at `next_retry_at='infinity'`, no claim ever makes it due, a turn's ending only drops a hold on it, and only the next live turn's steer inbox takes it in, since a turn of its own would start the computer again. Whether either is owed is `command_runs.COMMAND_OWED_SQL`; one that no turn will take in (its thread archived or deleted, its workspace deleted) is closed by `close_unowed_waiting_alerts`, as is a `command_notice` still waiting after a week (`COMMAND_NOTICE_TTL_DAYS`), since every waiting row costs each claim's scan; its `command_runs` row still says how the command ended. A command's end is delivered once: `result_delivered_at` is stamped when the notice reaches a turn or a `ThreadStatus` look shows the end first, and the executor asks again before posting, so an end already shown never opens a turn. The watcher (`services/commands/watcher.py`, every worker) claims rows by `next_check_at` with a lease longer than its per-computer budget, reads each computer's jobs in one exec, settles them by `services/commands/settle.py`, and waits out a computer that is starting or stopping rather than judging its commands lost.

- **`steer_only: true`** on `POST .../messages` only steers an in-flight workflow; if none is running it's rejected with an admission-conflict `error` event (`error_type="admission_conflict"`, `code="not_running"`) — it never starts a new turn.

## Config

`.env` = credentials (DB, API keys) + `HOST_MODE=oss|platform` (in `platform` mode langalpha calls the platform to gate a turn; `oss` runs standalone); `config.yaml` = behavioral settings (`workflow_timeout` 21600s, `sse_keepalive_interval` 15s, CORS `allowed_origins`), read via typed accessors in `src/config/settings.py` (`get_workflow_timeout`, …) — not `os.getenv`.

## SSE events

Chat-wire SSE events are emitted from **multiple layers** — the main turn in `services/runs/sse_producer.py` (content deltas `message_chunk`/`tool_call_chunks`/`tool_calls`/`tool_call_result`, plus `artifact`, `interrupt`, `error`/`warning`, `metadata`, …) and the subagent/reconnect sentinels from `services/runs/executor.py` + `handlers/chat/run_stream_reader.py`. No single file lists them all. The **authoritative, CI-enforced** index — with each type's replay home (checkpoint-projected / table-sourced / live-only) — is `tests/unit/server/services/history/test_event_ledger.py`; add a new type there (with a replay home) before you emit it. (`:keepalive` is an SSE comment, not an event.) **Lane ownership:** frames from a background task's run never ride the main stream (the per-task channel owns them; the main stream carries only main content + the parent-owned task artifacts). It is enforced at the source, where `arun_subagent_streaming` detaches every subagent run from its launcher, and locked by `tests/unit/middleware/test_subagent_run_isolation.py`. The frozen target for run-scoped identity, cursors, causal anchors, and retention is `src/server/STREAM_CONTRACT_V2.md`.

## Multi-worker (review checklist)

The server runs `--workers N`: a request, its SSE consumer, the outbox drainer, and the recovery scanner may each land on **different processes**. Truth lives in Postgres (run ledger + advisory-lock guards); Redis is coordination/transport; **process memory is execution context only** — no reader may treat it as truth. The failure signature of getting this wrong is always the same: *works when producer and consumer land on the same worker, breaks ~50% of the time at `--workers 2`* — one green run proves nothing.

Flag any diff that introduces one of these, and require a stated cross-worker story before it merges:

- **A module-level dict/set/registry consulted by a request path.** Another worker can't see it. Either derive from the ledger/Redis, or scope it to execution context the owning worker alone uses.

- **Treating local executor state as truth** — `LocalRunExecutor` handles, "is the task in my registry" liveness. Liveness = the advisory lock; status = the ledger row.

- **A pub/sub or Redis-only signal with no durable fallback.** A missed wake must degrade to a re-read (snapshot, rescan, reconcile), never to a stuck state.

- **Read-modify-write on shared rows without a CAS/guarded UPDATE or the right advisory lock.** Terminal transitions go through `RunCoordinator`'s single finalize; never a second writer.

- **Checkpoint writes outside a `WriterGuard` session.** The saver must share the guard's physical connection so lock loss fails writes structurally.

Verification: run `scripts/multiworker_gate/` (the automated subset of the fault-injection matrix) plus live wire checks (send/stream, reconnect, cancel, subagent report-back) against a 2-worker stack — and repeat wire assertions a few times, since worker routing is probabilistic.

## Working principles

- **Keep the tiers honest.** `app/` routes orchestrate; business logic lives in `handlers/`/`services/`, all DB access in `database/`, cross-cutting gates as `Depends`. Don't inline logic or raw SQL into a route because it's shorter.

- **One canonical path per concern — bypassing it is the bug.** Server-side LLM calls → `LLMService.complete` (never `create_llm()`, which skips BYOK/OAuth/prefs); tunables → `config/settings.py` accessors (never `os.getenv`); persistence → `services/persistence/` (never ad-hoc SQL); Redis keys → the `*_key()` builders. Reach for the wrapper before writing a second copy.

- **Async-first, non-blocking.** No sync drivers or blocking I/O on a request/SSE path; bound long work with `asyncio.wait_for`; open/close pools in the lifespan.

- **Errors go through the funnel, never around it.** Let `classify_error` decide recoverable-vs-terminal and don't swallow it; every chat-wire event needs a registered replay home (see Landmines). Sanitize error text before it reaches the wire.