diff --git a/.docs/architecture.md b/.docs/architecture.md index e02a5ae804d5..ce5a0afe92a1 100644 --- a/.docs/architecture.md +++ b/.docs/architecture.md @@ -5,17 +5,139 @@ T3 Code runs as a **Node.js WebSocket server** that wraps `codex app-server` (JS ``` ┌─────────────────────────────────┐ │ Browser (React + Vite) │ -│ Connected via WebSocket │ +│ wsTransport (state machine) │ +│ Typed push decode at boundary │ └──────────┬──────────────────────┘ │ ws://localhost:3773 ┌──────────▼──────────────────────┐ │ apps/server (Node.js) │ │ WebSocket + HTTP static server │ -│ ProviderManager │ -│ CodexAppServerManager │ +│ ServerPushBus (ordered pushes) │ +│ ServerReadiness (startup gate) │ +│ OrchestrationEngine │ +│ ProviderService │ +│ CheckpointReactor │ +│ RuntimeReceiptBus │ └──────────┬──────────────────────┘ │ JSON-RPC over stdio ┌──────────▼──────────────────────┐ │ codex app-server │ └─────────────────────────────────┘ ``` + +## Components + +- **Browser app**: The React app renders session state, owns the client-side WebSocket transport, and treats typed push events as the boundary between server runtime details and UI state. + +- **Server**: `apps/server` is the main coordinator. It serves the web app, accepts WebSocket requests, waits for startup readiness before welcoming clients, and sends all outbound pushes through a single ordered push path. + +- **Provider runtime**: `codex app-server` does the actual provider/session work. The server talks to it over JSON-RPC on stdio and translates those runtime events into the app's orchestration model. + +- **Background workers**: Long-running async flows such as runtime ingestion, command reaction, and checkpoint processing run as queue-backed workers. This keeps work ordered, reduces timing races, and gives tests a deterministic way to wait for the system to go idle. + +- **Runtime signals**: The server emits lightweight typed receipts when important async milestones finish, such as checkpoint capture, diff finalization, or a turn becoming fully quiescent. Tests and orchestration code wait on these signals instead of polling internal state. + +## Event Lifecycle + +### Startup and client connect + +```mermaid +sequenceDiagram + participant Browser + participant Transport as WsTransport + participant Server as wsServer + participant Layers as serverLayers + participant Ready as ServerReadiness + participant Push as ServerPushBus + + Browser->>Transport: Load app and open WebSocket + Transport->>Server: Connect + Server->>Layers: Start runtime services + Server->>Ready: Wait for startup barriers + Ready-->>Server: Ready + Server->>Push: Publish server.welcome + Push-->>Transport: Ordered welcome push + Transport-->>Browser: Hydrate initial state +``` + +1. The browser boots [`WsTransport`][1] and registers typed listeners in [`wsNativeApi`][2]. +2. The server accepts the connection in [`wsServer`][3] and brings up the runtime graph defined in [`serverLayers`][7]. +3. [`ServerReadiness`][4] waits until the key startup barriers are complete. +4. Once the server is ready, [`wsServer`][3] sends `server.welcome` from the contracts in [`ws.ts`][6] through [`ServerPushBus`][5]. +5. The browser receives that ordered push through [`WsTransport`][1], and [`wsNativeApi`][2] uses it to seed local client state. + +### User turn flow + +```mermaid +sequenceDiagram + participant Browser + participant Transport as WsTransport + participant Server as wsServer + participant Provider as ProviderService + participant Codex as codex app-server + participant Ingest as ProviderRuntimeIngestion + participant Engine as OrchestrationEngine + participant Push as ServerPushBus + + Browser->>Transport: Send user action + Transport->>Server: Typed WebSocket request + Server->>Provider: Route request + Provider->>Codex: JSON-RPC over stdio + Codex-->>Ingest: Provider runtime events + Ingest->>Engine: Normalize into orchestration events + Engine-->>Server: Domain events + Server->>Push: Publish orchestration.domainEvent + Push-->>Browser: Typed push +``` + +1. A user action in the browser becomes a typed request through [`WsTransport`][1] and the browser API layer in [`nativeApi`][12]. +2. [`wsServer`][3] decodes that request using the shared WebSocket contracts in [`ws.ts`][6] and routes it to the right service. +3. [`ProviderService`][8] starts or resumes a session and talks to `codex app-server` over JSON-RPC on stdio. +4. Provider-native events are pulled back into the server by [`ProviderRuntimeIngestion`][9], which converts them into orchestration events. +5. [`OrchestrationEngine`][10] persists those events, updates the read model, and exposes them as domain events. +6. [`wsServer`][3] pushes those updates to the browser through [`ServerPushBus`][5] on channels defined in [`orchestration.ts`][11]. + +### Async completion flow + +```mermaid +sequenceDiagram + participant Server as wsServer + participant Worker as Queue-backed workers + participant Cmd as ProviderCommandReactor + participant Checkpoint as CheckpointReactor + participant Receipt as RuntimeReceiptBus + participant Push as ServerPushBus + participant Browser + + Server->>Worker: Enqueue follow-up work + Worker->>Cmd: Process provider commands + Worker->>Checkpoint: Process checkpoint tasks + Checkpoint->>Receipt: Publish completion receipt + Cmd-->>Server: Produce orchestration changes + Checkpoint-->>Server: Produce orchestration changes + Server->>Push: Publish resulting state updates + Push-->>Browser: User-visible push +``` + +1. Some work continues after the initial request returns, especially in [`ProviderRuntimeIngestion`][9], [`ProviderCommandReactor`][13], and [`CheckpointReactor`][14]. +2. These flows run as queue-backed workers using [`DrainableWorker`][16], which helps keep side effects ordered and test synchronization deterministic. +3. When a milestone completes, the server emits a typed receipt on [`RuntimeReceiptBus`][15], such as checkpoint completion or turn quiescence. +4. Tests and orchestration code wait on those receipts instead of polling git state, projections, or timers. +5. Any user-visible state changes produced by that async work still go back through [`wsServer`][3] and [`ServerPushBus`][5]. + +[1]: ../apps/web/src/wsTransport.ts +[2]: ../apps/web/src/wsNativeApi.ts +[3]: ../apps/server/src/wsServer.ts +[4]: ../apps/server/src/wsServer/readiness.ts +[5]: ../apps/server/src/wsServer/pushBus.ts +[6]: ../packages/contracts/src/ws.ts +[7]: ../apps/server/src/serverLayers.ts +[8]: ../apps/server/src/provider/Layers/ProviderService.ts +[9]: ../apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +[10]: ../apps/server/src/orchestration/Layers/OrchestrationEngine.ts +[11]: ../packages/contracts/src/orchestration.ts +[12]: ../apps/web/src/nativeApi.ts +[13]: ../apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +[14]: ../apps/server/src/orchestration/Layers/CheckpointReactor.ts +[15]: ../apps/server/src/orchestration/Layers/RuntimeReceiptBus.ts +[16]: ../packages/shared/src/DrainableWorker.ts diff --git a/.docs/provider-architecture.md b/.docs/provider-architecture.md index fe84ab02b1ae..794b6aa5d5e5 100644 --- a/.docs/provider-architecture.md +++ b/.docs/provider-architecture.md @@ -3,7 +3,9 @@ The web app communicates with the server via WebSocket using a simple JSON-RPC-style protocol: - **Request/Response**: `{ id, method, params }` → `{ id, result }` or `{ id, error }` -- **Push events**: `{ type: "push", channel, data }` for orchestration read-model updates +- **Push events**: typed envelopes with `channel`, `sequence` (monotonic per connection), and channel-specific `data` + +Push channels: `server.welcome`, `server.configUpdated`, `terminal.event`, `orchestration.domainEvent`. Payloads are schema-validated at the transport boundary (`wsTransport.ts`). Decode failures produce structured `WsDecodeDiagnostic` with `code`, `reason`, and path info. Methods mirror the `NativeApi` interface defined in `@t3tools/contracts`: @@ -12,3 +14,17 @@ Methods mirror the `NativeApi` interface defined in `@t3tools/contracts`: - `shell.openInEditor`, `server.getConfig` Codex is the only implemented provider. `claudeCode` is reserved in contracts/UI. + +## Client transport + +`wsTransport.ts` manages connection state: `connecting` → `open` → `reconnecting` → `closed` → `disposed`. Outbound requests are queued while disconnected and flushed on reconnect. Inbound pushes are decoded and validated at the boundary, then cached per channel. Subscribers can opt into `replayLatest` to receive the last push on subscribe. + +## Server-side orchestration layers + +Provider runtime events flow through queue-based workers: + +1. **ProviderRuntimeIngestion** — consumes provider runtime streams, emits orchestration commands +2. **ProviderCommandReactor** — reacts to orchestration intent events, dispatches provider calls +3. **CheckpointReactor** — captures git checkpoints on turn start/complete, publishes runtime receipts + +All three use `DrainableWorker` internally and expose `drain()` for deterministic test synchronization. diff --git a/.docs/workspace-layout.md b/.docs/workspace-layout.md index 721f094f528e..be88f2b603b5 100644 --- a/.docs/workspace-layout.md +++ b/.docs/workspace-layout.md @@ -3,4 +3,5 @@ - `/apps/server`: Node.js WebSocket server. Wraps Codex app-server, serves the built web app, and opens the browser on start. - `/apps/web`: React + Vite UI. Session control, conversation, and provider event rendering. Connects to the server via WebSocket. - `/apps/desktop`: Electron shell. Spawns a desktop-scoped `t3` backend process and loads the shared web app. -- `/packages/contracts`: Shared Zod schemas and TypeScript contracts for provider events, WebSocket protocol, and model/session types. +- `/packages/contracts`: Shared effect/Schema schemas and TypeScript contracts for provider events, WebSocket protocol, and model/session types. +- `/packages/shared`: Shared runtime utilities consumed by both server and web. Uses explicit subpath exports (e.g. `@t3tools/shared/git`, `@t3tools/shared/DrainableWorker`) — no barrel index. diff --git a/.mise.toml b/.mise.toml new file mode 100644 index 000000000000..364d9793d2c6 --- /dev/null +++ b/.mise.toml @@ -0,0 +1,3 @@ +[tools] +node = "24.13.1" +bun = "1.3.9" diff --git a/.plans/17-provider-neutral-runtime-determinism.md b/.plans/17-provider-neutral-runtime-determinism.md new file mode 100644 index 000000000000..7795cdff194a --- /dev/null +++ b/.plans/17-provider-neutral-runtime-determinism.md @@ -0,0 +1,88 @@ +# Plan: Provider-Neutral Runtime Determinism and Flake Elimination + +## Summary +Replace timing-sensitive websocket and orchestration behavior with explicit typed runtime boundaries, ordered push delivery, and server-owned completion receipts. The cutover is broad and single-shot: no compatibility shim, no mixed old/new transport. The design must reduce flakes without baking Codex-specific lifecycle semantics into generic runtime code. + +## Key Changes +### 1. Strengthen the generic boundaries, not the Codex boundary +- Keep `ProviderRuntimeEvent` as the canonical provider event contract and `ProviderService` as the only cross-provider facade. +- Do not expose raw Codex payloads or Codex event ordering outside `CodexAdapter.ts` and `codexAppServerManager.ts`. +- Do not expand `ProviderKind` in this change. The runtime stays provider-neutral by contract, while the product remains Codex-only in concrete support. + +### 2. Replace loose websocket envelopes with channel-indexed typed pushes +- Refactor `packages/contracts/src/ws.ts` so push messages are derived from a channel-to-schema map instead of `channel: string` plus `data: unknown`. +- Add `sequence: number` to every server push. Ordering becomes explicit and testable. +- Add structured decode diagnostics with stable machine fields: `code`, `reason`, `expected`, `actual`, `path`, optional `jsonOffset`. +- Remove runtime/test dependence on engine-specific pretty strings. Human-readable formatting remains a logging helper only. + +### 3. Introduce explicit server readiness and a single push pipeline +- Add a `ServerPushBus` service in `apps/server` backed by one ordered queue/pubsub path. All pushes go through it: `server.welcome`, `server.configUpdated`, terminal events, orchestration domain events. +- Add a `ServerReadiness` service with explicit barriers for: + - HTTP listening + - push bus ready + - keybindings runtime ready + - terminal subscriptions ready + - orchestration subscriptions ready +- Strengthen `server.welcome` semantics: it is emitted only after connection-scoped and server-scoped readiness is complete. +- `wsServer` should never publish directly from ad hoc background streams once the bus exists. + +### 4. Turn background watchers into explicit runtimes +- Extract keybindings watching into a `KeybindingsRuntime` service with `start`, `ready`, `snapshot`, and `changes`. +- Initial config load, startup sync, cache warmup, and watcher attachment complete before `ready` resolves. +- Keep the real `fs.watch` adapter thin. Most behavior tests use a fake watch source and deterministic change stream. + +### 5. Replace polling-based orchestration waiting with receipts +- Add server-owned completion receipts for operations that tests currently infer by polling: + - checkpoint capture complete + - checkpoint diff finalized + - turn processing quiesced +- A receipt means append, projection, and required side effects are complete. It must not mean the provider emitted a certain event sequence. +- Update the orchestration harness and checkpoint tests to await receipts/barriers instead of polling snapshots and git refs. + +### 6. Centralize client transport state and decoding +- Refactor `apps/web/src/wsTransport.ts` into an explicit connection state machine: `connecting`, `open`, `reconnecting`, `closed`, `disposed`. +- Decode and validate typed push payloads at the transport boundary, not downstream in `apps/web/src/wsNativeApi.ts`. +- Keep cached latest welcome/config behavior only if it is modeled as explicit state, not as a late-subscriber race workaround. + +### 7. Replace ad hoc test helpers with semantic test clients +- Add a shared `WsTestClient` for server/websocket tests: + - connect + - await semantic welcome + - await typed push by channel and optional predicate + - track `sequence` + - match RPC responses by id +- Add one orchestration test harness that waits on receipts/barriers instead of custom `waitForThread`, `waitForGitRef`, and arbitrary retry loops. +- Keep only a narrow set of integration tests for real filesystem/watch/socket behavior. Move behavioral assertions to deterministic unit-style harnesses. + +## Provider-Coupling Guardrails +- No generic runtime API may depend on Codex-native event names, thread IDs, or request payload shapes. +- No readiness barrier may be defined as “Codex has emitted X.” Readiness is owned by the server runtime, not by provider event order. +- No websocket channel payload may contain raw provider-native payloads unless the channel is explicitly debug/internal. +- Any provider-specific divergence must be exposed through provider capabilities from `ProviderService.getCapabilities()`, not `if provider === "codex"` branches in shared runtime code. +- Generic tests must use canonical `ProviderRuntimeEvent` fixtures. Codex-specific ordering and translation tests stay in adapter/app-server suites only. +- Keep UI/provider-specific knobs such as Codex-only options scoped to provider UX code. Do not pull them into generic transport or orchestration state. + +## Test Plan +- Contracts: + - schema tests for typed push envelopes and structured decode diagnostics + - ordering tests for `sequence` +- Server: + - readiness tests proving `server.welcome` cannot precede runtime readiness + - push bus tests proving terminal/config/orchestration pushes are serialized and typed + - keybindings runtime tests with fake watch source plus one real watcher integration test +- Orchestration: + - receipt tests proving checkpoint refs and projections are complete before completion signals resolve + - replacement of polling-based checkpoint/integration waits with receipt-based waits +- Web: + - transport tests for invalid JSON, invalid envelope, invalid payload, reconnect queue flushing, cached semantic state +- Validation gate: + - `bun run lint` + - `bun run typecheck` + - `mise exec -- bun run test` + - repeated full-suite run after cutover to confirm flake removal + +## Assumptions and Defaults +- This remains a single-provider product during the cutover, but the runtime contracts must stay provider-neutral. +- No backward-compatibility layer is required for old websocket push envelopes. +- The goal is deterministic runtime behavior first; reducing retries and sleeps in tests is a consequence, not the primary mechanism. +- If a completion signal cannot be expressed provider-neutrally, it does not belong in the shared runtime layer and must stay adapter-local. diff --git a/apps/server/integration/OrchestrationEngineHarness.integration.ts b/apps/server/integration/OrchestrationEngineHarness.integration.ts index b6ae7ee98255..2dc0a2fba728 100644 --- a/apps/server/integration/OrchestrationEngineHarness.integration.ts +++ b/apps/server/integration/OrchestrationEngineHarness.integration.ts @@ -15,6 +15,7 @@ import { Layer, ManagedRuntime, Option, + Ref, Schedule, Schema, Scope, @@ -45,6 +46,7 @@ import { CheckpointReactorLive } from "../src/orchestration/Layers/CheckpointRea import { OrchestrationEngineLive } from "../src/orchestration/Layers/OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "../src/orchestration/Layers/ProjectionPipeline.ts"; import { OrchestrationProjectionSnapshotQueryLive } from "../src/orchestration/Layers/ProjectionSnapshotQuery.ts"; +import { RuntimeReceiptBusLive } from "../src/orchestration/Layers/RuntimeReceiptBus.ts"; import { OrchestrationReactorLive } from "../src/orchestration/Layers/OrchestrationReactor.ts"; import { ProviderCommandReactorLive } from "../src/orchestration/Layers/ProviderCommandReactor.ts"; import { ProviderRuntimeIngestionLive } from "../src/orchestration/Layers/ProviderRuntimeIngestion.ts"; @@ -54,6 +56,10 @@ import { } from "../src/orchestration/Services/OrchestrationEngine.ts"; import { OrchestrationReactor } from "../src/orchestration/Services/OrchestrationReactor.ts"; import { ProjectionSnapshotQuery } from "../src/orchestration/Services/ProjectionSnapshotQuery.ts"; +import { + RuntimeReceiptBus, + type OrchestrationRuntimeReceipt, +} from "../src/orchestration/Services/RuntimeReceiptBus.ts"; import { makeTestProviderAdapterHarness, @@ -114,7 +120,7 @@ function waitFor( read: Effect.Effect, predicate: (value: A) => boolean, description: string, - timeoutMs = 3000, + timeoutMs = 10_000, ): Effect.Effect { const RETRY_SIGNAL = "wait_for_retry"; const retryIntervalMs = 10; @@ -185,6 +191,16 @@ export interface OrchestrationIntegrationHarness { }, never >; + readonly waitForReceipt: { + ( + predicate: (receipt: OrchestrationRuntimeReceipt) => boolean, + timeoutMs?: number, + ): Effect.Effect; + ( + predicate: (receipt: OrchestrationRuntimeReceipt) => receipt is Receipt, + timeoutMs?: number, + ): Effect.Effect; + }; readonly dispose: Effect.Effect; } @@ -268,6 +284,7 @@ export const makeOrchestrationIntegrationHarness = ( ProjectionPendingApprovalRepositoryLive, CheckpointStoreLive, providerLayer, + RuntimeReceiptBusLive, ); const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe( Layer.provideMerge(runtimeServicesLayer), @@ -322,11 +339,18 @@ export const makeOrchestrationIntegrationHarness = ( "load ProjectionPendingApprovalRepository service", () => runtime.runPromise(Effect.service(ProjectionPendingApprovalRepository)), ).pipe(Effect.orDie); + const runtimeReceiptBus = yield* tryRuntimePromise("load RuntimeReceiptBus service", () => + runtime.runPromise(Effect.service(RuntimeReceiptBus)), + ).pipe(Effect.orDie); const scope = yield* Scope.make("sequential"); yield* tryRuntimePromise("start OrchestrationReactor", () => runtime.runPromise(reactor.start.pipe(Scope.provide(scope))), ).pipe(Effect.orDie); + const receiptHistory = yield* Ref.make>([]); + yield* Stream.runForEach(runtimeReceiptBus.stream, (receipt) => + Ref.update(receiptHistory, (history) => [...history, receipt]).pipe(Effect.asVoid), + ).pipe(Effect.forkIn(scope)); yield* sleep(10); const waitForThread: OrchestrationIntegrationHarness["waitForThread"] = ( @@ -398,6 +422,32 @@ export const makeOrchestrationIntegrationHarness = ( never >; + function waitForReceipt( + predicate: (receipt: OrchestrationRuntimeReceipt) => boolean, + timeoutMs?: number, + ): Effect.Effect; + function waitForReceipt( + predicate: (receipt: OrchestrationRuntimeReceipt) => receipt is Receipt, + timeoutMs?: number, + ): Effect.Effect; + function waitForReceipt( + predicate: (receipt: OrchestrationRuntimeReceipt) => boolean, + timeoutMs?: number, + ) { + const readMatchingReceipt = Ref.get(receiptHistory).pipe( + Effect.map((history) => history.find(predicate)), + ); + + return waitFor( + readMatchingReceipt, + ( + receipt, + ): receipt is OrchestrationRuntimeReceipt => receipt !== undefined, + "runtime receipt", + timeoutMs, + ); + } + let disposed = false; const dispose = Effect.gen(function* () { if (disposed) { @@ -443,6 +493,7 @@ export const makeOrchestrationIntegrationHarness = ( waitForThread, waitForDomainEvent, waitForPendingApproval, + waitForReceipt, dispose, } satisfies OrchestrationIntegrationHarness; }); diff --git a/apps/server/integration/orchestrationEngine.integration.test.ts b/apps/server/integration/orchestrationEngine.integration.test.ts index 3b0a3a4002f8..42dcfe34f8cc 100644 --- a/apps/server/integration/orchestrationEngine.integration.test.ts +++ b/apps/server/integration/orchestrationEngine.integration.test.ts @@ -21,6 +21,10 @@ import { type OrchestrationIntegrationHarness, } from "./OrchestrationEngineHarness.integration.ts"; import { checkpointRefForThreadTurn } from "../src/checkpointing/Utils.ts"; +import type { + CheckpointDiffFinalizedReceipt, + TurnProcessingQuiescedReceipt, +} from "../src/orchestration/Services/RuntimeReceiptBus.ts"; const asMessageId = (value: string): MessageId => MessageId.makeUnsafe(value); const asProjectId = (value: string): ProjectId => ProjectId.makeUnsafe(value); @@ -187,6 +191,22 @@ it.live("runs a single turn end-to-end and persists checkpoint state in sqlite + messageId: "msg-user-single", text: "Say hello", }); + const finalizedReceipt = yield* harness.waitForReceipt( + (receipt): receipt is CheckpointDiffFinalizedReceipt => + receipt.type === "checkpoint.diff.finalized" && + receipt.threadId === THREAD_ID && + receipt.checkpointTurnCount === 1, + ); + if (finalizedReceipt.type !== "checkpoint.diff.finalized") { + throw new Error("Expected checkpoint.diff.finalized receipt."); + } + assert.equal(finalizedReceipt.status, "ready"); + yield* harness.waitForReceipt( + (receipt): receipt is TurnProcessingQuiescedReceipt => + receipt.type === "turn.processing.quiesced" && + receipt.threadId === THREAD_ID && + receipt.checkpointTurnCount === 1, + ); const thread = yield* harness.waitForThread( THREAD_ID, @@ -208,8 +228,6 @@ it.live("runs a single turn end-to-end and persists checkpoint state in sqlite + assert.equal(checkpointRows[0]?.status, "ready"); assert.deepEqual(checkpointRows[0]?.files, []); - yield* harness.waitForDomainEvent((event) => event.type === "thread.turn-diff-completed"); - const ref0 = checkpointRefForThreadTurn(THREAD_ID, 0); const ref1 = checkpointRefForThreadTurn(THREAD_ID, 1); assert.equal(gitRefExists(harness.workspaceDir, ref0), true); @@ -367,6 +385,12 @@ it.live("runs multi-turn file edits and persists checkpoint diffs", () => messageId: "msg-user-multi-1", text: "Make first edit", }); + yield* harness.waitForReceipt( + (receipt): receipt is CheckpointDiffFinalizedReceipt => + receipt.type === "checkpoint.diff.finalized" && + receipt.threadId === THREAD_ID && + receipt.checkpointTurnCount === 1, + ); yield* harness.waitForThread( THREAD_ID, @@ -408,6 +432,22 @@ it.live("runs multi-turn file edits and persists checkpoint diffs", () => messageId: "msg-user-multi-2", text: "Make second edit", }); + const secondReceipt = yield* harness.waitForReceipt( + (receipt): receipt is CheckpointDiffFinalizedReceipt => + receipt.type === "checkpoint.diff.finalized" && + receipt.threadId === THREAD_ID && + receipt.checkpointTurnCount === 2, + ); + if (secondReceipt.type !== "checkpoint.diff.finalized") { + throw new Error("Expected checkpoint.diff.finalized receipt."); + } + assert.equal(secondReceipt.status, "ready"); + yield* harness.waitForReceipt( + (receipt): receipt is TurnProcessingQuiescedReceipt => + receipt.type === "turn.processing.quiesced" && + receipt.threadId === THREAD_ID && + receipt.checkpointTurnCount === 2, + ); const secondTurnThread = yield* harness.waitForThread( THREAD_ID, diff --git a/apps/server/src/git/Layers/GitCore.ts b/apps/server/src/git/Layers/GitCore.ts index a288b2f3799c..b535bfccd879 100644 --- a/apps/server/src/git/Layers/GitCore.ts +++ b/apps/server/src/git/Layers/GitCore.ts @@ -524,7 +524,7 @@ const makeGitCore = Effect.gen(function* () { const statusDetails: GitCoreShape["statusDetails"] = (cwd) => Effect.gen(function* () { - yield* refreshStatusUpstreamIfStale(cwd).pipe(Effect.catch(() => Effect.void)); + yield* refreshStatusUpstreamIfStale(cwd).pipe(Effect.ignoreCause({ log: true })); const [statusStdout, unstagedNumstatStdout, stagedNumstatStdout] = yield* Effect.all( [ @@ -1162,7 +1162,7 @@ const makeGitCore = Effect.gen(function* () { // Refresh upstream refs in the background so checkout remains responsive. yield* Effect.forkScoped( - refreshCheckedOutBranchUpstream(input.cwd).pipe(Effect.catch(() => Effect.void)), + refreshCheckedOutBranchUpstream(input.cwd).pipe(Effect.ignoreCause({ log: true })), ); }); diff --git a/apps/server/src/keybindings.ts b/apps/server/src/keybindings.ts index 28fa757c2c96..dd474fa6c7eb 100644 --- a/apps/server/src/keybindings.ts +++ b/apps/server/src/keybindings.ts @@ -22,7 +22,9 @@ import { Array, Cache, Cause, + Deferred, Effect, + Exit, FileSystem, Path, Layer, @@ -33,11 +35,14 @@ import { SchemaGetter, SchemaIssue, SchemaTransformation, + Ref, ServiceMap, + Scope, Stream, } from "effect"; import * as Semaphore from "effect/Semaphore"; import { ServerConfig } from "./config"; +import { watchFileWithStatPolling } from "./watchFileWithStatPolling"; export class KeybindingsConfigError extends Schema.TaggedErrorClass()( "KeybindingsConfigParseError", @@ -422,6 +427,7 @@ export interface KeybindingsConfigState { } export interface KeybindingsChangeEvent { + readonly keybindings: ResolvedKeybindingsConfig; readonly issues: readonly ServerConfigIssue[]; } @@ -468,6 +474,22 @@ function mergeWithDefaultKeybindings(custom: ResolvedKeybindingsConfig): Resolve * KeybindingsShape - Service API for keybinding configuration operations. */ export interface KeybindingsShape { + /** + * Start the keybindings runtime and attach file watching. + * + * Safe to call multiple times. The first successful call establishes the + * runtime; later calls await the same startup. + */ + readonly start: Effect.Effect; + + /** + * Await keybindings runtime readiness. + * + * Readiness means the config directory exists, the watcher is attached, the + * startup sync has completed, and the current snapshot has been loaded. + */ + readonly ready: Effect.Effect; + /** * Ensure the on-disk keybindings file exists and includes all default * commands so newly-added defaults are backfilled on startup. @@ -480,9 +502,22 @@ export interface KeybindingsShape { readonly loadConfigState: Effect.Effect; /** - * Stream keybindings config change events. + * Read the latest keybindings snapshot from cache/disk. */ - readonly changes: Stream.Stream; + readonly getSnapshot: Effect.Effect; + + /** + * Subscribe to keybindings config change events. + * + * Returns a scoped subscription. Callers should subscribe in the + * current fiber (not inside a fork) to avoid missing events due to + * lazy subscription setup in `Stream.fromPubSub`. + */ + readonly subscribeChanges: Effect.Effect< + PubSub.Subscription, + never, + Scope.Scope + >; /** * Upsert a keybinding rule and persist the resulting configuration. @@ -509,9 +544,12 @@ const makeKeybindings = Effect.gen(function* () { const upsertSemaphore = yield* Semaphore.make(1); const resolvedConfigCacheKey = "resolved" as const; const changesPubSub = yield* PubSub.unbounded(); - - const emitChange = (issues: readonly ServerConfigIssue[]) => - PubSub.publish(changesPubSub, { issues }).pipe(Effect.asVoid); + const startedRef = yield* Ref.make(false); + const startedDeferred = yield* Deferred.make(); + const watcherScope = yield* Scope.make("sequential"); + yield* Effect.addFinalizer(() => Scope.close(watcherScope, Exit.void)); + const emitChange = (configState: KeybindingsConfigState) => + PubSub.publish(changesPubSub, configState).pipe(Effect.asVoid); const readConfigExists = fs.exists(keybindingsConfigPath).pipe( Effect.mapError( @@ -531,7 +569,7 @@ const makeKeybindings = Effect.gen(function* () { configPath: keybindingsConfigPath, detail: "failed to read keybindings config", cause, - }), + }), ), ); @@ -676,43 +714,10 @@ const makeKeybindings = Effect.gen(function* () { Effect.gen(function* () { yield* Cache.invalidate(resolvedConfigCache, resolvedConfigCacheKey); const configState = yield* loadConfigStateFromCacheOrDisk; - yield* emitChange(configState.issues); + yield* emitChange(configState); }), ); - const keybindingsConfigDir = path.dirname(keybindingsConfigPath); - const keybindingsConfigFile = path.basename(keybindingsConfigPath); - const keybindingsConfigPathResolved = path.resolve(keybindingsConfigPath); - yield* fs - .makeDirectory(keybindingsConfigDir, { recursive: true }) - .pipe(Effect.orElseSucceed(() => undefined)); - yield* Stream.runForEach(fs.watch(keybindingsConfigDir), (event) => { - const isTargetConfigEvent = - event.path === keybindingsConfigFile || - event.path === keybindingsConfigPath || - path.resolve(keybindingsConfigDir, event.path) === keybindingsConfigPathResolved; - if (!isTargetConfigEvent) { - return Effect.void; - } - return revalidateAndEmit.pipe( - Effect.catch((error) => - Effect.logWarning("failed to revalidate keybindings config after file update", { - path: keybindingsConfigPath, - detail: error.detail, - cause: error.cause, - }), - ), - ); - }).pipe( - Effect.catch((cause) => - Effect.logWarning("keybindings config watcher stopped unexpectedly", { - path: keybindingsConfigPath, - cause, - }), - ), - Effect.forkScoped, - ); - const syncDefaultKeybindingsOnStartup = upsertSemaphore.withPermits(1)( Effect.gen(function* () { const configExists = yield* readConfigExists; @@ -803,10 +808,85 @@ const makeKeybindings = Effect.gen(function* () { }), ); + const startWatcher = Effect.gen(function* () { + const keybindingsConfigDir = path.dirname(keybindingsConfigPath); + const keybindingsConfigFile = path.basename(keybindingsConfigPath); + const keybindingsConfigPathResolved = path.resolve(keybindingsConfigPath); + + yield* fs.makeDirectory(keybindingsConfigDir, { recursive: true }).pipe( + Effect.mapError( + (cause) => + new KeybindingsConfigError({ + configPath: keybindingsConfigPath, + detail: "failed to prepare keybindings config directory", + cause, + }), + ), + ); + + const revalidateAndEmitSafely = revalidateAndEmit.pipe( + Effect.ignoreCause({ log: true }), + ); + + // fs.watch (inotify/kqueue) gives instant detection on most platforms + // but can silently miss events on Linux containers. + yield* Stream.runForEach(fs.watch(keybindingsConfigDir), (event) => { + const isTargetConfigEvent = + event.path === keybindingsConfigFile || + event.path === keybindingsConfigPath || + path.resolve(keybindingsConfigDir, event.path) === keybindingsConfigPathResolved; + if (!isTargetConfigEvent) { + return Effect.void; + } + return revalidateAndEmitSafely; + }).pipe( + Effect.ignoreCause({ log: true }), + Effect.forkIn(watcherScope), + Effect.asVoid, + ); + + // fs.watchFile (stat-based polling on libuv timers) as a reliable + // backup. Unlike an Effect.sleep loop in a forked fiber, libuv + // timers fire regardless of Effect fiber scheduling pressure. + yield* Stream.runForEach( + watchFileWithStatPolling({ + filePath: keybindingsConfigPath, + pollIntervalMs: 100, + }), + () => revalidateAndEmitSafely, + ).pipe(Effect.forkIn(watcherScope), Effect.asVoid); + }); + + const start = Effect.gen(function* () { + const alreadyStarted = yield* Ref.get(startedRef); + if (alreadyStarted) { + return yield* Deferred.await(startedDeferred); + } + + yield* Ref.set(startedRef, true); + const startup = Effect.gen(function* () { + yield* startWatcher; + yield* syncDefaultKeybindingsOnStartup; + yield* Cache.invalidate(resolvedConfigCache, resolvedConfigCacheKey); + yield* loadConfigStateFromCacheOrDisk; + }); + + const startupExit = yield* Effect.exit(startup); + if (startupExit._tag === "Failure") { + yield* Deferred.failCause(startedDeferred, startupExit.cause).pipe(Effect.orDie); + return yield* Effect.failCause(startupExit.cause); + } + + yield* Deferred.succeed(startedDeferred, undefined).pipe(Effect.orDie); + }); + return { + start, + ready: Deferred.await(startedDeferred), syncDefaultKeybindingsOnStartup, loadConfigState: loadConfigStateFromCacheOrDisk, - changes: Stream.fromPubSub(changesPubSub), + getSnapshot: loadConfigStateFromCacheOrDisk, + subscribeChanges: PubSub.subscribe(changesPubSub), upsertKeybindingRule: (rule) => upsertSemaphore.withPermits(1)( Effect.gen(function* () { @@ -833,7 +913,10 @@ const makeKeybindings = Effect.gen(function* () { keybindings: nextResolved, issues: [], }); - yield* emitChange([]); + yield* emitChange({ + keybindings: nextResolved, + issues: [], + }); return nextResolved; }), ), diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts index d675c85ff539..cd5c3487b4f8 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.test.ts @@ -22,6 +22,7 @@ import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts import { CheckpointReactorLive } from "./CheckpointReactor.ts"; import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; import { OrchestrationProjectionPipelineLive } from "./ProjectionPipeline.ts"; +import { RuntimeReceiptBusLive } from "./RuntimeReceiptBus.ts"; import { OrchestrationEventStoreLive } from "../../persistence/Layers/OrchestrationEventStore.ts"; import { OrchestrationCommandReceiptRepositoryLive } from "../../persistence/Layers/OrchestrationCommandReceipts.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; @@ -112,7 +113,7 @@ async function waitForThread( checkpoints: ReadonlyArray<{ checkpointTurnCount: number }>; activities: ReadonlyArray<{ kind: string }>; }) => boolean, - timeoutMs = 2000, + timeoutMs = 5000, ) { const deadline = Date.now() + timeoutMs; const poll = async (): Promise<{ @@ -137,7 +138,7 @@ async function waitForThread( async function waitForEvent( engine: OrchestrationEngineShape, predicate: (event: { type: string }) => boolean, - timeoutMs = 2000, + timeoutMs = 5000, ) { const deadline = Date.now() + timeoutMs; const poll = async () => { @@ -188,7 +189,7 @@ function gitShowFileAtRef(cwd: string, ref: string, filePath: string): string { return runGit(cwd, ["show", `${ref}:${filePath}`]); } -async function waitForGitRefExists(cwd: string, ref: string, timeoutMs = 2000) { +async function waitForGitRefExists(cwd: string, ref: string, timeoutMs = 5000) { const deadline = Date.now() + timeoutMs; const poll = async (): Promise => { if (gitRefExists(cwd, ref)) { @@ -252,6 +253,7 @@ describe("CheckpointReactor", () => { const layer = CheckpointReactorLive.pipe( Layer.provideMerge(orchestrationLayer), + Layer.provideMerge(RuntimeReceiptBusLive), Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), Layer.provideMerge(CheckpointStoreLive), Layer.provideMerge(ServerConfig.layerTest(process.cwd(), process.cwd())), @@ -264,7 +266,7 @@ describe("CheckpointReactor", () => { const checkpointStore = await runtime.runPromise(Effect.service(CheckpointStore)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(reactor.start.pipe(Scope.provide(scope))); - await Effect.runPromise(Effect.sleep("10 millis")); + const drain = () => Effect.runPromise(reactor.drain); const createdAt = new Date().toISOString(); await Effect.runPromise( @@ -321,6 +323,7 @@ describe("CheckpointReactor", () => { engine, provider, cwd, + drain, }; } @@ -449,7 +452,7 @@ describe("CheckpointReactor", () => { payload: { state: "completed" }, }); - await Effect.runPromise(Effect.sleep("40 millis")); + await harness.drain(); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); const midThread = midReadModel.threads.find( (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), @@ -643,7 +646,7 @@ describe("CheckpointReactor", () => { status: "completed", }); - await Effect.runPromise(Effect.sleep("40 millis")); + await harness.drain(); const readModel = await Effect.runPromise(harness.engine.getReadModel()); const thread = readModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); expect(thread?.checkpoints.some((checkpoint) => checkpoint.checkpointTurnCount === 3)).toBe( diff --git a/apps/server/src/orchestration/Layers/CheckpointReactor.ts b/apps/server/src/orchestration/Layers/CheckpointReactor.ts index 52243248fb03..2215d3b1950a 100644 --- a/apps/server/src/orchestration/Layers/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Layers/CheckpointReactor.ts @@ -7,7 +7,8 @@ import { type OrchestrationEvent, type ProviderRuntimeEvent, } from "@t3tools/contracts"; -import { Cause, Effect, Layer, Option, Queue, Stream } from "effect"; +import { Cause, Effect, Layer, Option, Stream } from "effect"; +import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { parseTurnDiffFilesFromUnifiedDiff } from "../../checkpointing/Diffs.ts"; import { @@ -18,6 +19,7 @@ import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { CheckpointReactor, type CheckpointReactorShape } from "../Services/CheckpointReactor.ts"; import { OrchestrationEngineService } from "../Services/OrchestrationEngine.ts"; +import { RuntimeReceiptBus } from "../Services/RuntimeReceiptBus.ts"; import { CheckpointStoreError } from "../../checkpointing/Errors.ts"; import { OrchestrationDispatchError } from "../Errors.ts"; import { isGitRepository } from "../../git/isRepo.ts"; @@ -63,6 +65,7 @@ const make = Effect.gen(function* () { const orchestrationEngine = yield* OrchestrationEngineService; const providerService = yield* ProviderService; const checkpointStore = yield* CheckpointStore; + const receiptBus = yield* RuntimeReceiptBus; const appendRevertFailureActivity = (input: { readonly threadId: ThreadId; @@ -115,9 +118,7 @@ const make = Effect.gen(function* () { const resolveSessionRuntimeForThread = Effect.fnUntraced(function* ( threadId: ThreadId, - ): Effect.fn.Return< - Option.Option<{ readonly threadId: ThreadId; readonly cwd: string }> - > { + ): Effect.fn.Return> { const readModel = yield* orchestrationEngine.getReadModel(); const thread = readModel.threads.find((entry) => entry.id === threadId); @@ -133,9 +134,7 @@ const make = Effect.gen(function* () { }; if (thread) { - const projectedSession = sessions.find( - (session) => session.threadId === thread.id, - ); + const projectedSession = sessions.find((session) => session.threadId === thread.id); const fromProjected = findSessionWithCwd(projectedSession); if (Option.isSome(fromProjected)) { return fromProjected; @@ -276,6 +275,22 @@ const make = Effect.gen(function* () { checkpointTurnCount: nextTurnCount, createdAt: now, }); + yield* receiptBus.publish({ + type: "checkpoint.diff.finalized", + threadId: thread.id, + turnId, + checkpointTurnCount: nextTurnCount, + checkpointRef: targetCheckpointRef, + status: checkpointStatusFromRuntime(event.payload.state), + createdAt: now, + }); + yield* receiptBus.publish({ + type: "turn.processing.quiesced", + threadId: thread.id, + turnId, + checkpointTurnCount: nextTurnCount, + createdAt: now, + }); yield* orchestrationEngine.dispatch({ type: "thread.activity.append", @@ -306,9 +321,7 @@ const make = Effect.gen(function* () { } const readModel = yield* orchestrationEngine.getReadModel(); - const thread = readModel.threads.find( - (entry) => entry.id === event.threadId, - ); + const thread = readModel.threads.find((entry) => entry.id === event.threadId); if (!thread) { return; } @@ -351,6 +364,13 @@ const make = Effect.gen(function* () { cwd: checkpointCwd, checkpointRef: baselineCheckpointRef, }); + yield* receiptBus.publish({ + type: "checkpoint.baseline.captured", + threadId: thread.id, + checkpointTurnCount: currentTurnCount, + checkpointRef: baselineCheckpointRef, + createdAt: event.createdAt, + }); }); const ensurePreTurnBaselineFromDomainTurnStart = Effect.fnUntraced(function* ( @@ -413,6 +433,13 @@ const make = Effect.gen(function* () { cwd: checkpointCwd, checkpointRef: baselineCheckpointRef, }); + yield* receiptBus.publish({ + type: "checkpoint.baseline.captured", + threadId, + checkpointTurnCount: currentTurnCount, + checkpointRef: baselineCheckpointRef, + createdAt: event.occurredAt, + }); }); const handleRevertRequested = Effect.fnUntraced(function* ( @@ -600,14 +627,9 @@ const make = Effect.gen(function* () { }), ); - const start: CheckpointReactorShape["start"] = Effect.gen(function* () { - const queue = yield* Queue.unbounded(); - yield* Effect.addFinalizer(() => Queue.shutdown(queue).pipe(Effect.asVoid)); - - yield* Effect.forkScoped( - Effect.forever(Queue.take(queue).pipe(Effect.flatMap(processInputSafely))), - ); + const worker = yield* makeDrainableWorker(processInputSafely); + const start: CheckpointReactorShape["start"] = Effect.gen(function* () { yield* Effect.forkScoped( Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => { if ( @@ -617,7 +639,7 @@ const make = Effect.gen(function* () { ) { return Effect.void; } - return Queue.offer(queue, { source: "domain", event }).pipe(Effect.asVoid); + return worker.enqueue({ source: "domain", event }); }), ); @@ -626,13 +648,14 @@ const make = Effect.gen(function* () { if (event.type !== "turn.started" && event.type !== "turn.completed") { return Effect.void; } - return Queue.offer(queue, { source: "runtime", event }).pipe(Effect.asVoid); + return worker.enqueue({ source: "runtime", event }); }), ); }); return { start, + drain: worker.drain, } satisfies CheckpointReactorShape; }); diff --git a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts index 44f726142c87..1514bef595ff 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts @@ -27,6 +27,7 @@ describe("OrchestrationReactor", () => { start: Effect.sync(() => { started.push("provider-runtime-ingestion"); }), + drain: Effect.void, }), ), Layer.provideMerge( @@ -34,6 +35,7 @@ describe("OrchestrationReactor", () => { start: Effect.sync(() => { started.push("provider-command-reactor"); }), + drain: Effect.void, }), ), Layer.provideMerge( @@ -41,6 +43,7 @@ describe("OrchestrationReactor", () => { start: Effect.sync(() => { started.push("checkpoint-reactor"); }), + drain: Effect.void, }), ), ), diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts index 4f352435fe50..43439fe0a82a 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.test.ts @@ -213,7 +213,7 @@ describe("ProviderCommandReactor", () => { const reactor = await runtime.runPromise(Effect.service(ProviderCommandReactor)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(reactor.start.pipe(Scope.provide(scope))); - await Effect.runPromise(Effect.sleep("10 millis")); + const drain = () => Effect.runPromise(reactor.drain); await Effect.runPromise( engine.dispatch({ @@ -253,6 +253,7 @@ describe("ProviderCommandReactor", () => { renameBranch, generateBranchName, stateDir, + drain, }; } @@ -567,7 +568,7 @@ describe("ProviderCommandReactor", () => { return thread?.runtimeMode === "approval-required"; }); await waitFor(() => harness.startSession.mock.calls.length === 2); - await Effect.runPromise(Effect.sleep("30 millis")); + await harness.drain(); expect(harness.stopSession.mock.calls.length).toBe(0); expect(harness.sendTurn.mock.calls.length).toBe(1); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index d34791bc20ce..8f70cd9ad945 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -12,7 +12,8 @@ import { type RuntimeMode, type TurnId, } from "@t3tools/contracts"; -import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Schema, Stream } from "effect"; +import { Cache, Cause, Duration, Effect, Layer, Option, Schema, Stream } from "effect"; +import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts"; import { GitCore } from "../../git/Services/GitCore.ts"; @@ -222,9 +223,9 @@ const make = Effect.gen(function* () { }); const resolveActiveSession = (threadId: ThreadId) => - providerService.listSessions().pipe( - Effect.map((sessions) => sessions.find((session) => session.threadId === threadId)), - ); + providerService + .listSessions() + .pipe(Effect.map((sessions) => sessions.find((session) => session.threadId === threadId))); const startProviderSession = (input?: { readonly resumeCursor?: unknown; @@ -232,7 +233,7 @@ const make = Effect.gen(function* () { }) => providerService.startSession(threadId, { threadId, - ...(input?.provider ?? preferredProvider + ...((input?.provider ?? preferredProvider) ? { provider: input?.provider ?? preferredProvider } : {}), ...(effectiveCwd ? { cwd: effectiveCwd } : {}), @@ -263,16 +264,15 @@ const make = Effect.gen(function* () { thread.session && thread.session.status !== "stopped" ? thread.id : null; if (existingSessionThreadId) { const runtimeModeChanged = thread.runtimeMode !== thread.session?.runtimeMode; - const providerChanged = options?.provider !== undefined && options.provider !== currentProvider; + const providerChanged = + options?.provider !== undefined && options.provider !== currentProvider; const activeSession = yield* resolveActiveSession(existingSessionThreadId); const sessionModelSwitch = currentProvider === undefined ? "in-session" : (yield* providerService.getCapabilities(currentProvider)).sessionModelSwitch; - const modelChanged = - options?.model !== undefined && options.model !== activeSession?.model; - const shouldRestartForModelChange = - modelChanged && sessionModelSwitch === "restart-session"; + const modelChanged = options?.model !== undefined && options.model !== activeSession?.model; + const shouldRestartForModelChange = modelChanged && sessionModelSwitch === "restart-session"; if (!runtimeModeChanged && !providerChanged && !shouldRestartForModelChange) { return existingSessionThreadId; @@ -340,15 +340,16 @@ const make = Effect.gen(function* () { }); const normalizedInput = toNonEmptyProviderInput(input.messageText); const normalizedAttachments = input.attachments ?? []; - const activeSession = yield* providerService.listSessions().pipe( - Effect.map((sessions) => sessions.find((session) => session.threadId === input.threadId)), - ); + const activeSession = yield* providerService + .listSessions() + .pipe( + Effect.map((sessions) => sessions.find((session) => session.threadId === input.threadId)), + ); const sessionModelSwitch = activeSession === undefined ? "in-session" : (yield* providerService.getCapabilities(activeSession.provider)).sessionModelSwitch; - const modelForTurn = - sessionModelSwitch === "unsupported" ? activeSession?.model : input.model; + const modelForTurn = sessionModelSwitch === "unsupported" ? activeSession?.model : input.model; yield* providerService.sendTurn({ threadId: input.threadId, @@ -470,8 +471,12 @@ const make = Effect.gen(function* () { ...(message.attachments !== undefined ? { attachments: message.attachments } : {}), ...(event.payload.provider !== undefined ? { provider: event.payload.provider } : {}), ...(event.payload.model !== undefined ? { model: event.payload.model } : {}), - ...(event.payload.serviceTier !== undefined ? { serviceTier: event.payload.serviceTier } : {}), - ...(event.payload.modelOptions !== undefined ? { modelOptions: event.payload.modelOptions } : {}), + ...(event.payload.serviceTier !== undefined + ? { serviceTier: event.payload.serviceTier } + : {}), + ...(event.payload.modelOptions !== undefined + ? { modelOptions: event.payload.modelOptions } + : {}), interactionMode: event.payload.interactionMode, createdAt: event.payload.createdAt, }); @@ -661,34 +666,28 @@ const make = Effect.gen(function* () { }), ); - const start: ProviderCommandReactorShape["start"] = Effect.gen(function* () { - const queue = yield* Queue.unbounded(); - yield* Effect.addFinalizer(() => Queue.shutdown(queue).pipe(Effect.asVoid)); - - yield* Effect.forkScoped( - Effect.forever(Queue.take(queue).pipe(Effect.flatMap(processDomainEventSafely))), - ); - - yield* Effect.forkScoped( - Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => { - if ( - event.type !== "thread.runtime-mode-set" && - event.type !== "thread.turn-start-requested" && - event.type !== "thread.turn-interrupt-requested" && - event.type !== "thread.approval-response-requested" && - event.type !== "thread.user-input-response-requested" && - event.type !== "thread.session-stop-requested" - ) { - return Effect.void; - } + const worker = yield* makeDrainableWorker(processDomainEventSafely); + + const start: ProviderCommandReactorShape["start"] = Effect.forkScoped( + Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => { + if ( + event.type !== "thread.runtime-mode-set" && + event.type !== "thread.turn-start-requested" && + event.type !== "thread.turn-interrupt-requested" && + event.type !== "thread.approval-response-requested" && + event.type !== "thread.user-input-response-requested" && + event.type !== "thread.session-stop-requested" + ) { + return Effect.void; + } - return Queue.offer(queue, event).pipe(Effect.asVoid); - }), - ); - }); + return worker.enqueue(event); + }), + ).pipe(Effect.asVoid); return { start, + drain: worker.drain, } satisfies ProviderCommandReactorShape; }); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index 96242b846cbe..17fbde0acd5f 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -159,7 +159,7 @@ describe("ProviderRuntimeIngestion", () => { const ingestion = await runtime.runPromise(Effect.service(ProviderRuntimeIngestionService)); scope = await Effect.runPromise(Scope.make("sequential")); await Effect.runPromise(ingestion.start.pipe(Scope.provide(scope))); - await Effect.runPromise(Effect.sleep("10 millis")); + const drain = () => Effect.runPromise(ingestion.drain); const createdAt = new Date().toISOString(); await Effect.runPromise( @@ -209,6 +209,7 @@ describe("ProviderRuntimeIngestion", () => { return { engine, emit: provider.emit, + drain, }; } @@ -377,7 +378,7 @@ describe("ProviderRuntimeIngestion", () => { threadId: asThreadId("thread-1"), }); - await Effect.runPromise(Effect.sleep("40 millis")); + await harness.drain(); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); const midThread = midReadModel.threads.find((entry) => entry.id === ThreadId.makeUnsafe("thread-1")); expect(midThread?.session?.status).toBe("running"); @@ -428,7 +429,7 @@ describe("ProviderRuntimeIngestion", () => { status: "completed", }); - await Effect.runPromise(Effect.sleep("40 millis")); + await harness.drain(); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); const midThread = midReadModel.threads.find( (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), @@ -482,7 +483,7 @@ describe("ProviderRuntimeIngestion", () => { status: "completed", }); - await Effect.runPromise(Effect.sleep("40 millis")); + await harness.drain(); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); const midThread = midReadModel.threads.find( (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), @@ -720,7 +721,7 @@ describe("ProviderRuntimeIngestion", () => { }, }); - await Effect.runPromise(Effect.sleep("30 millis")); + await harness.drain(); const midReadModel = await Effect.runPromise(harness.engine.getReadModel()); const midThread = midReadModel.threads.find( (entry) => entry.id === ThreadId.makeUnsafe("thread-1"), @@ -779,7 +780,7 @@ describe("ProviderRuntimeIngestion", () => { createdAt: now, }), ); - await Effect.runPromise(Effect.sleep("30 millis")); + await harness.drain(); harness.emit({ type: "turn.started", diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 022a19667480..ae01977ea78a 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -10,7 +10,8 @@ import { type OrchestrationThreadActivity, type ProviderRuntimeEvent, } from "@t3tools/contracts"; -import { Cache, Cause, Duration, Effect, Layer, Option, Queue, Ref, Stream } from "effect"; +import { Cache, Cause, Duration, Effect, Layer, Option, Ref, Stream } from "effect"; +import { makeDrainableWorker } from "@t3tools/shared/DrainableWorker"; import { ProviderService } from "../../provider/Services/ProviderService.ts"; import { resolveThreadWorkspaceCwd } from "../../checkpointing/Utils.ts"; @@ -210,9 +211,9 @@ function runtimeEventToActivities( ? "Command approval requested" : requestKind === "file-read" ? "File-read approval requested" - : requestKind === "file-change" - ? "File-change approval requested" - : "Approval requested", + : requestKind === "file-change" + ? "File-change approval requested" + : "Approval requested", payload: { requestId: toApprovalRequestId(event.requestId), ...(requestKind ? { requestKind } : {}), @@ -298,7 +299,9 @@ function runtimeEventToActivities( summary: "Plan updated", payload: { plan: event.payload.plan, - ...(event.payload.explanation !== undefined ? { explanation: event.payload.explanation } : {}), + ...(event.payload.explanation !== undefined + ? { explanation: event.payload.explanation } + : {}), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -358,7 +361,9 @@ function runtimeEventToActivities( payload: { taskId: event.payload.taskId, ...(event.payload.taskType ? { taskType: event.payload.taskType } : {}), - ...(event.payload.description ? { detail: truncateDetail(event.payload.description) } : {}), + ...(event.payload.description + ? { detail: truncateDetail(event.payload.description) } + : {}), }, turnId: toTurnId(event.turnId) ?? null, ...maybeSequence, @@ -525,11 +530,7 @@ const make = Effect.gen(function* () { return isGitRepository(workspaceCwd); }); - const rememberAssistantMessageId = ( - threadId: ThreadId, - turnId: TurnId, - messageId: MessageId, - ) => + const rememberAssistantMessageId = (threadId: ThreadId, turnId: TurnId, messageId: MessageId) => Cache.getOption(turnMessageIdsByTurnKey, providerTurnKey(threadId, turnId)).pipe( Effect.flatMap((existingIds) => Cache.set( @@ -547,11 +548,7 @@ const make = Effect.gen(function* () { ), ); - const forgetAssistantMessageId = ( - threadId: ThreadId, - turnId: TurnId, - messageId: MessageId, - ) => + const forgetAssistantMessageId = (threadId: ThreadId, turnId: TurnId, messageId: MessageId) => Cache.getOption(turnMessageIdsByTurnKey, providerTurnKey(threadId, turnId)).pipe( Effect.flatMap((existingIds) => Option.match(existingIds, { @@ -616,7 +613,8 @@ const make = Effect.gen(function* () { const existing = Option.getOrUndefined(existingEntry); return Cache.set(bufferedProposedPlanById, planId, { text: `${existing?.text ?? ""}${delta}`, - createdAt: existing?.createdAt && existing.createdAt.length > 0 ? existing.createdAt : createdAt, + createdAt: + existing?.createdAt && existing.createdAt.length > 0 ? existing.createdAt : createdAt, }); }), ); @@ -633,7 +631,8 @@ const make = Effect.gen(function* () { const clearBufferedProposedPlan = (planId: string) => Cache.invalidate(bufferedProposedPlanById, planId); - const clearAssistantMessageState = (messageId: MessageId) => clearBufferedAssistantText(messageId); + const clearAssistantMessageState = (messageId: MessageId) => + clearBufferedAssistantText(messageId); const finalizeAssistantMessage = (input: { event: ProviderRuntimeEvent; @@ -862,8 +861,8 @@ const make = Effect.gen(function* () { : event.type === "turn.completed" && runtimeTurnState(event) === "failed" ? (runtimeTurnErrorMessage(event) ?? thread.session?.lastError ?? "Turn failed") : status === "ready" - ? null - : (thread.session?.lastError ?? null); + ? null + : (thread.session?.lastError ?? null); if (shouldApplyThreadLifecycle) { yield* orchestrationEngine.dispatch({ @@ -935,7 +934,9 @@ const make = Effect.gen(function* () { const assistantCompletion = event.type === "item.completed" && event.payload.itemType === "assistant_message" ? { - messageId: MessageId.makeUnsafe(`assistant:${event.itemId ?? event.turnId ?? event.eventId}`), + messageId: MessageId.makeUnsafe( + `assistant:${event.itemId ?? event.turnId ?? event.eventId}`, + ), fallbackText: event.payload.detail, } : undefined; @@ -1025,9 +1026,7 @@ const make = Effect.gen(function* () { const shouldApplyRuntimeError = !STRICT_PROVIDER_LIFECYCLE_GUARD ? true - : activeTurnId === null || - eventTurnId === undefined || - sameId(activeTurnId, eventTurnId); + : activeTurnId === null || eventTurnId === undefined || sameId(activeTurnId, eventTurnId); if (shouldApplyRuntimeError) { yield* orchestrationEngine.dispatch({ @@ -1115,16 +1114,12 @@ const make = Effect.gen(function* () { }), ); - const start: ProviderRuntimeIngestionShape["start"] = Effect.gen(function* () { - const inputQueue = yield* Queue.unbounded(); - yield* Effect.addFinalizer(() => Queue.shutdown(inputQueue).pipe(Effect.asVoid)); + const worker = yield* makeDrainableWorker(processInputSafely); - yield* Effect.forkScoped( - Effect.forever(Queue.take(inputQueue).pipe(Effect.flatMap(processInputSafely))), - ); + const start: ProviderRuntimeIngestionShape["start"] = Effect.gen(function* () { yield* Effect.forkScoped( Stream.runForEach(providerService.streamEvents, (event) => - Queue.offer(inputQueue, { source: "runtime", event }).pipe(Effect.asVoid), + worker.enqueue({ source: "runtime", event }), ), ); yield* Effect.forkScoped( @@ -1132,13 +1127,14 @@ const make = Effect.gen(function* () { if (event.type !== "thread.turn-start-requested") { return Effect.void; } - return Queue.offer(inputQueue, { source: "domain", event }).pipe(Effect.asVoid); + return worker.enqueue({ source: "domain", event }); }), ); }); return { start, + drain: worker.drain, } satisfies ProviderRuntimeIngestionShape; }); diff --git a/apps/server/src/orchestration/Layers/RuntimeReceiptBus.ts b/apps/server/src/orchestration/Layers/RuntimeReceiptBus.ts new file mode 100644 index 000000000000..56c526c08b5f --- /dev/null +++ b/apps/server/src/orchestration/Layers/RuntimeReceiptBus.ts @@ -0,0 +1,18 @@ +import { Effect, Layer, PubSub, Stream } from "effect"; + +import { + RuntimeReceiptBus, + type RuntimeReceiptBusShape, + type OrchestrationRuntimeReceipt, +} from "../Services/RuntimeReceiptBus.ts"; + +const makeRuntimeReceiptBus = Effect.gen(function* () { + const pubSub = yield* PubSub.unbounded(); + + return { + publish: (receipt) => PubSub.publish(pubSub, receipt).pipe(Effect.asVoid), + stream: Stream.fromPubSub(pubSub), + } satisfies RuntimeReceiptBusShape; +}); + +export const RuntimeReceiptBusLive = Layer.effect(RuntimeReceiptBus, makeRuntimeReceiptBus); diff --git a/apps/server/src/orchestration/Services/CheckpointReactor.ts b/apps/server/src/orchestration/Services/CheckpointReactor.ts index 7989341b9de1..612bc22acb66 100644 --- a/apps/server/src/orchestration/Services/CheckpointReactor.ts +++ b/apps/server/src/orchestration/Services/CheckpointReactor.ts @@ -23,6 +23,12 @@ export interface CheckpointReactorShape { * internal queue. */ readonly start: Effect.Effect; + + /** + * Resolves when the internal processing queue is empty and idle. + * Intended for test use to replace timing-sensitive sleeps. + */ + readonly drain: Effect.Effect; } /** diff --git a/apps/server/src/orchestration/Services/ProviderCommandReactor.ts b/apps/server/src/orchestration/Services/ProviderCommandReactor.ts index c394435ba9fb..8f20571c1570 100644 --- a/apps/server/src/orchestration/Services/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Services/ProviderCommandReactor.ts @@ -23,6 +23,12 @@ export interface ProviderCommandReactorShape { * processing. */ readonly start: Effect.Effect; + + /** + * Resolves when the internal processing queue is empty and idle. + * Intended for test use to replace timing-sensitive sleeps. + */ + readonly drain: Effect.Effect; } /** diff --git a/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts index edd1c207c11a..2bbb157f10c6 100644 --- a/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Services/ProviderRuntimeIngestion.ts @@ -23,6 +23,12 @@ export interface ProviderRuntimeIngestionShape { * logging warnings. */ readonly start: Effect.Effect; + + /** + * Resolves when the internal processing queue is empty and idle. + * Intended for test use to replace timing-sensitive sleeps. + */ + readonly drain: Effect.Effect; } /** diff --git a/apps/server/src/orchestration/Services/RuntimeReceiptBus.ts b/apps/server/src/orchestration/Services/RuntimeReceiptBus.ts new file mode 100644 index 000000000000..a08ff916856d --- /dev/null +++ b/apps/server/src/orchestration/Services/RuntimeReceiptBus.ts @@ -0,0 +1,49 @@ +import { CheckpointRef, IsoDateTime, NonNegativeInt, ThreadId, TurnId } from "@t3tools/contracts"; +import { Schema, ServiceMap } from "effect"; +import type { Effect, Stream } from "effect"; + +export const CheckpointBaselineCapturedReceipt = Schema.Struct({ + type: Schema.Literal("checkpoint.baseline.captured"), + threadId: ThreadId, + checkpointTurnCount: NonNegativeInt, + checkpointRef: CheckpointRef, + createdAt: IsoDateTime, +}); +export type CheckpointBaselineCapturedReceipt = typeof CheckpointBaselineCapturedReceipt.Type; + +export const CheckpointDiffFinalizedReceipt = Schema.Struct({ + type: Schema.Literal("checkpoint.diff.finalized"), + threadId: ThreadId, + turnId: TurnId, + checkpointTurnCount: NonNegativeInt, + checkpointRef: CheckpointRef, + status: Schema.Literals(["ready", "missing", "error"]), + createdAt: IsoDateTime, +}); +export type CheckpointDiffFinalizedReceipt = typeof CheckpointDiffFinalizedReceipt.Type; + +export const TurnProcessingQuiescedReceipt = Schema.Struct({ + type: Schema.Literal("turn.processing.quiesced"), + threadId: ThreadId, + turnId: TurnId, + checkpointTurnCount: NonNegativeInt, + createdAt: IsoDateTime, +}); +export type TurnProcessingQuiescedReceipt = typeof TurnProcessingQuiescedReceipt.Type; + +export const OrchestrationRuntimeReceipt = Schema.Union([ + CheckpointBaselineCapturedReceipt, + CheckpointDiffFinalizedReceipt, + TurnProcessingQuiescedReceipt, +]); +export type OrchestrationRuntimeReceipt = typeof OrchestrationRuntimeReceipt.Type; + +export interface RuntimeReceiptBusShape { + readonly publish: (receipt: OrchestrationRuntimeReceipt) => Effect.Effect; + readonly stream: Stream.Stream; +} + +export class RuntimeReceiptBus extends ServiceMap.Service< + RuntimeReceiptBus, + RuntimeReceiptBusShape +>()("t3/orchestration/Services/RuntimeReceiptBus") {} diff --git a/apps/server/src/serverLayers.ts b/apps/server/src/serverLayers.ts index b0630a55b954..ff9b10d96f22 100644 --- a/apps/server/src/serverLayers.ts +++ b/apps/server/src/serverLayers.ts @@ -17,6 +17,7 @@ import { ProviderCommandReactorLive } from "./orchestration/Layers/ProviderComma import { OrchestrationProjectionPipelineLive } from "./orchestration/Layers/ProjectionPipeline"; import { OrchestrationProjectionSnapshotQueryLive } from "./orchestration/Layers/ProjectionSnapshotQuery"; import { ProviderRuntimeIngestionLive } from "./orchestration/Layers/ProviderRuntimeIngestion"; +import { RuntimeReceiptBusLive } from "./orchestration/Layers/RuntimeReceiptBus"; import { ProviderUnsupportedError } from "./provider/Errors"; import { makeCodexAdapterLive } from "./provider/Layers/CodexAdapter"; import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry"; @@ -87,6 +88,7 @@ export function makeServerRuntimeServicesLayer() { OrchestrationProjectionSnapshotQueryLive, CheckpointStoreLive, checkpointDiffQueryLayer, + RuntimeReceiptBusLive, ); const runtimeIngestionLayer = ProviderRuntimeIngestionLive.pipe( Layer.provideMerge(runtimeServicesLayer), diff --git a/apps/server/src/watchFileWithStatPolling.ts b/apps/server/src/watchFileWithStatPolling.ts new file mode 100644 index 000000000000..36f390f4fa59 --- /dev/null +++ b/apps/server/src/watchFileWithStatPolling.ts @@ -0,0 +1,34 @@ +import nodeFs from "node:fs"; +import { Effect, Queue, Stream } from "effect"; + +export interface WatchFileWithStatPollingOptions { + readonly filePath: string; + readonly pollIntervalMs?: number; +} + +/** + * Expose Node's stat-based file poller as a scoped Effect stream. + * + * `fs.watchFile` uses libuv timers, so it remains a reliable fallback when + * the Effect scheduler is under load and prompt fiber wake-ups are not + * guaranteed. + */ +export const watchFileWithStatPolling = ({ + filePath, + pollIntervalMs = 100, +}: WatchFileWithStatPollingOptions) => + Stream.callback((queue) => + Effect.acquireRelease( + Effect.sync(() => { + const onChange = () => { + Queue.offerUnsafe(queue, undefined); + }; + nodeFs.watchFile(filePath, { interval: pollIntervalMs }, onChange); + return onChange; + }), + (onChange) => + Effect.sync(() => { + nodeFs.unwatchFile(filePath, onChange); + }), + ), + ); diff --git a/apps/server/src/wsServer.test.ts b/apps/server/src/wsServer.test.ts index 285028cca6e8..160eedd74a13 100644 --- a/apps/server/src/wsServer.test.ts +++ b/apps/server/src/wsServer.test.ts @@ -27,6 +27,8 @@ import { type ServerProviderStatus, type KeybindingsConfig, type ResolvedKeybindingsConfig, + type WsPushChannel, + type WsPushMessage, type WsPush, } from "@t3tools/contracts"; import { compileResolvedKeybindingRule, DEFAULT_KEYBINDINGS } from "./keybindings"; @@ -52,13 +54,6 @@ import { GitCommandError, GitManagerError } from "./git/Errors.ts"; import { MigrationError } from "@effect/sql-sqlite-bun/SqliteMigrator"; import { AnalyticsService } from "./telemetry/Services/AnalyticsService.ts"; -interface PendingMessages { - queue: unknown[]; - waiters: Array<(message: unknown) => void>; -} - -const pendingBySocket = new WeakMap(); - const asEventId = (value: string): EventId => EventId.makeUnsafe(value); const asProviderItemId = (value: string): ProviderItemId => ProviderItemId.makeUnsafe(value); const asThreadId = (value: string): ThreadId => ThreadId.makeUnsafe(value); @@ -214,21 +209,97 @@ class MockTerminalManager implements TerminalManagerShape { readonly dispose: TerminalManagerShape["dispose"] = Effect.void; } -function connectWs(port: number, token?: string): Promise { +// --------------------------------------------------------------------------- +// WebSocket test harness +// +// Incoming messages are split into two channels: +// - pushChannel: server push envelopes (type === "push") +// - responseChannel: request/response envelopes (have an "id" field) +// +// This means sendRequest never has to skip push messages and waitForPush +// never has to skip response messages, eliminating a class of ordering bugs. +// --------------------------------------------------------------------------- + +interface MessageChannel { + queue: T[]; + waiters: Array<{ + resolve: (value: T) => void; + reject: (error: Error) => void; + timeoutId: ReturnType | null; + }>; +} + +interface SocketChannels { + push: MessageChannel; + response: MessageChannel; +} + +const channelsBySocket = new WeakMap(); + +function enqueue(channel: MessageChannel, item: T) { + const waiter = channel.waiters.shift(); + if (waiter) { + if (waiter.timeoutId !== null) clearTimeout(waiter.timeoutId); + waiter.resolve(item); + return; + } + channel.queue.push(item); +} + +function dequeue(channel: MessageChannel, timeoutMs: number): Promise { + const queued = channel.queue.shift(); + if (queued !== undefined) { + return Promise.resolve(queued); + } + + return new Promise((resolve, reject) => { + const waiter = { + resolve, + reject, + timeoutId: setTimeout(() => { + const index = channel.waiters.indexOf(waiter); + if (index >= 0) channel.waiters.splice(index, 1); + reject(new Error(`Timed out waiting for WebSocket message after ${timeoutMs}ms`)); + }, timeoutMs) as ReturnType, + }; + channel.waiters.push(waiter); + }); +} + +function isWsPushEnvelope(message: unknown): message is WsPush { + if (typeof message !== "object" || message === null) return false; + if (!("type" in message) || !("channel" in message)) return false; + return (message as { type?: unknown }).type === "push"; +} + +function asWebSocketResponse(message: unknown): WebSocketResponse | null { + if (typeof message !== "object" || message === null) return null; + if (!("id" in message)) return null; + const id = (message as { id?: unknown }).id; + if (typeof id !== "string") return null; + return message as WebSocketResponse; +} + +function connectWsOnce(port: number, token?: string): Promise { return new Promise((resolve, reject) => { const query = token ? `?token=${encodeURIComponent(token)}` : ""; const ws = new WebSocket(`ws://127.0.0.1:${port}/${query}`); - const pending: PendingMessages = { queue: [], waiters: [] }; - pendingBySocket.set(ws, pending); + const channels: SocketChannels = { + push: { queue: [], waiters: [] }, + response: { queue: [], waiters: [] }, + }; + channelsBySocket.set(ws, channels); ws.on("message", (raw) => { const parsed = JSON.parse(String(raw)); - const waiter = pending.waiters.shift(); - if (waiter) { - waiter(parsed); - return; + if (isWsPushEnvelope(parsed)) { + enqueue(channels.push, parsed); + } else { + const response = asWebSocketResponse(parsed); + if (response) { + enqueue(channels.response, response); + } } - pending.queue.push(parsed); }); ws.once("open", () => resolve(ws)); @@ -236,28 +307,31 @@ function connectWs(port: number, token?: string): Promise { }); } -function waitForMessage(ws: WebSocket): Promise { - const pending = pendingBySocket.get(ws); - if (!pending) { - return Promise.reject(new Error("WebSocket not initialized")); - } +async function connectWs(port: number, token?: string, attempts = 5): Promise { + let lastError: unknown = new Error("WebSocket connection failed"); - const queued = pending.queue.shift(); - if (queued !== undefined) { - return Promise.resolve(queued); + for (let attempt = 0; attempt < attempts; attempt += 1) { + try { + return await connectWsOnce(port, token); + } catch (error) { + lastError = error; + if (attempt < attempts - 1) { + await new Promise((resolve) => setTimeout(resolve, 25)); + } + } } - return new Promise((resolve) => { - pending.waiters.push(resolve); - }); + throw lastError; } -function asWebSocketResponse(message: unknown): WebSocketResponse | null { - if (typeof message !== "object" || message === null) return null; - if (!("id" in message)) return null; - const id = (message as { id?: unknown }).id; - if (typeof id !== "string") return null; - return message as WebSocketResponse; +/** Connect and wait for the server.welcome push. Returns [ws, welcomeData]. */ +async function connectAndAwaitWelcome( + port: number, + token?: string, +): Promise<[WebSocket, WsPushMessage]> { + const ws = await connectWs(port, token); + const welcome = await waitForPush(ws, WS_CHANNELS.serverWelcome); + return [ws, welcome]; } async function sendRequest( @@ -265,6 +339,9 @@ async function sendRequest( method: string, params?: unknown, ): Promise { + const channels = channelsBySocket.get(ws); + if (!channels) throw new Error("WebSocket not initialized"); + const id = crypto.randomUUID(); const body = method === ORCHESTRATION_WS_METHODS.dispatchCommand @@ -272,44 +349,55 @@ async function sendRequest( : params && typeof params === "object" && !Array.isArray(params) ? { _tag: method, ...(params as Record) } : { _tag: method }; - const message = JSON.stringify({ id, body }); - ws.send(message); + ws.send(JSON.stringify({ id, body })); - // Wait for response with matching id + // Response channel only contains responses — no push filtering needed while (true) { - const parsed = asWebSocketResponse(await waitForMessage(ws)); - if (!parsed) { - continue; - } - if (parsed.id === id) { - return parsed; - } - if (parsed.id === "unknown") { - return parsed; + const response = await dequeue(channels.response, 60_000); + if (response.id === id || response.id === "unknown") { + return response; } } } -async function waitForPush( +async function waitForPush( ws: WebSocket, - channel: string, - predicate?: (push: WsPush) => boolean, + channel: C, + predicate?: (push: WsPushMessage) => boolean, maxMessages = 120, -): Promise { - const take = async (remaining: number): Promise => { - if (remaining <= 0) { - throw new Error(`Timed out waiting for push on ${channel}`); - } - const message = (await waitForMessage(ws)) as WsPush; - if (message.type !== "push" || message.channel !== channel) { - return take(remaining - 1); - } - if (!predicate || predicate(message)) { - return message; + idleTimeoutMs = 5_000, +): Promise> { + const channels = channelsBySocket.get(ws); + if (!channels) throw new Error("WebSocket not initialized"); + + for (let remaining = maxMessages; remaining > 0; remaining--) { + const push = await dequeue(channels.push, idleTimeoutMs); + if (push.channel !== channel) continue; + const typed = push as WsPushMessage; + if (!predicate || predicate(typed)) return typed; + } + throw new Error(`Timed out waiting for push on ${channel}`); +} + +async function rewriteKeybindingsAndWaitForPush( + ws: WebSocket, + keybindingsPath: string, + contents: string, + predicate: ( + push: WsPushMessage, + ) => boolean, + attempts = 3, +): Promise> { + let lastError: unknown; + for (let attempt = 0; attempt < attempts; attempt++) { + fs.writeFileSync(keybindingsPath, contents, "utf8"); + try { + return await waitForPush(ws, WS_CHANNELS.serverConfigUpdated, predicate, 20, 3_000); + } catch (error) { + lastError = error; } - return take(remaining - 1); - }; - return take(maxMessages); + } + throw lastError; } async function requestPath( @@ -489,23 +577,57 @@ describe("WebSocket Server", () => { it("sends welcome message on connect", async () => { server = await createTestServer({ cwd: "/test/project" }); - // Get the actual port after listen const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; expect(port).toBeGreaterThan(0); - const ws = await connectWs(port); + const [ws, welcome] = await connectAndAwaitWelcome(port); connections.push(ws); - const message = (await waitForMessage(ws)) as WsPush; - expect(message.type).toBe("push"); - expect(message.channel).toBe(WS_CHANNELS.serverWelcome); - expect(message.data).toEqual({ + expect(welcome.type).toBe("push"); + expect(welcome.data).toEqual({ cwd: "/test/project", projectName: "project", }); }); + it("delivers request-triggered pushes for requests sent immediately after open", async () => { + server = await createTestServer({ cwd: "/test/project" }); + const addr = server.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + expect(port).toBeGreaterThan(0); + + const ws = await connectWs(port); + connections.push(ws); + + const responsePromise = sendRequest(ws, WS_METHODS.terminalOpen, { + threadId: asThreadId("thread-1"), + cwd: "/test/project", + }); + + const welcome = await waitForPush(ws, WS_CHANNELS.serverWelcome); + const terminalEvent = await waitForPush(ws, WS_CHANNELS.terminalEvent, (push) => push.data.type === "started"); + const response = await responsePromise; + + expect(welcome.channel).toBe(WS_CHANNELS.serverWelcome); + expect(terminalEvent.channel).toBe(WS_CHANNELS.terminalEvent); + expect(terminalEvent.sequence).toBeGreaterThan(welcome.sequence); + expect(response.id).toBeDefined(); + expect(response.result).toEqual(expect.objectContaining({ threadId: "thread-1" })); + }); + + + it("continues startup when keybindings runtime bootstrap fails", async () => { + server = await createTestServer({ cwd: "/test/project", stateDir: "/dev/null" }); + const addr = server.address(); + const port = typeof addr === "object" && addr !== null ? addr.port : 0; + expect(port).toBeGreaterThan(0); + + const [ws, welcome] = await connectAndAwaitWelcome(port); + connections.push(ws); + expect(welcome.channel).toBe(WS_CHANNELS.serverWelcome); + }); + it("serves persisted attachments from stateDir", async () => { const stateDir = makeTempDir("t3code-state-attachments-"); const attachmentPath = path.join(stateDir, "attachments", "thread-a", "message-a", "0.png"); @@ -589,10 +711,8 @@ describe("WebSocket Server", () => { const port = typeof addr === "object" && addr !== null ? addr.port : 0; expect(port).toBeGreaterThan(0); - const ws = await connectWs(port); + const [ws, welcome] = await connectAndAwaitWelcome(port); connections.push(ws); - const welcome = (await waitForMessage(ws)) as WsPush; // welcome - expect(welcome.channel).toBe(WS_CHANNELS.serverWelcome); expect(welcome.data).toEqual( expect.objectContaining({ cwd: "/test/bootstrap-workspace", @@ -666,9 +786,8 @@ describe("WebSocket Server", () => { let port = typeof addr === "object" && addr !== null ? addr.port : 0; expect(port).toBeGreaterThan(0); - const firstWs = await connectWs(port); + const [firstWs, firstWelcome] = await connectAndAwaitWelcome(port); connections.push(firstWs); - const firstWelcome = (await waitForMessage(firstWs)) as WsPush; const firstBootstrapProjectId = (firstWelcome.data as { bootstrapProjectId?: string }) .bootstrapProjectId; const firstBootstrapThreadId = (firstWelcome.data as { bootstrapThreadId?: string }) @@ -690,10 +809,8 @@ describe("WebSocket Server", () => { port = typeof addr === "object" && addr !== null ? addr.port : 0; expect(port).toBeGreaterThan(0); - const secondWs = await connectWs(port); + const [secondWs, secondWelcome] = await connectAndAwaitWelcome(port); connections.push(secondWs); - const secondWelcome = (await waitForMessage(secondWs)) as WsPush; - expect(secondWelcome.channel).toBe(WS_CHANNELS.serverWelcome); expect(secondWelcome.data).toEqual( expect.objectContaining({ cwd, @@ -717,9 +834,8 @@ describe("WebSocket Server", () => { const port = typeof addr === "object" && addr !== null ? addr.port : 0; expect(port).toBeGreaterThan(0); - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); expect( logSpy.mock.calls.some(([message]) => { @@ -742,12 +858,9 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - // Consume welcome message - await waitForMessage(ws); - const response = await sendRequest(ws, WS_METHODS.serverGetConfig); expect(response.error).toBeUndefined(); expect(response.result).toEqual({ @@ -770,9 +883,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.serverGetConfig); expect(response.error).toBeUndefined(); @@ -801,9 +913,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.serverGetConfig); expect(response.error).toBeUndefined(); @@ -841,9 +952,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.serverGetConfig); expect(response.error).toBeUndefined(); @@ -885,32 +995,28 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); - fs.writeFileSync(keybindingsPath, "{ not-json", "utf8"); - const malformedPush = await waitForPush( + const malformedPush = await rewriteKeybindingsAndWaitForPush( ws, - WS_CHANNELS.serverConfigUpdated, + keybindingsPath, + "{ not-json", (push) => - Array.isArray((push.data as { issues?: unknown[] }).issues) && - Boolean((push.data as { issues: Array<{ kind: string }> }).issues[0]) && - (push.data as { issues: Array<{ kind: string }> }).issues[0]!.kind === - "keybindings.malformed-config", + Array.isArray(push.data.issues) && + Boolean(push.data.issues[0]) && + push.data.issues[0]!.kind === "keybindings.malformed-config", ); expect(malformedPush.data).toEqual({ issues: [{ kind: "keybindings.malformed-config", message: expect.any(String) }], providers: defaultProviderStatuses, }); - fs.writeFileSync(keybindingsPath, "[]", "utf8"); - const successPush = await waitForPush( + const successPush = await rewriteKeybindingsAndWaitForPush( ws, - WS_CHANNELS.serverConfigUpdated, - (push) => - Array.isArray((push.data as { issues?: unknown[] }).issues) && - (push.data as { issues: unknown[] }).issues.length === 0, + keybindingsPath, + "[]", + (push) => Array.isArray(push.data.issues) && push.data.issues.length === 0, ); expect(successPush.data).toEqual({ issues: [], providers: defaultProviderStatuses }); }); @@ -929,9 +1035,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.shellOpenInEditor, { cwd: "/my/workspace", @@ -957,11 +1062,9 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); - const response = await sendRequest(ws, WS_METHODS.serverGetConfig); expect(response.error).toBeUndefined(); const persistedConfig = JSON.parse( @@ -991,9 +1094,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const upsertResponse = await sendRequest(ws, WS_METHODS.serverUpsertKeybinding, { key: "mod+shift+r", @@ -1033,12 +1135,9 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - // Consume welcome push - await waitForMessage(ws); - const response = await sendRequest(ws, "nonexistent.method"); expect(response.error).toBeDefined(); expect(response.error!.message).toContain("Invalid request format"); @@ -1049,9 +1148,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, ORCHESTRATION_WS_METHODS.getTurnDiff, { threadId: "thread-missing", @@ -1067,9 +1165,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, ORCHESTRATION_WS_METHODS.getTurnDiff, { threadId: "thread-any", @@ -1087,9 +1184,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, ORCHESTRATION_WS_METHODS.getFullThreadDiff, { threadId: "thread-missing", @@ -1104,9 +1200,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const workspaceRoot = makeTempDir("t3code-ws-diff-project-"); const createdAt = new Date().toISOString(); @@ -1183,9 +1278,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const workspaceRoot = makeTempDir("t3code-ws-project-"); const createdAt = new Date().toISOString(); @@ -1278,9 +1372,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const open = await sendRequest(ws, WS_METHODS.terminalOpen, { threadId: "thread-1", @@ -1333,10 +1426,13 @@ describe("WebSocket Server", () => { }; terminalManager.emitEvent(manualEvent); - const push = (await waitForMessage(ws)) as WsPush; + const push = await waitForPush( + ws, + WS_CHANNELS.terminalEvent, + (candidate) => (candidate.data as TerminalEvent).type === "output", + ); expect(push.type).toBe("push"); expect(push.channel).toBe(WS_CHANNELS.terminalEvent); - expect((push.data as TerminalEvent).type).toBe("output"); }); it("detaches terminal event listener on stop for injected manager", async () => { @@ -1359,9 +1455,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.terminalOpen, { threadId: "", @@ -1377,21 +1472,17 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - // Consume welcome - await waitForMessage(ws); - // Send garbage ws.send("not json at all"); + // Error response goes to the response channel + const channels = channelsBySocket.get(ws)!; let response: WebSocketResponse | null = null; for (let attempt = 0; attempt < 5; attempt += 1) { - const message = asWebSocketResponse(await waitForMessage(ws)); - if (!message) { - continue; - } + const message = await dequeue(channels.response, 5_000); if (message.id === "unknown") { response = message; break; @@ -1424,9 +1515,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); ws.send( JSON.stringify({ @@ -1470,9 +1560,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const listResponse = await sendRequest(ws, WS_METHODS.projectsList); expect(listResponse.result).toBeUndefined(); @@ -1507,9 +1596,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.projectsSearchEntries, { cwd: workspace, @@ -1533,9 +1621,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.projectsWriteFile, { cwd: workspace, @@ -1559,9 +1646,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.projectsWriteFile, { cwd: workspace, @@ -1604,9 +1690,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const listResponse = await sendRequest(ws, WS_METHODS.gitListBranches, { cwd: "/repo/path" }); expect(listResponse.error).toBeUndefined(); @@ -1646,9 +1731,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.gitStatus, { cwd: "/test", @@ -1676,9 +1760,8 @@ describe("WebSocket Server", () => { const addr = server.address(); const port = typeof addr === "object" && addr !== null ? addr.port : 0; - const ws = await connectWs(port); + const [ws] = await connectAndAwaitWelcome(port); connections.push(ws); - await waitForMessage(ws); const response = await sendRequest(ws, WS_METHODS.gitRunStackedAction, { cwd: "/test", @@ -1699,9 +1782,7 @@ describe("WebSocket Server", () => { await expect(connectWs(port)).rejects.toThrow("WebSocket connection failed"); - const authorizedWs = await connectWs(port, "secret-token"); + const [authorizedWs] = await connectAndAwaitWelcome(port, "secret-token"); connections.push(authorizedWs); - const welcome = (await waitForMessage(authorizedWs)) as WsPush; - expect(welcome.channel).toBe(WS_CHANNELS.serverWelcome); }); }); diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index d8859c2fa5b7..51367574c1f7 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -20,7 +20,6 @@ import { PROVIDER_SEND_TURN_MAX_IMAGE_BYTES, ProjectId, ThreadId, - TerminalEvent, WS_CHANNELS, WS_METHODS, WebSocketRequest, @@ -35,7 +34,7 @@ import { FileSystem, Layer, Path, - Ref, + PubSub, Schema, Scope, ServiceMap, @@ -73,6 +72,8 @@ import { import { parseBase64DataUrl } from "./imageMime.ts"; import { AnalyticsService } from "./telemetry/Services/AnalyticsService.ts"; import { expandHomePath } from "./os-jank.ts"; +import { makeServerPushBus } from "./wsServer/pushBus.ts"; +import { makeServerReadiness } from "./wsServer/readiness.ts"; /** * ServerShape - Service API for server lifecycle control. @@ -197,8 +198,7 @@ function stripRequestTag(body: T) { function messageFromCause(cause: Cause.Cause): string { const squashed = Cause.squash(cause); - const message = - squashed instanceof Error ? squashed.message.trim() : String(squashed).trim(); + const message = squashed instanceof Error ? squashed.message.trim() : String(squashed).trim(); return message.length > 0 ? message : Cause.pretty(cause); } @@ -258,50 +258,33 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< const fileSystem = yield* FileSystem.FileSystem; const path = yield* Path.Path; - yield* keybindingsManager.syncDefaultKeybindingsOnStartup.pipe( - Effect.catch((error) => - Effect.logWarning("failed to sync keybindings defaults on startup", { - path: error.configPath, - detail: error.detail, - cause: error.cause, - }), - ), - ); - const providerStatuses = yield* providerHealth.getStatuses; - const clients = yield* Ref.make(new Set()); const logger = createLogger("ws"); + const readiness = yield* makeServerReadiness; function logOutgoingPush(push: WsPush, recipients: number) { if (!logWebSocketEvents) return; logger.event("outgoing push", { channel: push.channel, + sequence: push.sequence, recipients, payload: push.data, }); } - const encodePush = Schema.encodeEffect(Schema.fromJsonString(WsPush)); - const broadcastPush = Effect.fnUntraced(function* (push: WsPush) { - const message = yield* encodePush(push); - let recipients = 0; - for (const client of yield* Ref.get(clients)) { - if (client.readyState === client.OPEN) { - client.send(message); - recipients += 1; - } - } - logOutgoingPush(push, recipients); - }); - - const onTerminalEvent = Effect.fnUntraced(function* (event: TerminalEvent) { - yield* broadcastPush({ - type: "push", - channel: WS_CHANNELS.terminalEvent, - data: event, - }); + const pushBus = yield* makeServerPushBus({ + logOutgoingPush, }); + yield* readiness.markPushBusReady; + yield* keybindingsManager.start.pipe( + Effect.catchAllCause((cause) => + Effect.logWarning("keybindings runtime failed to start; continuing server bootstrap", { + error: Cause.pretty(cause), + }), + ), + ); + yield* readiness.markKeybindingsReady; const normalizeDispatchCommand = Effect.fnUntraced(function* (input: { readonly command: ClientOrchestrationCommand; @@ -331,10 +314,7 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< } satisfies OrchestrationCommand; } - if ( - input.command.type === "project.meta.update" && - input.command.workspaceRoot !== undefined - ) { + if (input.command.type === "project.meta.update" && input.command.workspaceRoot !== undefined) { return { ...input.command, workspaceRoot: yield* normalizeProjectWorkspaceRoot(input.command.workspaceRoot), @@ -601,10 +581,11 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< }); }); - const closeAllClients = Ref.get(clients).pipe( - Effect.flatMap(Effect.forEach((client) => Effect.sync(() => client.close()))), - Effect.flatMap(() => Ref.set(clients, new Set())), - ); + const closeAllClients = Effect.sync(() => { + wss.clients.forEach((client) => { + client.close(); + }); + }); const listenOptions = host ? { host, port } : { port }; @@ -618,25 +599,28 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< yield* Effect.addFinalizer(() => Scope.close(subscriptionsScope, Exit.void)); yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) => - broadcastPush({ - type: "push", - channel: ORCHESTRATION_WS_CHANNELS.domainEvent, - data: event, - }), + pushBus.publishAll(ORCHESTRATION_WS_CHANNELS.domainEvent, event), ).pipe(Effect.forkIn(subscriptionsScope)); - yield* Stream.runForEach(keybindingsManager.changes, (event) => - broadcastPush({ - type: "push", - channel: WS_CHANNELS.serverConfigUpdated, - data: { - issues: event.issues, - providers: providerStatuses, - }, - }), + // Subscribe synchronously so the subscription exists before the + // readiness barrier — avoids a race where the polling watcher + // publishes before a lazily-created Stream.fromPubSub subscription. + const keybindingsChangesSub = yield* keybindingsManager.subscribeChanges.pipe( + Scope.provide(subscriptionsScope), + ); + yield* Effect.forever( + PubSub.take(keybindingsChangesSub).pipe( + Effect.flatMap((event) => + pushBus.publishAll(WS_CHANNELS.serverConfigUpdated, { + issues: event.issues, + providers: providerStatuses, + }), + ), + ), ).pipe(Effect.forkIn(subscriptionsScope)); yield* Scope.provide(orchestrationReactor.start, subscriptionsScope); + yield* readiness.markOrchestrationSubscriptionsReady; let welcomeBootstrapProjectId: ProjectId | undefined; let welcomeBootstrapThreadId: ThreadId | undefined; @@ -707,22 +691,20 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< const runPromise = Effect.runPromiseWith(runtimeServices); const unsubscribeTerminalEvents = yield* terminalManager.subscribe( - (event) => void Effect.runPromise(onTerminalEvent(event)), + (event) => void Effect.runPromise(pushBus.publishAll(WS_CHANNELS.terminalEvent, event)), ); yield* Effect.addFinalizer(() => Effect.sync(() => unsubscribeTerminalEvents())); + yield* readiness.markTerminalSubscriptionsReady; yield* NodeHttpServer.make(() => httpServer, listenOptions).pipe( Effect.mapError((cause) => new ServerLifecycleError({ operation: "httpServerListen", cause })), ); + yield* readiness.markHttpListening; yield* Effect.addFinalizer(() => Effect.all([ closeAllClients, - closeWebSocketServer.pipe( - Effect.catch((error) => - Effect.logWarning("failed to close web socket server", { cause: error }), - ), - ), + closeWebSocketServer.pipe(Effect.ignoreCause({ log: true })), ]), ); @@ -777,14 +759,16 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< relativePath: body.relativePath, path, }); - yield* fileSystem.makeDirectory(path.dirname(target.absolutePath), { recursive: true }).pipe( - Effect.mapError( - (cause) => - new RouteRequestError({ - message: `Failed to prepare workspace path: ${String(cause)}`, - }), - ), - ); + yield* fileSystem + .makeDirectory(path.dirname(target.absolutePath), { recursive: true }) + .pipe( + Effect.mapError( + (cause) => + new RouteRequestError({ + message: `Failed to prepare workspace path: ${String(cause)}`, + }), + ), + ); yield* fileSystem.writeFileString(target.absolutePath, body.contents).pipe( Effect.mapError( (cause) => @@ -968,48 +952,36 @@ export const createServer = Effect.fn(function* (): Effect.fn.Return< }); wss.on("connection", (ws) => { - void runPromise(Ref.update(clients, (clients) => clients.add(ws))); - const segments = cwd.split(/[/\\]/).filter(Boolean); const projectName = segments[segments.length - 1] ?? "project"; - const welcome: WsPush = { - type: "push", - channel: WS_CHANNELS.serverWelcome, - data: { - cwd, - projectName, - ...(welcomeBootstrapProjectId ? { bootstrapProjectId: welcomeBootstrapProjectId } : {}), - ...(welcomeBootstrapThreadId ? { bootstrapThreadId: welcomeBootstrapThreadId } : {}), - }, + const welcomeData = { + cwd, + projectName, + ...(welcomeBootstrapProjectId ? { bootstrapProjectId: welcomeBootstrapProjectId } : {}), + ...(welcomeBootstrapThreadId ? { bootstrapThreadId: welcomeBootstrapThreadId } : {}), }; - logOutgoingPush(welcome, 1); - ws.send(JSON.stringify(welcome)); + void runPromise( + Effect.gen(function* () { + yield* pushBus.registerClient(ws); + yield* readiness.awaitServerReady; + yield* pushBus.publishClient(ws, WS_CHANNELS.serverWelcome, welcomeData); + yield* pushBus.activateClient(ws); + }), + ); ws.on("message", (raw) => { void runPromise( - handleMessage(ws, raw).pipe( - Effect.catch((error) => Effect.logError("Error handling message", error)), - ), + handleMessage(ws, raw).pipe(Effect.ignoreCause({ log: true })), ); }); ws.on("close", () => { - void runPromise( - Ref.update(clients, (clients) => { - clients.delete(ws); - return clients; - }), - ); + void runPromise(pushBus.unregisterClient(ws)); }); ws.on("error", () => { - void runPromise( - Ref.update(clients, (clients) => { - clients.delete(ws); - return clients; - }), - ); + void runPromise(pushBus.unregisterClient(ws)); }); }); diff --git a/apps/server/src/wsServer/pushBus.test.ts b/apps/server/src/wsServer/pushBus.test.ts new file mode 100644 index 000000000000..133b76928962 --- /dev/null +++ b/apps/server/src/wsServer/pushBus.test.ts @@ -0,0 +1,111 @@ +import type { WebSocket } from "ws"; +import { afterEach, describe, expect, it } from "vitest"; +import { Effect, Exit, Scope } from "effect"; +import { WS_CHANNELS } from "@t3tools/contracts"; + +import { makeServerPushBus } from "./pushBus"; + +class MockWebSocket { + static readonly OPEN = 1; + + readonly OPEN = MockWebSocket.OPEN; + readyState = MockWebSocket.OPEN; + readonly sent: string[] = []; + private readonly waiters = new Set<() => void>(); + + send(message: string) { + this.sent.push(message); + for (const waiter of this.waiters) { + waiter(); + } + } + + waitForSentCount(count: number): Promise { + if (this.sent.length >= count) { + return Promise.resolve(); + } + + return new Promise((resolve) => { + const check = () => { + if (this.sent.length < count) { + return; + } + this.waiters.delete(check); + resolve(); + }; + + this.waiters.add(check); + }); + } +} + +describe("makeServerPushBus", () => { + let scope: Scope.Closeable | null = null; + + afterEach(async () => { + if (scope) { + await Effect.runPromise(Scope.close(scope, Exit.void)); + } + scope = null; + }); + + it("queues publishAll pushes for pre_welcome clients and flushes after activation", async () => { + scope = await Effect.runPromise(Scope.make("sequential")); + + const client = new MockWebSocket(); + const pushBus = await Effect.runPromise( + makeServerPushBus({ + logOutgoingPush: () => {}, + }).pipe(Scope.provide(scope)), + ); + + await Effect.runPromise( + Effect.gen(function* () { + yield* pushBus.registerClient(client as unknown as WebSocket); + yield* pushBus.publishAll(WS_CHANNELS.serverConfigUpdated, { + issues: [{ kind: "keybindings.malformed-config", message: "queued-before-welcome" }], + providers: [], + }); + + const delivered = yield* pushBus.publishClient( + client as unknown as WebSocket, + WS_CHANNELS.serverWelcome, + { + cwd: "/tmp/project", + projectName: "project", + }, + ); + expect(delivered).toBe(true); + + yield* pushBus.activateClient(client as unknown as WebSocket); + }), + ); + + await client.waitForSentCount(2); + + const messages = client.sent.map( + (message) => JSON.parse(message) as { channel: string; data: unknown; sequence: number }, + ); + + expect(messages).toEqual([ + { + type: "push", + sequence: 2, + channel: WS_CHANNELS.serverWelcome, + data: { + cwd: "/tmp/project", + projectName: "project", + }, + }, + { + type: "push", + sequence: 1, + channel: WS_CHANNELS.serverConfigUpdated, + data: { + issues: [{ kind: "keybindings.malformed-config", message: "queued-before-welcome" }], + providers: [], + }, + }, + ]); + }); +}); diff --git a/apps/server/src/wsServer/pushBus.ts b/apps/server/src/wsServer/pushBus.ts new file mode 100644 index 000000000000..bdb56dc65b1f --- /dev/null +++ b/apps/server/src/wsServer/pushBus.ts @@ -0,0 +1,206 @@ +import { type WsPushChannel, type WsPushData, type WsPush } from "@t3tools/contracts"; +import { Deferred, Effect, Queue, Ref } from "effect"; +import type { Scope } from "effect"; +import type { WebSocket } from "ws"; + +type PushTarget = + | { readonly kind: "all" } + | { readonly kind: "client"; readonly client: WebSocket }; + +type PushJob = + | { + readonly kind: "publish"; + readonly target: PushTarget; + readonly push: WsPush; + readonly delivered: Deferred.Deferred | null; + } + | { + readonly kind: "activate"; + readonly client: WebSocket; + readonly activated: Deferred.Deferred; + }; + +type RegisteredClientState = + | { readonly phase: "pre_welcome"; readonly backlog: WsPush[] } + | { readonly phase: "active" }; + +const PRE_WELCOME_BACKLOG_LIMIT = 256; + +export interface ServerPushBus { + readonly registerClient: (client: WebSocket) => Effect.Effect; + readonly activateClient: (client: WebSocket) => Effect.Effect; + readonly unregisterClient: (client: WebSocket) => Effect.Effect; + readonly publishAll: ( + channel: C, + data: WsPushData, + ) => Effect.Effect; + readonly publishClient: ( + client: WebSocket, + channel: C, + data: WsPushData, + ) => Effect.Effect; +} + +export const makeServerPushBus = (input: { + readonly logOutgoingPush: (push: WsPush, recipients: number) => void; +}): Effect.Effect => + Effect.gen(function* () { + const nextSequence = yield* Ref.make(0); + const queue = yield* Queue.unbounded(); + const clients = yield* Ref.make(new Map()); + + const settleDelivery = (job: PushJob, delivered: boolean) => + job.kind !== "publish" || job.delivered === null + ? Effect.void + : Deferred.succeed(job.delivered, delivered).pipe(Effect.orDie); + + const sendPushToClient = (client: WebSocket, message: string) => + Effect.sync(() => { + if (client.readyState !== client.OPEN) { + return false; + } + client.send(message); + return true; + }); + + const send = Effect.fnUntraced(function* (job: PushJob) { + if (job.kind === "activate") { + const state = yield* Ref.get(clients).pipe(Effect.map((current) => current.get(job.client))); + if (!state || state.phase !== "pre_welcome") { + yield* Deferred.succeed(job.activated, undefined).pipe(Effect.orDie); + return false; + } + + yield* Ref.update(clients, (current) => { + const next = new Map(current); + next.set(job.client, { phase: "active" }); + return next; + }); + + for (const queuedPush of state.backlog) { + const message = JSON.stringify(queuedPush); + const delivered = yield* sendPushToClient(job.client, message); + input.logOutgoingPush(queuedPush, delivered ? 1 : 0); + } + + yield* Deferred.succeed(job.activated, undefined).pipe(Effect.orDie); + return false; + } + + const { push } = job; + const message = JSON.stringify(push); + + let recipientCount = 0; + if (job.target.kind === "all") { + yield* Ref.update(clients, (current) => { + const next = new Map(current); + for (const [client, state] of next) { + if (state.phase === "active") { + continue; + } + const nextBacklog = [...state.backlog, push]; + if (nextBacklog.length > PRE_WELCOME_BACKLOG_LIMIT) { + nextBacklog.splice(0, nextBacklog.length - PRE_WELCOME_BACKLOG_LIMIT); + } + next.set(client, { + phase: "pre_welcome", + backlog: nextBacklog, + }); + } + return next; + }); + + const current = yield* Ref.get(clients); + for (const [client, state] of current) { + if (state.phase !== "active") { + continue; + } + if (yield* sendPushToClient(client, message)) { + recipientCount += 1; + } + } + } else if (yield* sendPushToClient(job.target.client, message)) { + recipientCount = 1; + } + + input.logOutgoingPush(push, recipientCount); + return recipientCount > 0; + }); + + yield* Effect.forkScoped( + Effect.forever( + Queue.take(queue).pipe( + Effect.flatMap((job) => + send(job).pipe( + Effect.tap((delivered) => settleDelivery(job, delivered)), + Effect.tapCause(() => settleDelivery(job, false)), + Effect.ignoreCause({ log: true }), + ), + ), + ), + ), + ); + + const makePush = (channel: C, data: WsPushData) => + Ref.updateAndGet(nextSequence, (current) => current + 1).pipe( + Effect.map( + (sequence) => + ({ + type: "push", + sequence, + channel, + data, + }) as WsPush, + ), + ); + + const publish = + (target: PushTarget) => + (channel: C, data: WsPushData) => + Effect.gen(function* () { + const push = yield* makePush(channel, data); + yield* Queue.offer(queue, { + kind: "publish", + target, + push, + delivered: null, + }).pipe(Effect.asVoid); + }); + + return { + registerClient: (client) => + Ref.update(clients, (current) => { + const next = new Map(current); + next.set(client, { phase: "pre_welcome", backlog: [] }); + return next; + }), + activateClient: (client) => + Effect.gen(function* () { + const activated = yield* Deferred.make(); + yield* Queue.offer(queue, { + kind: "activate", + client, + activated, + }).pipe(Effect.asVoid); + yield* Deferred.await(activated); + }), + unregisterClient: (client) => + Ref.update(clients, (current) => { + const next = new Map(current); + next.delete(client); + return next; + }), + publishAll: publish({ kind: "all" }), + publishClient: (client, channel, data) => + Effect.gen(function* () { + const delivered = yield* Deferred.make(); + yield* Queue.offer(queue, { + kind: "publish", + target: { kind: "client", client }, + push: yield* makePush(channel, data), + delivered, + }).pipe(Effect.asVoid); + return yield* Deferred.await(delivered); + }), + } satisfies ServerPushBus; + }); diff --git a/apps/server/src/wsServer/readiness.ts b/apps/server/src/wsServer/readiness.ts new file mode 100644 index 000000000000..2a973a8636f6 --- /dev/null +++ b/apps/server/src/wsServer/readiness.ts @@ -0,0 +1,36 @@ +import { Deferred, Effect } from "effect"; + +export interface ServerReadiness { + readonly awaitServerReady: Effect.Effect; + readonly markHttpListening: Effect.Effect; + readonly markPushBusReady: Effect.Effect; + readonly markKeybindingsReady: Effect.Effect; + readonly markTerminalSubscriptionsReady: Effect.Effect; + readonly markOrchestrationSubscriptionsReady: Effect.Effect; +} + +export const makeServerReadiness = Effect.gen(function* () { + const httpListening = yield* Deferred.make(); + const pushBusReady = yield* Deferred.make(); + const keybindingsReady = yield* Deferred.make(); + const terminalSubscriptionsReady = yield* Deferred.make(); + const orchestrationSubscriptionsReady = yield* Deferred.make(); + + const complete = (deferred: Deferred.Deferred) => + Deferred.succeed(deferred, undefined).pipe(Effect.orDie); + + return { + awaitServerReady: Effect.all([ + Deferred.await(httpListening), + Deferred.await(pushBusReady), + Deferred.await(keybindingsReady), + Deferred.await(terminalSubscriptionsReady), + Deferred.await(orchestrationSubscriptionsReady), + ]).pipe(Effect.asVoid), + markHttpListening: complete(httpListening), + markPushBusReady: complete(pushBusReady), + markKeybindingsReady: complete(keybindingsReady), + markTerminalSubscriptionsReady: complete(terminalSubscriptionsReady), + markOrchestrationSubscriptionsReady: complete(orchestrationSubscriptionsReady), + } satisfies ServerReadiness; +}); diff --git a/apps/server/vitest.config.ts b/apps/server/vitest.config.ts new file mode 100644 index 000000000000..cb86636fb6f1 --- /dev/null +++ b/apps/server/vitest.config.ts @@ -0,0 +1,13 @@ +import { defineConfig, mergeConfig } from "vitest/config"; + +import baseConfig from "../../vitest.config"; + +export default mergeConfig( + baseConfig, + defineConfig({ + test: { + testTimeout: 15_000, + hookTimeout: 15_000, + }, + }), +); diff --git a/apps/web/src/components/ChatView.browser.tsx b/apps/web/src/components/ChatView.browser.tsx index e2fd573fe8ce..9d69f5995a9b 100644 --- a/apps/web/src/components/ChatView.browser.tsx +++ b/apps/web/src/components/ChatView.browser.tsx @@ -305,6 +305,7 @@ const worker = setupWorker( client.send( JSON.stringify({ type: "push", + sequence: 1, channel: WS_CHANNELS.serverWelcome, data: fixture.welcome, }), diff --git a/apps/web/src/routes/__root.tsx b/apps/web/src/routes/__root.tsx index eb3eca9cbd60..8947fcc881e7 100644 --- a/apps/web/src/routes/__root.tsx +++ b/apps/web/src/routes/__root.tsx @@ -187,10 +187,29 @@ function EventRouter() { void syncSnapshot().catch(() => undefined); + const recoverOrchestrationGap = async (fromSequenceExclusive: number) => { + try { + const replayed = await api.orchestration.replayEvents(fromSequenceExclusive); + if (disposed) { + return; + } + for (const replayedEvent of replayed) { + latestSequence = Math.max(latestSequence, replayedEvent.sequence); + } + } catch { + // Fall back to snapshot sync when replay is unavailable. + } + await syncSnapshot(); + }; + const unsubDomainEvent = api.orchestration.onDomainEvent((event) => { if (event.sequence <= latestSequence) { return; } + if (event.sequence > latestSequence + 1) { + void recoverOrchestrationGap(latestSequence).catch(() => undefined); + return; + } latestSequence = event.sequence; if (event.type === "thread.turn-diff-completed" || event.type === "thread.reverted") { void queryClient.invalidateQueries({ queryKey: providerQueryKeys.all }); diff --git a/apps/web/src/wsNativeApi.test.ts b/apps/web/src/wsNativeApi.test.ts index 142174fb012c..11ab73a56adc 100644 --- a/apps/web/src/wsNativeApi.test.ts +++ b/apps/web/src/wsNativeApi.test.ts @@ -1,12 +1,18 @@ import { CommandId, type ContextMenuItem, + EventId, ORCHESTRATION_WS_CHANNELS, ORCHESTRATION_WS_METHODS, + type OrchestrationEvent, ProjectId, ThreadId, + type WsPushChannel, + type WsPushData, + type WsPushMessage, WS_CHANNELS, WS_METHODS, + type WsPush, type ServerProviderStatus, } from "@t3tools/contracts"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; @@ -18,12 +24,19 @@ const showContextMenuFallbackMock = vi.fn< position?: { x: number; y: number }, ) => Promise >(); -const channelListeners = new Map void>>(); -const subscribeMock = vi.fn<(channel: string, listener: (data: unknown) => void) => () => void>( - (channel, listener) => { - const listeners = channelListeners.get(channel) ?? new Set<(data: unknown) => void>(); +const channelListeners = new Map void>>(); +const latestPushByChannel = new Map(); +const subscribeMock = vi.fn< + (channel: string, listener: (message: WsPush) => void, options?: { replayLatest?: boolean }) => () => void +>( + (channel, listener, options) => { + const listeners = channelListeners.get(channel) ?? new Set<(message: WsPush) => void>(); listeners.add(listener); channelListeners.set(channel, listeners); + const latest = latestPushByChannel.get(channel); + if (latest && options?.replayLatest) { + listener(latest); + } return () => { listeners.delete(listener); if (listeners.size === 0) { @@ -38,6 +51,9 @@ vi.mock("./wsTransport", () => { WsTransport: class MockWsTransport { request = requestMock; subscribe = subscribeMock; + getLatestPush(channel: string) { + return latestPushByChannel.get(channel) ?? null; + } }, }; }); @@ -46,11 +62,20 @@ vi.mock("./contextMenuFallback", () => ({ showContextMenuFallback: showContextMenuFallbackMock, })); -function emitPush(channel: string, data: unknown): void { +let nextPushSequence = 1; + +function emitPush(channel: C, data: WsPushData): void { const listeners = channelListeners.get(channel); + const message = { + type: "push" as const, + sequence: nextPushSequence++, + channel, + data, + } as WsPushMessage; + latestPushByChannel.set(channel, message); if (!listeners) return; for (const listener of listeners) { - listener(data); + listener(message); } } @@ -80,6 +105,8 @@ beforeEach(() => { showContextMenuFallbackMock.mockReset(); subscribeMock.mockClear(); channelListeners.clear(); + latestPushByChannel.clear(); + nextPushSequence = 1; Reflect.deleteProperty(getWindowForTest(), "desktopBridge"); }); @@ -89,7 +116,6 @@ afterEach(() => { describe("wsNativeApi", () => { it("delivers and caches valid server.welcome payloads", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); const { createWsNativeApi, onServerWelcome } = await import("./wsNativeApi"); createWsNativeApi(); @@ -107,7 +133,6 @@ describe("wsNativeApi", () => { expect(lateListener).toHaveBeenCalledTimes(1); expect(lateListener).toHaveBeenCalledWith(expect.objectContaining(payload)); - expect(warnSpy).not.toHaveBeenCalled(); }); it("preserves bootstrap ids from server.welcome payloads", async () => { @@ -120,8 +145,8 @@ describe("wsNativeApi", () => { emitPush(WS_CHANNELS.serverWelcome, { cwd: "/tmp/workspace", projectName: "t3-code", - bootstrapProjectId: "project-1", - bootstrapThreadId: "thread-1", + bootstrapProjectId: ProjectId.makeUnsafe("project-1"), + bootstrapThreadId: ThreadId.makeUnsafe("thread-1"), }); expect(listener).toHaveBeenCalledTimes(1); @@ -135,31 +160,26 @@ describe("wsNativeApi", () => { ); }); - it("ignores invalid server.welcome payloads and keeps subscription active", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + it("delivers successive server.welcome payloads to active listeners", async () => { const { createWsNativeApi, onServerWelcome } = await import("./wsNativeApi"); createWsNativeApi(); const listener = vi.fn(); onServerWelcome(listener); - emitPush(WS_CHANNELS.serverWelcome, { cwd: 42, projectName: "t3-code" }); + emitPush(WS_CHANNELS.serverWelcome, { cwd: "/tmp/one", projectName: "one" }); emitPush(WS_CHANNELS.serverWelcome, { cwd: "/tmp/workspace", projectName: "t3-code" }); - expect(listener).toHaveBeenCalledTimes(1); - expect(listener).toHaveBeenCalledWith( - expect.objectContaining({ cwd: "/tmp/workspace", projectName: "t3-code" }), + expect(listener).toHaveBeenCalledTimes(2); + expect(listener).toHaveBeenLastCalledWith( + expect.objectContaining({ + cwd: "/tmp/workspace", + projectName: "t3-code", + }), ); - expect(warnSpy).toHaveBeenCalledTimes(1); - expect(warnSpy).toHaveBeenCalledWith("Dropped inbound WebSocket push payload", { - reason: "decode-failed", - raw: { cwd: 42, projectName: "t3-code" }, - issue: expect.stringContaining("SchemaError"), - }); }); it("delivers and caches valid server.configUpdated payloads", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); const { createWsNativeApi, onServerConfigUpdated } = await import("./wsNativeApi"); createWsNativeApi(); @@ -185,11 +205,9 @@ describe("wsNativeApi", () => { onServerConfigUpdated(lateListener); expect(lateListener).toHaveBeenCalledTimes(1); expect(lateListener).toHaveBeenCalledWith(payload); - expect(warnSpy).not.toHaveBeenCalled(); }); - it("drops malformed server.configUpdated payloads", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); + it("delivers successive server.configUpdated payloads to active listeners", async () => { const { createWsNativeApi, onServerConfigUpdated } = await import("./wsNativeApi"); createWsNativeApi(); @@ -197,24 +215,22 @@ describe("wsNativeApi", () => { onServerConfigUpdated(listener); emitPush(WS_CHANNELS.serverConfigUpdated, { - issues: [{ kind: "keybindings.invalid-entry", message: "missing index" }], + issues: [{ kind: "keybindings.malformed-config", message: "bad json" }], providers: defaultProviders, }); emitPush(WS_CHANNELS.serverConfigUpdated, { - issues: [{ kind: "keybindings.malformed-config", message: "bad json" }], + issues: [], providers: defaultProviders, }); - expect(listener).toHaveBeenCalledTimes(1); - expect(listener).toHaveBeenCalledWith({ - issues: [{ kind: "keybindings.malformed-config", message: "bad json" }], + expect(listener).toHaveBeenCalledTimes(2); + expect(listener).toHaveBeenLastCalledWith({ + issues: [], providers: defaultProviders, }); - expect(warnSpy).toHaveBeenCalledTimes(1); }); it("forwards valid terminal and orchestration events", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); const { createWsNativeApi } = await import("./wsNativeApi"); const api = createWsNativeApi(); @@ -235,9 +251,9 @@ describe("wsNativeApi", () => { const orchestrationEvent = { sequence: 1, - eventId: "event-1", + eventId: EventId.makeUnsafe("event-1"), aggregateKind: "project", - aggregateId: "project-1", + aggregateId: ProjectId.makeUnsafe("project-1"), occurredAt: "2026-02-24T00:00:00.000Z", commandId: null, causationEventId: null, @@ -245,7 +261,7 @@ describe("wsNativeApi", () => { metadata: {}, type: "project.created", payload: { - projectId: "project-1", + projectId: ProjectId.makeUnsafe("project-1"), title: "Project", workspaceRoot: "/tmp/workspace", defaultModel: null, @@ -253,61 +269,13 @@ describe("wsNativeApi", () => { createdAt: "2026-02-24T00:00:00.000Z", updatedAt: "2026-02-24T00:00:00.000Z", }, - } as const; + } satisfies Extract; emitPush(ORCHESTRATION_WS_CHANNELS.domainEvent, orchestrationEvent); expect(onTerminalEvent).toHaveBeenCalledTimes(1); expect(onTerminalEvent).toHaveBeenCalledWith(terminalEvent); expect(onDomainEvent).toHaveBeenCalledTimes(1); expect(onDomainEvent).toHaveBeenCalledWith(orchestrationEvent); - expect(warnSpy).not.toHaveBeenCalled(); - }); - - it("drops malformed terminal and orchestration push payloads", async () => { - const warnSpy = vi.spyOn(console, "warn").mockImplementation(() => {}); - const { createWsNativeApi } = await import("./wsNativeApi"); - - const api = createWsNativeApi(); - const onTerminalEvent = vi.fn(); - const onDomainEvent = vi.fn(); - - api.terminal.onEvent(onTerminalEvent); - api.orchestration.onDomainEvent(onDomainEvent); - - emitPush(WS_CHANNELS.terminalEvent, { - threadId: "thread-1", - terminalId: "", - createdAt: "2026-02-24T00:00:00.000Z", - type: "output", - data: "hello", - }); - emitPush(ORCHESTRATION_WS_CHANNELS.domainEvent, { - sequence: -1, - type: "project.created", - }); - - expect(onTerminalEvent).not.toHaveBeenCalled(); - expect(onDomainEvent).not.toHaveBeenCalled(); - expect(warnSpy).toHaveBeenCalledTimes(2); - expect(warnSpy).toHaveBeenNthCalledWith(1, "Dropped inbound WebSocket push payload", { - reason: "decode-failed", - raw: { - threadId: "thread-1", - terminalId: "", - createdAt: "2026-02-24T00:00:00.000Z", - type: "output", - data: "hello", - }, - issue: expect.stringContaining("SchemaError"), - }); - expect(warnSpy).toHaveBeenNthCalledWith(2, "Dropped inbound WebSocket push payload", { - reason: "decode-failed", - raw: { - sequence: -1, - type: "project.created", - }, - issue: expect.stringContaining("SchemaError"), - }); }); it("wraps orchestration dispatch commands in the command envelope", async () => { diff --git a/apps/web/src/wsNativeApi.ts b/apps/web/src/wsNativeApi.ts index 91e6a61107c3..51de31518a6e 100644 --- a/apps/web/src/wsNativeApi.ts +++ b/apps/web/src/wsNativeApi.ts @@ -1,16 +1,13 @@ import { - OrchestrationEvent, ORCHESTRATION_WS_CHANNELS, ORCHESTRATION_WS_METHODS, type ContextMenuItem, type NativeApi, ServerConfigUpdatedPayload, - TerminalEvent, WS_CHANNELS, WS_METHODS, - WsWelcomePayload, + type WsWelcomePayload, } from "@t3tools/contracts"; -import { Cause, Schema } from "effect"; import { showContextMenuFallback } from "./contextMenuFallback"; import { WsTransport } from "./wsTransport"; @@ -18,24 +15,6 @@ import { WsTransport } from "./wsTransport"; let instance: { api: NativeApi; transport: WsTransport } | null = null; const welcomeListeners = new Set<(payload: WsWelcomePayload) => void>(); const serverConfigUpdatedListeners = new Set<(payload: ServerConfigUpdatedPayload) => void>(); -let lastWelcome: WsWelcomePayload | null = null; -let lastServerConfigUpdated: ServerConfigUpdatedPayload | null = null; - -const decodeAndWarnOnFailure = ( - schema: Schema.Schema & { readonly DecodingServices: never }, - raw: unknown, -): T | null => { - const decoded = Schema.decodeUnknownExit(schema)(raw); - if (decoded._tag === "Failure") { - console.warn("Dropped inbound WebSocket push payload", { - reason: "decode-failed", - raw, - issue: Cause.pretty(decoded.cause), - }); - return null; - } - return decoded.value; -}; /** * Subscribe to the server welcome message. If a welcome was already received @@ -45,10 +24,10 @@ const decodeAndWarnOnFailure = ( export function onServerWelcome(listener: (payload: WsWelcomePayload) => void): () => void { welcomeListeners.add(listener); - // Replay cached welcome for late subscribers - if (lastWelcome) { + const latestWelcome = instance?.transport.getLatestPush(WS_CHANNELS.serverWelcome)?.data ?? null; + if (latestWelcome) { try { - listener(lastWelcome); + listener(latestWelcome); } catch { // Swallow listener errors } @@ -68,9 +47,10 @@ export function onServerConfigUpdated( ): () => void { serverConfigUpdatedListeners.add(listener); - if (lastServerConfigUpdated) { + const latestConfig = instance?.transport.getLatestPush(WS_CHANNELS.serverConfigUpdated)?.data ?? null; + if (latestConfig) { try { - listener(lastServerConfigUpdated); + listener(latestConfig); } catch { // Swallow listener errors } @@ -86,12 +66,8 @@ export function createWsNativeApi(): NativeApi { const transport = new WsTransport(); - // Listen for server welcome and forward to registered listeners. - // Also cache it so late subscribers (React effects) get it immediately. - transport.subscribe(WS_CHANNELS.serverWelcome, (data) => { - const payload = decodeAndWarnOnFailure(WsWelcomePayload, data); - if (!payload) return; - lastWelcome = payload; + transport.subscribe(WS_CHANNELS.serverWelcome, (message) => { + const payload = message.data; for (const listener of welcomeListeners) { try { listener(payload); @@ -100,10 +76,8 @@ export function createWsNativeApi(): NativeApi { } } }); - transport.subscribe(WS_CHANNELS.serverConfigUpdated, (data) => { - const payload = decodeAndWarnOnFailure(ServerConfigUpdatedPayload, data); - if (!payload) return; - lastServerConfigUpdated = payload; + transport.subscribe(WS_CHANNELS.serverConfigUpdated, (message) => { + const payload = message.data; for (const listener of serverConfigUpdatedListeners) { try { listener(payload); @@ -134,10 +108,7 @@ export function createWsNativeApi(): NativeApi { restart: (input) => transport.request(WS_METHODS.terminalRestart, input), close: (input) => transport.request(WS_METHODS.terminalClose, input), onEvent: (callback) => - transport.subscribe(WS_CHANNELS.terminalEvent, (data) => { - const payload = decodeAndWarnOnFailure(TerminalEvent, data); - if (payload) callback(payload); - }), + transport.subscribe(WS_CHANNELS.terminalEvent, (message) => callback(message.data)), }, projects: { searchEntries: (input) => transport.request(WS_METHODS.projectsSearchEntries, input), @@ -196,10 +167,9 @@ export function createWsNativeApi(): NativeApi { replayEvents: (fromSequenceExclusive) => transport.request(ORCHESTRATION_WS_METHODS.replayEvents, { fromSequenceExclusive }), onDomainEvent: (callback) => - transport.subscribe(ORCHESTRATION_WS_CHANNELS.domainEvent, (data) => { - const payload = decodeAndWarnOnFailure(OrchestrationEvent, data); - if (payload) callback(payload); - }), + transport.subscribe(ORCHESTRATION_WS_CHANNELS.domainEvent, (message) => + callback(message.data), + ), }, }; diff --git a/apps/web/src/wsTransport.test.ts b/apps/web/src/wsTransport.test.ts index e47073fc21c0..0aae1f1fffe9 100644 --- a/apps/web/src/wsTransport.test.ts +++ b/apps/web/src/wsTransport.test.ts @@ -1,3 +1,4 @@ +import { WS_CHANNELS } from "@t3tools/contracts"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { WsTransport } from "./wsTransport"; @@ -80,6 +81,7 @@ beforeEach(() => { afterEach(() => { globalThis.WebSocket = originalWebSocket; + vi.useRealTimers(); vi.restoreAllMocks(); }); @@ -90,18 +92,24 @@ describe("WsTransport", () => { socket.open(); const listener = vi.fn(); - transport.subscribe("providers.event", listener); + transport.subscribe(WS_CHANNELS.serverConfigUpdated, listener); socket.serverMessage( JSON.stringify({ type: "push", - channel: "providers.event", - data: { status: "ok" }, + sequence: 1, + channel: WS_CHANNELS.serverConfigUpdated, + data: { issues: [], providers: [] }, }), ); expect(listener).toHaveBeenCalledTimes(1); - expect(listener).toHaveBeenCalledWith({ status: "ok" }); + expect(listener).toHaveBeenCalledWith({ + type: "push", + sequence: 1, + channel: WS_CHANNELS.serverConfigUpdated, + data: { issues: [], providers: [] }, + }); transport.dispose(); }); @@ -137,12 +145,13 @@ describe("WsTransport", () => { socket.open(); const listener = vi.fn(); - transport.subscribe("providers.event", listener); + transport.subscribe(WS_CHANNELS.serverConfigUpdated, listener); socket.serverMessage("{ invalid-json"); socket.serverMessage( JSON.stringify({ type: "push", + sequence: 2, channel: 42, data: { bad: true }, }), @@ -150,26 +159,93 @@ describe("WsTransport", () => { socket.serverMessage( JSON.stringify({ type: "push", - channel: "providers.event", - data: { ok: true }, + sequence: 3, + channel: WS_CHANNELS.serverConfigUpdated, + data: { issues: [], providers: [] }, }), ); expect(listener).toHaveBeenCalledTimes(1); - expect(listener).toHaveBeenCalledWith({ ok: true }); - expect(warnSpy).toHaveBeenCalledTimes(2); - expect(warnSpy).toHaveBeenNthCalledWith(1, "Dropped inbound WebSocket envelope", { - reason: "decode-failed", - issue: - "SchemaError: SyntaxError: Expected property name or '}' in JSON at position 2 (line 1 column 3)", - raw: "{ invalid-json", + expect(listener).toHaveBeenCalledWith({ + type: "push", + sequence: 3, + channel: WS_CHANNELS.serverConfigUpdated, + data: { issues: [], providers: [] }, }); + expect(warnSpy).toHaveBeenCalledTimes(2); + expect(warnSpy).toHaveBeenNthCalledWith( + 1, + "Dropped inbound WebSocket envelope", + expect.objectContaining({ + code: "invalid-json", + reason: expect.stringContaining("Expected property name or '}'"), + rawKind: "string", + }), + ); expect(warnSpy).toHaveBeenNthCalledWith(2, "Dropped inbound WebSocket envelope", { - reason: "decode-failed", - issue: expect.stringContaining("SchemaError: Expected string, got 42"), - raw: '{"type":"push","channel":42,"data":{"bad":true}}', + code: "invalid-envelope", + reason: expect.stringContaining("Expected"), + rawKind: "object", + expected: "WsResponse", + actual: "object", }); transport.dispose(); }); + + it("queues requests until the websocket opens", async () => { + const transport = new WsTransport("ws://localhost:3020"); + const socket = getSocket(); + + const requestPromise = transport.request("projects.list"); + expect(socket.sent).toHaveLength(0); + + socket.open(); + expect(socket.sent).toHaveLength(1); + const requestEnvelope = JSON.parse(socket.sent[0] ?? "{}") as { id: string }; + socket.serverMessage( + JSON.stringify({ + id: requestEnvelope.id, + result: { projects: [] }, + }), + ); + + await expect(requestPromise).resolves.toEqual({ projects: [] }); + transport.dispose(); + }); + + it("does not send timed-out queued requests after reconnect", async () => { + vi.useFakeTimers(); + + const transport = new WsTransport("ws://localhost:3020"); + const firstSocket = getSocket(); + firstSocket.open(); + firstSocket.close(); + + const requestPromise = transport.request("projects.list"); + + await vi.advanceTimersByTimeAsync(60_000); + await expect(requestPromise).rejects.toThrow("Request timed out: projects.list"); + + await vi.advanceTimersByTimeAsync(500); + const secondSocket = getSocket(); + secondSocket.open(); + + expect(secondSocket.sent).toHaveLength(0); + transport.dispose(); + }); + + it("rejects and purges queued requests on dispose", async () => { + const transport = new WsTransport("ws://localhost:3020"); + const socket = getSocket(); + + const requestPromise = transport.request("projects.list"); + expect(socket.sent).toHaveLength(0); + + transport.dispose(); + + await expect(requestPromise).rejects.toThrow("Transport disposed"); + socket.open(); + expect(socket.sent).toHaveLength(0); + }); }); diff --git a/apps/web/src/wsTransport.ts b/apps/web/src/wsTransport.ts index ef9dddb347c7..fa222486538e 100644 --- a/apps/web/src/wsTransport.ts +++ b/apps/web/src/wsTransport.ts @@ -1,20 +1,42 @@ -import { WebSocketResponse, WsPush, WsResponse } from "@t3tools/contracts"; -import { Cause, Schema } from "effect"; +import { + type WsDecodeDiagnostic, + type WsPush, + type WsPushChannel, + type WsPushMessage, + WebSocketResponse, + type WsResponse as WsResponseMessage, + WsResponse as WsResponseSchema, +} from "@t3tools/contracts"; +import { Schema, SchemaIssue } from "effect"; -type PushListener = (data: unknown) => void; +type PushListener = (message: WsPushMessage) => void; interface PendingRequest { + method: string; resolve: (result: unknown) => void; reject: (error: Error) => void; timeout: ReturnType; } +interface OutboundRequest { + id: string; + encoded: string; +} + +interface SubscribeOptions { + readonly replayLatest?: boolean; +} + +type TransportState = "connecting" | "open" | "reconnecting" | "closed" | "disposed"; + const REQUEST_TIMEOUT_MS = 60_000; const RECONNECT_DELAYS_MS = [500, 1_000, 2_000, 4_000, 8_000]; -const decodeWsResponseFromJson = Schema.decodeUnknownExit(Schema.fromJsonString(WsResponse)); -const isWsPushEnvelope = Schema.is(WsPush); +const decodeWsResponse = Schema.decodeUnknownSync(WsResponseSchema); const isWebSocketResponseEnvelope = Schema.is(WebSocketResponse); +const isWsPushMessage = (value: WsResponseMessage): value is WsPush => + "type" in value && value.type === "push"; + interface WsRequestEnvelope { id: string; body: { @@ -23,20 +45,106 @@ interface WsRequestEnvelope { }; } +function describeValue(value: unknown): string { + if (value === null) return "null"; + if (Array.isArray(value)) return "array"; + return typeof value; +} + +function parseJsonOffset(error: unknown): number | undefined { + if (!(error instanceof Error)) { + return undefined; + } + const match = /position\s+(\d+)/i.exec(error.message); + if (!match) { + return undefined; + } + const parsed = Number(match[1]); + return Number.isFinite(parsed) && parsed >= 0 ? parsed : undefined; +} + +function makeEnvelopeDiagnostic(raw: unknown, reason: string): WsDecodeDiagnostic { + return { + code: "invalid-envelope", + reason, + rawKind: describeValue(raw), + expected: "WsResponse", + actual: describeValue(raw), + }; +} + +function decodeInboundMessage( + raw: unknown, +): { readonly ok: true; readonly message: WsResponseMessage } | { readonly ok: false; readonly diagnostic: WsDecodeDiagnostic } { + if (typeof raw !== "string") { + return { + ok: false, + diagnostic: { + code: "invalid-envelope", + reason: "Expected a text WebSocket frame.", + rawKind: describeValue(raw), + expected: "string", + actual: describeValue(raw), + }, + }; + } + + let parsed: unknown; + try { + parsed = JSON.parse(raw); + } catch (error) { + return { + ok: false, + diagnostic: { + code: "invalid-json", + reason: error instanceof Error ? error.message : "Failed to parse JSON.", + rawKind: "string", + expected: "valid JSON string", + actual: raw, + ...(parseJsonOffset(error) !== undefined ? { jsonOffset: parseJsonOffset(error) } : {}), + }, + }; + } + + try { + return { ok: true, message: decodeWsResponse(parsed) }; + } catch (error) { + return { + ok: false, + diagnostic: makeEnvelopeDiagnostic( + parsed, + typeof error === "object" && error !== null && "issue" in error + ? SchemaIssue.makeFormatterDefault()((error as Schema.SchemaError).issue) + : error instanceof Error + ? error.message + : "Failed to decode WebSocket envelope.", + ), + }; + } +} + +function asError(value: unknown, fallback: string): Error { + if (value instanceof Error) { + return value; + } + return new Error(fallback); +} + export class WsTransport { private ws: WebSocket | null = null; private nextId = 1; private readonly pending = new Map(); - private readonly listeners = new Map>(); + private readonly listeners = new Map void>>(); + private readonly latestPushByChannel = new Map(); + private readonly outboundQueue: OutboundRequest[] = []; private reconnectAttempt = 0; private reconnectTimer: ReturnType | null = null; private disposed = false; + private state: TransportState = "connecting"; private readonly url: string; constructor(url?: string) { const bridgeUrl = window.desktopBridge?.getWsUrl(); - // In dev mode, VITE_WS_URL points to the server's WebSocket endpoint. - // In production, the page is served by the WS server on the same host:port. const envUrl = import.meta.env.VITE_WS_URL as string | undefined; this.url = url ?? @@ -52,44 +160,77 @@ export class WsTransport { if (typeof method !== "string" || method.length === 0) { throw new Error("Request method is required"); } + const id = String(this.nextId++); const body = params != null ? { ...params, _tag: method } : { _tag: method }; const message: WsRequestEnvelope = { id, body }; + const encoded = JSON.stringify(message); return new Promise((resolve, reject) => { const timeout = setTimeout(() => { + const pending = this.pending.get(id); + if (!pending) { + return; + } this.pending.delete(id); + this.removeQueuedRequest(id); reject(new Error(`Request timed out: ${method}`)); }, REQUEST_TIMEOUT_MS); this.pending.set(id, { + method, resolve: resolve as (result: unknown) => void, reject, timeout, }); - this.send(message); + this.send({ id, encoded }); }); } - subscribe(channel: string, listener: PushListener): () => void { + subscribe( + channel: C, + listener: PushListener, + options?: SubscribeOptions, + ): () => void { let channelListeners = this.listeners.get(channel); if (!channelListeners) { - channelListeners = new Set(); + channelListeners = new Set<(message: WsPush) => void>(); this.listeners.set(channel, channelListeners); } - channelListeners.add(listener); + + const wrappedListener = (message: WsPush) => { + listener(message as WsPushMessage); + }; + channelListeners.add(wrappedListener); + + if (options?.replayLatest) { + const latest = this.latestPushByChannel.get(channel); + if (latest) { + wrappedListener(latest); + } + } return () => { - channelListeners!.delete(listener); - if (channelListeners!.size === 0) { + channelListeners?.delete(wrappedListener); + if (channelListeners?.size === 0) { this.listeners.delete(channel); } }; } + getLatestPush(channel: C): WsPushMessage | null { + const latest = this.latestPushByChannel.get(channel); + return latest ? (latest as WsPushMessage) : null; + } + + getState(): TransportState { + return this.state; + } + dispose() { this.disposed = true; + this.state = "disposed"; if (this.reconnectTimer !== null) { clearTimeout(this.reconnectTimer); this.reconnectTimer = null; @@ -99,18 +240,24 @@ export class WsTransport { pending.reject(new Error("Transport disposed")); } this.pending.clear(); + this.outboundQueue.length = 0; this.ws?.close(); this.ws = null; } private connect() { - if (this.disposed) return; + if (this.disposed) { + return; + } + this.state = this.reconnectAttempt > 0 ? "reconnecting" : "connecting"; const ws = new WebSocket(this.url); ws.addEventListener("open", () => { this.ws = ws; + this.state = "open"; this.reconnectAttempt = 0; + this.flushQueue(); }); ws.addEventListener("message", (event) => { @@ -118,34 +265,37 @@ export class WsTransport { }); ws.addEventListener("close", () => { - this.ws = null; + if (this.ws === ws) { + this.ws = null; + } + if (this.disposed) { + this.state = "disposed"; + return; + } + this.state = "closed"; this.scheduleReconnect(); }); ws.addEventListener("error", () => { - // close event will fire after error + // close will follow }); } private handleMessage(raw: unknown) { - const exit = decodeWsResponseFromJson(raw); - if (exit._tag === "Failure") { - console.warn("Dropped inbound WebSocket envelope", { - reason: "decode-failed", - raw, - issue: Cause.pretty(exit.cause), - }); + const decoded = decodeInboundMessage(raw); + if (!decoded.ok) { + console.warn("Dropped inbound WebSocket envelope", decoded.diagnostic); return; } - const message = exit.value; - // Push event - if (isWsPushEnvelope(message)) { + const message = decoded.message; + if (isWsPushMessage(message)) { + this.latestPushByChannel.set(message.channel, message); const channelListeners = this.listeners.get(message.channel); if (channelListeners) { for (const listener of channelListeners) { try { - listener(message.data); + listener(message); } catch { // Swallow listener errors } @@ -154,57 +304,79 @@ export class WsTransport { return; } - // Response to a request if (!isWebSocketResponseEnvelope(message)) { return; } const pending = this.pending.get(message.id); - if (!pending) return; + if (!pending) { + return; + } clearTimeout(pending.timeout); this.pending.delete(message.id); if (message.error) { pending.reject(new Error(message.error.message)); - } else { - pending.resolve(message.result); + return; } + + pending.resolve(message.result); } - private send(message: WsRequestEnvelope) { - if (this.ws?.readyState === WebSocket.OPEN) { - this.ws.send(JSON.stringify(message)); + private send(entry: OutboundRequest) { + if (this.disposed) { return; } - // If not connected, wait for connection - const waitForOpen = () => { - const check = setInterval(() => { - if (this.disposed) { - clearInterval(check); - return; - } - if (this.ws?.readyState === WebSocket.OPEN) { - clearInterval(check); - this.ws.send(JSON.stringify(message)); - } - }, 50); + this.outboundQueue.push(entry); + try { + this.flushQueue(); + } catch { + // Swallow: flushQueue has queued the message for retry on reconnect + } + } - // Give up after timeout (the pending request will time out on its own) - setTimeout(() => clearInterval(check), REQUEST_TIMEOUT_MS); - }; - waitForOpen(); + private flushQueue() { + if (this.ws?.readyState !== WebSocket.OPEN) { + return; + } + + while (this.outboundQueue.length > 0) { + const message = this.outboundQueue.shift(); + if (!message) { + continue; + } + if (!this.pending.has(message.id)) { + continue; + } + try { + this.ws.send(message.encoded); + } catch (error) { + this.outboundQueue.unshift(message); + throw asError(error, "Failed to send WebSocket request."); + } + } + } + + private removeQueuedRequest(requestId: string) { + const index = this.outboundQueue.findIndex((entry) => entry.id === requestId); + if (index < 0) { + return; + } + this.outboundQueue.splice(index, 1); } private scheduleReconnect() { - if (this.disposed) return; + if (this.disposed || this.reconnectTimer !== null) { + return; + } const delay = RECONNECT_DELAYS_MS[Math.min(this.reconnectAttempt, RECONNECT_DELAYS_MS.length - 1)] ?? RECONNECT_DELAYS_MS[0]!; - this.reconnectAttempt++; + this.reconnectAttempt += 1; this.reconnectTimer = setTimeout(() => { this.reconnectTimer = null; this.connect(); diff --git a/packages/contracts/src/ws.test.ts b/packages/contracts/src/ws.test.ts index be103f3046ff..1876624e67f7 100644 --- a/packages/contracts/src/ws.test.ts +++ b/packages/contracts/src/ws.test.ts @@ -1,10 +1,11 @@ import { assert, it } from "@effect/vitest"; import { Effect, Schema } from "effect"; -import { ORCHESTRATION_WS_METHODS } from "./orchestration"; -import { WebSocketRequest } from "./ws"; +import { ORCHESTRATION_WS_CHANNELS, ORCHESTRATION_WS_METHODS } from "./orchestration"; +import { WebSocketRequest, WsResponse, WS_CHANNELS } from "./ws"; const decodeWebSocketRequest = Schema.decodeUnknownEffect(WebSocketRequest); +const decodeWsResponse = Schema.decodeUnknownEffect(WsResponse); it.effect("accepts getTurnDiff requests when fromTurnCount <= toTurnCount", () => Effect.gen(function* () { @@ -56,3 +57,43 @@ it.effect("trims websocket request id and nested orchestration ids", () => } }), ); + +it.effect("accepts typed websocket push envelopes with sequence", () => + Effect.gen(function* () { + const parsed = yield* decodeWsResponse({ + type: "push", + sequence: 1, + channel: WS_CHANNELS.serverWelcome, + data: { + cwd: "/tmp/workspace", + projectName: "workspace", + }, + }); + + if (!("type" in parsed) || parsed.type !== "push") { + assert.fail("expected websocket response to decode as a push envelope"); + } + + assert.strictEqual(parsed.type, "push"); + assert.strictEqual(parsed.sequence, 1); + assert.strictEqual(parsed.channel, WS_CHANNELS.serverWelcome); + }), +); + +it.effect("rejects push envelopes when channel payload does not match the channel schema", () => + Effect.gen(function* () { + const result = yield* Effect.exit( + decodeWsResponse({ + type: "push", + sequence: 2, + channel: ORCHESTRATION_WS_CHANNELS.domainEvent, + data: { + cwd: "/tmp/workspace", + projectName: "workspace", + }, + }), + ); + + assert.strictEqual(result._tag, "Failure"); + }), +); diff --git a/packages/contracts/src/ws.ts b/packages/contracts/src/ws.ts index 1100b4f9df5a..35f8ad430b4d 100644 --- a/packages/contracts/src/ws.ts +++ b/packages/contracts/src/ws.ts @@ -1,8 +1,10 @@ import { Schema, Struct } from "effect"; -import { ProjectId, ThreadId, TrimmedNonEmptyString } from "./baseSchemas"; +import { NonNegativeInt, ProjectId, ThreadId, TrimmedNonEmptyString } from "./baseSchemas"; import { ClientOrchestrationCommand, + OrchestrationEvent, + ORCHESTRATION_WS_CHANNELS, OrchestrationGetFullThreadDiffInput, ORCHESTRATION_WS_METHODS, OrchestrationGetSnapshotInput, @@ -23,6 +25,7 @@ import { import { TerminalClearInput, TerminalCloseInput, + TerminalEvent, TerminalOpenInput, TerminalResizeInput, TerminalRestartInput, @@ -31,6 +34,7 @@ import { import { KeybindingRule } from "./keybindings"; import { ProjectSearchEntriesInput, ProjectWriteFileInput } from "./project"; import { OpenInEditorInput } from "./editor"; +import { ServerConfigUpdatedPayload } from "./server"; // ── WebSocket RPC Method Names ─────────────────────────────────────── @@ -148,19 +152,8 @@ export const WebSocketResponse = Schema.Struct({ }); export type WebSocketResponse = typeof WebSocketResponse.Type; -export const WsPush = Schema.Struct({ - type: Schema.Literal("push"), - channel: TrimmedNonEmptyString, - data: Schema.Unknown, -}); -export type WsPush = typeof WsPush.Type; - -// ── Union of all server → client messages ───────────────────────────── - -export const WsResponse = Schema.Union([WebSocketResponse, WsPush]); -export type WsResponse = typeof WsResponse.Type; - -// ── Server welcome payload ─────────────────────────────────────────── +export const WsPushSequence = NonNegativeInt; +export type WsPushSequence = typeof WsPushSequence.Type; export const WsWelcomePayload = Schema.Struct({ cwd: TrimmedNonEmptyString, @@ -169,3 +162,85 @@ export const WsWelcomePayload = Schema.Struct({ bootstrapThreadId: Schema.optional(ThreadId), }); export type WsWelcomePayload = typeof WsWelcomePayload.Type; + +export const WsDecodeDiagnosticCode = Schema.Literals([ + "invalid-json", + "invalid-envelope", + "invalid-request", + "invalid-response", +]); +export type WsDecodeDiagnosticCode = typeof WsDecodeDiagnosticCode.Type; + +export const WsDecodeDiagnostic = Schema.Struct({ + code: WsDecodeDiagnosticCode, + reason: TrimmedNonEmptyString, + rawKind: Schema.optional(TrimmedNonEmptyString), + expected: Schema.optional(TrimmedNonEmptyString), + actual: Schema.optional(TrimmedNonEmptyString), + path: Schema.optional(Schema.Array(TrimmedNonEmptyString)), + jsonOffset: Schema.optional(NonNegativeInt), +}); +export type WsDecodeDiagnostic = typeof WsDecodeDiagnostic.Type; + +export interface WsPushPayloadByChannel { + readonly [WS_CHANNELS.serverWelcome]: WsWelcomePayload; + readonly [WS_CHANNELS.serverConfigUpdated]: typeof ServerConfigUpdatedPayload.Type; + readonly [WS_CHANNELS.terminalEvent]: typeof TerminalEvent.Type; + readonly [ORCHESTRATION_WS_CHANNELS.domainEvent]: OrchestrationEvent; +} + +export type WsPushChannel = keyof WsPushPayloadByChannel; +export type WsPushData = WsPushPayloadByChannel[C]; + +const makeWsPushSchema = >( + channel: Channel, + payload: Payload, +) => + Schema.Struct({ + type: Schema.Literal("push"), + sequence: WsPushSequence, + channel: Schema.Literal(channel), + data: payload, + }); + +export const WsPushServerWelcome = makeWsPushSchema(WS_CHANNELS.serverWelcome, WsWelcomePayload); +export const WsPushServerConfigUpdated = makeWsPushSchema( + WS_CHANNELS.serverConfigUpdated, + ServerConfigUpdatedPayload, +); +export const WsPushTerminalEvent = makeWsPushSchema(WS_CHANNELS.terminalEvent, TerminalEvent); +export const WsPushOrchestrationDomainEvent = makeWsPushSchema( + ORCHESTRATION_WS_CHANNELS.domainEvent, + OrchestrationEvent, +); + +export const WsPushChannelSchema = Schema.Literals([ + WS_CHANNELS.serverWelcome, + WS_CHANNELS.serverConfigUpdated, + WS_CHANNELS.terminalEvent, + ORCHESTRATION_WS_CHANNELS.domainEvent, +]); +export type WsPushChannelSchema = typeof WsPushChannelSchema.Type; + +export const WsPush = Schema.Union([ + WsPushServerWelcome, + WsPushServerConfigUpdated, + WsPushTerminalEvent, + WsPushOrchestrationDomainEvent, +]); +export type WsPush = typeof WsPush.Type; + +export type WsPushMessage = Extract; + +export const WsPushEnvelopeBase = Schema.Struct({ + type: Schema.Literal("push"), + sequence: WsPushSequence, + channel: WsPushChannelSchema, + data: Schema.Unknown, +}); +export type WsPushEnvelopeBase = typeof WsPushEnvelopeBase.Type; + +// ── Union of all server → client messages ───────────────────────────── + +export const WsResponse = Schema.Union([WebSocketResponse, WsPush]); +export type WsResponse = typeof WsResponse.Type; diff --git a/packages/shared/package.json b/packages/shared/package.json index b1a94c76048f..ad7697af5c28 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -23,6 +23,10 @@ "./Net": { "types": "./src/Net.ts", "import": "./src/Net.ts" + }, + "./DrainableWorker": { + "types": "./src/DrainableWorker.ts", + "import": "./src/DrainableWorker.ts" } }, "scripts": { diff --git a/packages/shared/src/DrainableWorker.test.ts b/packages/shared/src/DrainableWorker.test.ts new file mode 100644 index 000000000000..3dcb5e6eeb50 --- /dev/null +++ b/packages/shared/src/DrainableWorker.test.ts @@ -0,0 +1,76 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { Deferred, Effect, Exit, Scope } from "effect"; + +import { makeDrainableWorker } from "./DrainableWorker"; + +describe("makeDrainableWorker", () => { + let scope: Scope.Closeable | null = null; + + afterEach(async () => { + if (scope) { + await Effect.runPromise(Scope.close(scope, Exit.void)); + } + scope = null; + }); + + it("waits for work enqueued during active processing before draining", async () => { + scope = await Effect.runPromise(Scope.make("sequential")); + + const processed: string[] = []; + const { worker, firstStarted, releaseFirst, secondStarted, releaseSecond } = + await Effect.runPromise( + Effect.gen(function* () { + const firstStarted = yield* Deferred.make(); + const releaseFirst = yield* Deferred.make(); + const secondStarted = yield* Deferred.make(); + const releaseSecond = yield* Deferred.make(); + + const worker = yield* makeDrainableWorker((item: string) => + Effect.gen(function* () { + if (item === "first") { + yield* Deferred.succeed(firstStarted, undefined).pipe(Effect.orDie); + yield* Deferred.await(releaseFirst); + } + + if (item === "second") { + yield* Deferred.succeed(secondStarted, undefined).pipe(Effect.orDie); + yield* Deferred.await(releaseSecond); + } + + processed.push(item); + }), + ); + + return { + worker, + firstStarted, + releaseFirst, + secondStarted, + releaseSecond, + }; + }).pipe(Scope.provide(scope)), + ); + + await Effect.runPromise(worker.enqueue("first")); + await Effect.runPromise(Deferred.await(firstStarted)); + + const drainPromise = Effect.runPromise(worker.drain); + + await Effect.runPromise(worker.enqueue("second")); + await Effect.runPromise(Deferred.succeed(releaseFirst, undefined)); + await Effect.runPromise(Deferred.await(secondStarted)); + + const earlyResult = await Promise.race([ + drainPromise.then(() => "resolved" as const), + new Promise<"pending">((resolve) => { + setTimeout(() => resolve("pending"), 20); + }), + ]); + expect(earlyResult).toBe("pending"); + + await Effect.runPromise(Deferred.succeed(releaseSecond, undefined)); + await drainPromise; + + expect(processed).toEqual(["first", "second"]); + }); +}); diff --git a/packages/shared/src/DrainableWorker.ts b/packages/shared/src/DrainableWorker.ts new file mode 100644 index 000000000000..3a05523e1328 --- /dev/null +++ b/packages/shared/src/DrainableWorker.ts @@ -0,0 +1,82 @@ +/** + * DrainableWorker - A queue-based worker that exposes a `drain()` effect. + * + * Wraps the common `Queue.unbounded` + `Effect.forever` pattern and adds + * a signal that resolves when the queue is empty **and** the current item + * has finished processing. This lets tests replace timing-sensitive + * `Effect.sleep` calls with deterministic `drain()`. + * + * @module DrainableWorker + */ +import { Effect, Queue, Ref, Schedule } from "effect"; +import type { Scope } from "effect"; + +export interface DrainableWorker { + /** + * Enqueue a work item and track it for `drain()`. + * + * This wraps `Queue.offer` so drain state is updated atomically with the + * enqueue path instead of inferring it from queue internals. + */ + readonly enqueue: (item: A) => Effect.Effect; + + /** + * Resolves when the queue is empty and the worker is idle (not processing). + * + * Uses a tight `Schedule.spaced("1 millis")` poll which resolves in + * microseconds in practice — intended for test use only. + */ + readonly drain: Effect.Effect; +} + +/** + * Create a drainable worker that processes items from an unbounded queue. + * + * The worker is forked into the current scope and will be interrupted when + * the scope closes. A finalizer shuts down the queue. + * + * @param process - The effect to run for each queued item. + * @returns A `DrainableWorker` with `queue` and `drain`. + */ +export const makeDrainableWorker = ( + process: (item: A) => Effect.Effect, +): Effect.Effect, never, Scope.Scope | R> => + Effect.gen(function* () { + const queue = yield* Queue.unbounded(); + const outstanding = yield* Ref.make(0); + + yield* Effect.addFinalizer(() => Queue.shutdown(queue).pipe(Effect.asVoid)); + + yield* Effect.forkScoped( + Effect.forever( + Queue.take(queue).pipe( + Effect.flatMap((item) => + process(item).pipe( + Effect.ensuring(Ref.update(outstanding, (count) => Math.max(0, count - 1))), + ), + ), + ), + ), + ); + + const enqueue: DrainableWorker["enqueue"] = (item) => + Ref.update(outstanding, (count) => count + 1).pipe( + Effect.flatMap(() => Queue.offer(queue, item)), + Effect.flatMap((accepted) => + accepted === false + ? Ref.update(outstanding, (count) => Math.max(0, count - 1)) + : Effect.void, + ), + Effect.asVoid, + ); + + const drain: DrainableWorker["drain"] = Ref.get(outstanding).pipe( + Effect.repeat({ + while: (count) => count > 0, + schedule: Schedule.spaced("1 millis"), + }), + Effect.asVoid, + ); + + return { enqueue, drain } satisfies DrainableWorker; + });