diff --git a/packages/app-bundle/manifest.json b/packages/app-bundle/manifest.json index ab712920..b92d79ef 100644 --- a/packages/app-bundle/manifest.json +++ b/packages/app-bundle/manifest.json @@ -338,6 +338,9 @@ "packages/llm/src/protocols/bedrock-converse.ts": "834a415fa8ddf9ad1224133f8352d0a926694f19f0954201d182111c171ec03b", "packages/llm/test/provider/bedrock-converse.test.ts": "e8e14d60e8502e928f170f8ac5b406c44bf16d2dd34996ddc2fb7b9261decf85", "packages/opencode/src/agent/agent.ts": "512372a4c62997e9565266638ada19b9dc0a3685b0c0483fd28fede2abdb5c9b", + "packages/opencode/src/bus/global-replay.ts": "d4d64e027ab136655db088419b33e3ae524c1d3d8f892b5bb8697050ddc03d33", + "packages/opencode/src/bus/global.ts": "65b3b380936fc34b2c5ef9f3ab419a6c8e6d4cfc696302a266188f26ccdc320a", + "packages/opencode/src/control-plane/workspace.ts": "ab986282d1bd15b7f4473e33766884e17c97684ef0778cd8d84fd173604175cc", "packages/opencode/src/effect/runner.ts": "b660b5b22df810b16293c23c6d832e5e8dcfb12b9a8a858c875d2c051c855b81", "packages/opencode/src/mcp/browser.ts": "ecdda42cbd55a97b8d14050aa15c6e6450a1d63b0a641c8c8e5b81e50f957c87", "packages/opencode/src/mcp/index.ts": "da45165e6791a5b30c8962c7fd0ea391bb8e1931cc2357c6b90e9fd090cdeb0d", @@ -415,11 +418,15 @@ "packages/opencode/test/server/amicode-widgets.test.ts": "14fc6ccff4e83a5b005a724936e9f0d7ce411fab74109f319e8bc748665f7ab8", "packages/opencode/test/server/auth-accept-set.test.ts": "8fe698812ed9a809d21ace1a9751bd3f1976ab1e58ad5bda65dbfff58df8e3bc", "packages/opencode/test/server/gateway-secret.test.ts": "8d38ec187098be99f698c5e795b867a0e2fe5e2ef2aa3a9e2c03a8b0147d4473", + "packages/opencode/test/server/global-bus-sequencing.test.ts": "db0998c0f4715bc5d600b006c18c37133e5405e735853d9b31346501da31367f", + "packages/opencode/test/server/global-event-replay.test.ts": "c6c1fdba829c7b9a82976a90d71073086078d438f678c11f8577b21044cf3f56", "packages/opencode/test/server/httpapi-compression.test.ts": "5f7603d7018e0b95f1c2f36c01dd297c01770bfcdd91ce29778e03b7d6143f75", "packages/opencode/test/server/httpapi-event-replay.test.ts": "c94c13dbfb4ca5d5e9b80f4c1a17955d1530a8bf9141b9a83a810ee56743269f", + "packages/opencode/test/server/httpapi-event-sequencing.test.ts": "a2ec57f999b26b8ac9e4dfd6208c8028264cb0b267f4e6383429df62738967fa", "packages/opencode/test/server/httpapi-event.test.ts": "36263284302434461c4c6cf3e6652e094f4a02782acbe5de26c489748fe4cb50", "packages/opencode/test/server/httpapi-file-write-gate.test.ts": "0c91b7417e615850edfde3f2878207dab6b6624d4cfbb6e6783b2e6d9aa4dff0", "packages/opencode/test/server/httpapi-file.test.ts": "d3386129bc81467ecbcc57a3861f53767ee56ac41b4ff6ad58d8b168fb3b3373", + "packages/opencode/test/server/httpapi-global-event.test.ts": "6359f12508e6a34bfd9e0abe72b03113f933e1c83b9c55aae2a84c4bc5222bad", "packages/opencode/test/server/httpapi-instance-route-auth.test.ts": "4b73a89d2431c02bc3b665d93f9002914e51eb1add979dc1867efbcbc0c36e25", "packages/opencode/test/server/httpapi-provider.test.ts": "9e8b3e085b478e3f99b689ffff600e10c0b4b50b3cabdbd749f88dae19bf5904", "packages/opencode/test/server/httpapi-session.test.ts": "eda6fd763ac4c17891cedaa4ccd49597c03e7227fcfdcf0f94a1b4545ba623ed", @@ -842,6 +849,7 @@ "packages/opencode/src/server/amicode/widget-runtime.ts": "a5fe95eba46df87c4b4c54bc07bb91587b73e88eb90db889bc87358ece4b3f27", "packages/opencode/src/server/amicode/widgets.ts": "d15481086c41b365c28937f002bfcdeb8c2a9f68066413da695fbac8b7acdb61", "packages/opencode/src/server/shared/public-ui.ts": "a843b4e84673d9a5e34f36eb7cffbb499850f7588154c23061fc8a81cd1369ef", + "packages/opencode/src/server/shared/sse-bounded-writer.ts": "ba2b6cd332ae6a1aa8a0fe72af0f9faabccc28cc5ab5b66b45fa6bd7ee08310e", "packages/opencode/src/server/shared/sse-tick-telemetry.ts": "66ec419c2c32584c1f546b63fe93dbd7f7061d45be3dd37e9a261aa129027ec2", "packages/opencode/src/server/shared/ui.ts": "c66ef686ee090f442ee69b6a5f3c985ff005dc135b8265eaf5955f83ecfb86ea", "packages/opencode/src/server/shared/workspace-routing.ts": "3704b6cf218fc165ce9c1cdad0ee42fc9cf604976e4a552478200f85db80876a", @@ -970,9 +978,10 @@ "packages/opencode/src/server/routes/instance/httpapi/groups/mcp.ts": "048efa7b74d43c4ad4a75d0f8421c808760c8c0dcd593f0051a87044e40487d6", "packages/opencode/src/server/routes/instance/httpapi/groups/provider.ts": "34c6f1ab1a085d1b047c220bbff32a36032dc5e924784e9272cda166cf0ec84b", "packages/opencode/src/server/routes/instance/httpapi/groups/session.ts": "1421f8c07948b121c7e2005bd9b0fcd6b2c456565ddd1da32d8e6c35d03fac9c", - "packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "5824d237bedb868d97ca24f68da6b137b7c2d84e42111d7c17637619ce87c866", + "packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "494b471bd797a0b20abdb94417b1479c14460c33347800e7caa435c4959b8e41", "packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts": "69ca3dd48d95157fc822e1f76eb42ac873df50f22616b85194619ca658613867", "packages/opencode/src/server/routes/instance/httpapi/handlers/file.ts": "10bcb79b117beebee04184fb48c1ad789e782e2f7ae4b0c68603edc00f03aee9", + "packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts": "97f8e20d4f1425152a9719ed77af9c42163bbcd34f2cc7ae5ede991bf27430fd", "packages/opencode/src/server/routes/instance/httpapi/handlers/mcp.ts": "7da08d58765cdf2b6ce4648fc70de371402949f5bb7b75696f674d43f2d5365b", "packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts": "8baf16cad0992c5351b61b0c0fb27903bb25b17e1b25fc7c5d77f807c24413a6", "packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts": "b37ff4136c4aea85b23083bf441f03e09820e3e4cdd95e4118ca570d8435b94f", @@ -993,7 +1002,7 @@ "packages/ui/src/v2/components/text-shimmer-v2.tsx" ], "exceptions": [], - "extracted_at": "2026-10-15T07:32:24.625Z", + "extracted_at": "2026-10-15T12:25:06.525Z", "per_package": { "packages/app": { "A": 251, @@ -1011,8 +1020,8 @@ "D": 0 }, "packages/opencode": { - "A": 87, - "M": 94, + "A": 93, + "M": 97, "D": 0 }, "packages/schema": { @@ -1042,7 +1051,7 @@ } }, "counts": { - "overlay_total": 971, + "overlay_total": 980, "deletions": 7, "server_coupled": 0 }, @@ -1377,6 +1386,9 @@ "packages/llm/src/protocols/bedrock-converse.ts": "M", "packages/llm/test/provider/bedrock-converse.test.ts": "M", "packages/opencode/src/agent/agent.ts": "M", + "packages/opencode/src/bus/global-replay.ts": "A", + "packages/opencode/src/bus/global.ts": "M", + "packages/opencode/src/control-plane/workspace.ts": "M", "packages/opencode/src/effect/runner.ts": "M", "packages/opencode/src/mcp/browser.ts": "M", "packages/opencode/src/mcp/index.ts": "M", @@ -1454,11 +1466,15 @@ "packages/opencode/test/server/amicode-widgets.test.ts": "A", "packages/opencode/test/server/auth-accept-set.test.ts": "A", "packages/opencode/test/server/gateway-secret.test.ts": "A", + "packages/opencode/test/server/global-bus-sequencing.test.ts": "A", + "packages/opencode/test/server/global-event-replay.test.ts": "A", "packages/opencode/test/server/httpapi-compression.test.ts": "M", "packages/opencode/test/server/httpapi-event-replay.test.ts": "A", + "packages/opencode/test/server/httpapi-event-sequencing.test.ts": "A", "packages/opencode/test/server/httpapi-event.test.ts": "M", "packages/opencode/test/server/httpapi-file-write-gate.test.ts": "A", "packages/opencode/test/server/httpapi-file.test.ts": "M", + "packages/opencode/test/server/httpapi-global-event.test.ts": "A", "packages/opencode/test/server/httpapi-instance-route-auth.test.ts": "M", "packages/opencode/test/server/httpapi-provider.test.ts": "M", "packages/opencode/test/server/httpapi-session.test.ts": "M", @@ -1881,6 +1897,7 @@ "packages/opencode/src/server/amicode/widget-runtime.ts": "A", "packages/opencode/src/server/amicode/widgets.ts": "A", "packages/opencode/src/server/shared/public-ui.ts": "M", + "packages/opencode/src/server/shared/sse-bounded-writer.ts": "A", "packages/opencode/src/server/shared/sse-tick-telemetry.ts": "A", "packages/opencode/src/server/shared/ui.ts": "M", "packages/opencode/src/server/shared/workspace-routing.ts": "M", @@ -2012,6 +2029,7 @@ "packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/experimental.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/file.ts": "M", + "packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/mcp.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/provider.ts": "M", "packages/opencode/src/server/routes/instance/httpapi/handlers/session.ts": "M", diff --git a/packages/app-bundle/overlay/packages/opencode/src/bus/global-replay.ts b/packages/app-bundle/overlay/packages/opencode/src/bus/global-replay.ts new file mode 100644 index 00000000..7d49f376 --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/src/bus/global-replay.ts @@ -0,0 +1,277 @@ +// The global event stream's replay buffer (#1776, ported onto the lane for +// #1806 P1a). One tap on the global bus serializes each event once, at emit, +// into a frame stamped with its `:` id; the buffer keeps recent +// frames (one buffer shared by every subscriber, bounded by bytes and age) and +// fans each frame out to the live subscriptions. A connection that resumes +// from a servable cursor gets the frames after it, then live frames; any other +// cursor gets an explicit reset. +// +// Lane adaptations vs main (#1806): +// - a frame keeps a reference to the emitted event (`event`): this engine +// serves TWO wire shapes off the one buffer (the global route serializes +// `data` verbatim; the instance /event route projects the payload view from +// the live object), so subscribers never re-parse the wire serialization; +// - a cursor that is not `:` resolves against retained frames by +// payload id instead of resetting outright: the lane's client tracks +// payload ids (#1264), and wire compatibility requires its cursors to keep +// being served — resolved when the retained history covers them, reset as +// evicted when it does not; +// - main's patch-body wire cut (#1778's wireEvent, 06eac851) is P3 and is +// deliberately NOT ported — the serialized frame is the whole event. +import { GlobalBus, type GlobalEvent } from "./global" + +export type Frame = { + /** The SSE id: `:`. */ + readonly id: string + readonly seq: number + /** The serialized global event — the SSE `data`. */ + readonly data: string + /** The emitted event itself — the buffer's in-process view (see header). */ + readonly event: GlobalEvent + /** UTF-8 size of `data`, charged against the byte bound. */ + readonly bytes: number + /** Push time (ms), checked against the age bound. */ + readonly time: number + /** Push order; unlike `seq` it is unique per frame (a sync copy shares its source's seq). */ + readonly ordinal: number + /** + * A sync copy sharing the sequence of the frame it mirrors (#1778). Sent only + * to subscribers that request sync copies; dropping it leaves no hole in a + * subscriber's sequence, because its source carries the same one. + */ + readonly syncCopy: boolean +} + +export type Bounds = { + /** Retained frame bytes; the oldest frames are evicted past it. */ + readonly maxBytes: number + /** Retained frame age in milliseconds; older frames are evicted. */ + readonly maxAgeMs: number +} + +/** About 32 MB and 5 minutes of history: covers link blips and window reloads. */ +export const DEFAULT_BOUNDS: Bounds = { maxBytes: 32 * 1024 * 1024, maxAgeMs: 5 * 60_000 } + +export const BOUNDS_ENV = { + maxBytes: "OPENCODE_GLOBAL_EVENT_REPLAY_MAX_BYTES", + maxAgeMs: "OPENCODE_GLOBAL_EVENT_REPLAY_MAX_AGE_MS", +} as const + +/** The buffer bounds from the environment; an unset or invalid value keeps its default. */ +export function boundsFromEnv(env: Record): Bounds { + const read = (key: string, fallback: number) => { + const raw = env[key]?.trim() + if (!raw || !/^\d+$/.test(raw)) return fallback + return Number(raw) + } + return { + maxBytes: read(BOUNDS_ENV.maxBytes, DEFAULT_BOUNDS.maxBytes), + maxAgeMs: read(BOUNDS_ENV.maxAgeMs, DEFAULT_BOUNDS.maxAgeMs), + } +} + +export function formatID(epoch: string, seq: number) { + return `${epoch}:${seq}` +} + +/** Parse an `:` cursor; `undefined` when malformed. */ +export function parseCursor(raw: string): { epoch: string; seq: number } | undefined { + const match = /^([A-Za-z0-9_-]+):(\d+)$/.exec(raw.trim()) + if (!match) return undefined + return { epoch: match[1], seq: Number(match[2]) } +} + +/** Why a cursor cannot be served. */ +export type ResetReason = "malformed" | "epoch" | "ahead" | "evicted" + +export type Subscription = { + /** The connection's starting position: the sequence its first frame follows. */ + readonly position: number + /** Set when the cursor could not be served: the connection must start with a reset frame. */ + readonly reset: ResetReason | undefined + /** Frames after the cursor, in order — delivered before any live frame. */ + readonly replay: readonly Frame[] + /** Begin delivering live frames (any that arrived since subscribing first, minus duplicates of `replay`). */ + start(deliver: (frame: Frame) => void): void + /** Unregister. Idempotent. */ + close(): void +} + +export class ReplayBuffer { + readonly #listeners = new Set<(frame: Frame) => void>() + // Retained frames are #frames[#head..]; evicted slots before #head are + // compacted away in batches so eviction stays O(1) amortized. + #frames: (Frame | undefined)[] = [] + #head = 0 + #bytes = 0 + #ordinal = 0 + #current: number + // The highest sequence whose frames are no longer all retained (evicted, or + // emitted before this buffer existed). A cursor below it cannot be served. + #floor: number + + readonly epoch: string + readonly bounds: Bounds + readonly #now: () => number + + constructor(input: { + epoch: string + bounds?: Bounds + /** The sequence the buffer starts at (events before it were never recorded). */ + start?: number + now?: () => number + }) { + this.epoch = input.epoch + this.bounds = input.bounds ?? DEFAULT_BOUNDS + this.#now = input.now ?? Date.now + this.#current = input.start ?? 0 + this.#floor = this.#current + } + + /** The current sequence: the last one pushed (or skipped). */ + get current() { + return this.#current + } + + /** Live subscriptions currently registered. */ + get subscribers() { + return this.#listeners.size + } + + push(seq: number, event: GlobalEvent, data: string, options: { syncCopy?: boolean } = {}): Frame { + const frame: Frame = { + id: formatID(this.epoch, seq), + seq, + data, + event, + bytes: Buffer.byteLength(data), + time: this.#now(), + ordinal: ++this.#ordinal, + syncCopy: options.syncCopy === true, + } + this.#current = Math.max(this.#current, seq) + this.#frames.push(frame) + this.#bytes += frame.bytes + for (const listener of this.#listeners) listener(frame) + this.#evict() + return frame + } + + /** Advance past a sequence that produced no frame (the hole stays visible to clients). */ + skip(seq: number) { + this.#current = Math.max(this.#current, seq) + } + + /** + * Subscribe, optionally resuming after `cursor`. The live listener is + * registered BEFORE the buffer is read, and live frames already covered by + * the replay are dropped, so nothing published around the reconnect is + * missed or delivered twice. Sync copies reach the subscription only when it + * asks for them (`sync`), in its replay and its live frames alike. + */ + subscribe(cursor?: string, options: { sync?: boolean } = {}): Subscription { + const wanted = options.sync === true ? () => true : (frame: Frame) => !frame.syncCopy + let deliver: ((frame: Frame) => void) | undefined + let pending: Frame[] = [] + const listener = (frame: Frame) => { + if (!wanted(frame)) return + if (deliver) deliver(frame) + else pending.push(frame) + } + this.#listeners.add(listener) + + this.#evict() + const watermark = this.#ordinal + const resolved = this.#resolve(cursor) + const replay = resolved.reset + ? [] + : this.#retained().filter((frame) => frame.seq > resolved.position && wanted(frame)) + + return { + ...resolved, + replay, + start: (fn) => { + const queued = pending + pending = [] + for (const frame of queued) if (frame.ordinal > watermark) fn(frame) + deliver = fn + }, + close: () => { + this.#listeners.delete(listener) + pending = [] + }, + } + } + + #resolve(cursor: string | undefined): { position: number; reset: ResetReason | undefined } { + if (cursor === undefined) return { position: this.#current, reset: undefined } + const parsed = parseCursor(cursor) + if (parsed) { + const reset: ResetReason | undefined = + parsed.epoch !== this.epoch + ? "epoch" + : parsed.seq > this.#current + ? "ahead" + : parsed.seq < this.#floor + ? "evicted" + : undefined + return reset ? { position: this.#current, reset } : { position: parsed.seq, reset: undefined } + } + // Legacy vocabulary (the lane's client sends the last payload id it + // applied, #1264): resolve to the newest retained frame at or behind the + // cursor and serve the frames after it; a cursor no retained frame covers + // predates retention and resets. + let resolved: Frame | undefined + for (const frame of this.#retained()) { + const id = frame.event.payload?.id + if (typeof id === "string" && id <= cursor) resolved = frame + } + if (resolved) return { position: resolved.seq, reset: undefined } + return { position: this.#current, reset: "evicted" } + } + + #retained(): Frame[] { + return this.#frames.slice(this.#head) as Frame[] + } + + // Oldest-first, until retained bytes and the oldest frame's age are both within bounds. + #evict() { + const oldest = this.#now() - this.bounds.maxAgeMs + while (this.#head < this.#frames.length) { + const frame = this.#frames[this.#head]! + if (this.#bytes <= this.bounds.maxBytes && frame.time >= oldest) break + this.#frames[this.#head++] = undefined + this.#bytes -= frame.bytes + this.#floor = Math.max(this.#floor, frame.seq) + } + if (this.#head > 1024 && this.#head * 2 > this.#frames.length) { + this.#frames = this.#frames.slice(this.#head) + this.#head = 0 + } + } +} + +export const GlobalEventReplay = new ReplayBuffer({ + epoch: GlobalBus.epoch, + bounds: boundsFromEnv(process.env), + start: GlobalBus.seq, +}) + +GlobalBus.tap((event, seq) => { + // A sync copy whose sequence was already handed out mirrors its source. One + // that consumed a fresh sequence (no source framed nearby) is a frame of its + // own and reaches every subscriber, so filtering never opens a hole. + const syncCopy = event.payload?.type === "sync" && seq <= GlobalEventReplay.current + let data: string + try { + // serialized once: every subscriber, live or replayed, gets this same frame + data = JSON.stringify(event) + } catch (error) { + // An unserializable event cannot be framed; its sequence becomes a hole the + // client detects (heartbeats carry the current sequence), never a dead stream. + console.error("global event not serializable", { seq, type: event.payload?.type, error }) + GlobalEventReplay.skip(seq) + return + } + GlobalEventReplay.push(seq, event, data, { syncCopy }) +}) diff --git a/packages/app-bundle/overlay/packages/opencode/src/bus/global.ts b/packages/app-bundle/overlay/packages/opencode/src/bus/global.ts new file mode 100644 index 00000000..2e6ee4e2 --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/src/bus/global.ts @@ -0,0 +1,76 @@ +import { EventEmitter } from "events" +import { randomBytes } from "crypto" +import { Identifier } from "@/id/id" + +export type GlobalEvent = { + directory?: string + project?: string + workspace?: string + payload: any +} + +/** + * A pre-dispatch observer. Taps see every emitted event with its sequence + * BEFORE any ordinary listener runs, so they observe events in strict + * sequence order even when a listener emits from inside its handler (a nested + * emit is sequenced and tapped only after the outer event was). A tap must not + * throw and must not emit. + */ +export type GlobalBusTap = (event: GlobalEvent, seq: number) => void + +// How many recent source-event ids are remembered so a sync copy (which the +// event-v2 bridge emits right after its source) can share the source's +// sequence. The copy follows its source immediately; the window only has to +// absorb events a listener emits in between. +const MIRROR_WINDOW = 256 + +class GlobalBusEmitter extends EventEmitter<{ + event: [GlobalEvent] +}> { + /** Random per-process token; `:` is a global-stream frame id (#1776). */ + readonly epoch = randomBytes(6).toString("base64url") + #seq = 0 + #mirrors = new Map() + #taps = new Set() + + /** The current sequence: the one assigned to the last emitted event (0 before any). */ + get seq() { + return this.#seq + } + + /** Register a pre-dispatch tap; returns its unregister function. */ + tap(fn: GlobalBusTap): () => void { + this.#taps.add(fn) + return () => { + this.#taps.delete(fn) + } + } + + override emit(eventName: "event", event: GlobalEvent): boolean { + if (event.payload && typeof event.payload === "object" && !("id" in event.payload)) { + event.payload.id = event.payload.syncEvent?.id ?? Identifier.create("evt", "ascending") + } + const seq = this.#sequence(event.payload) + for (const tap of this.#taps) tap(event, seq) + return super.emit(eventName, event) + } + + // One strictly increasing counter, assigned once per event at emit and shared + // by every subscriber. A sync copy is not a new event: it carries the + // sequence of the event it mirrors, so filtering sync copies leaves no holes. + #sequence(payload: any): number { + if (payload?.type === "sync") { + const source = typeof payload.syncEvent?.id === "string" ? this.#mirrors.get(payload.syncEvent.id) : undefined + if (source !== undefined) return source + return ++this.#seq + } + const seq = ++this.#seq + if (typeof payload?.id === "string") { + this.#mirrors.set(payload.id, seq) + if (this.#mirrors.size > MIRROR_WINDOW) this.#mirrors.delete(this.#mirrors.keys().next().value!) + } + return seq + } +} + +export const GlobalBus = new GlobalBusEmitter() diff --git a/packages/app-bundle/overlay/packages/opencode/src/control-plane/workspace.ts b/packages/app-bundle/overlay/packages/opencode/src/control-plane/workspace.ts new file mode 100644 index 00000000..a4674f79 --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/src/control-plane/workspace.ts @@ -0,0 +1,972 @@ +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { httpClient } from "@opencode-ai/core/effect/app-node-platform" +import { Context, Effect, FiberMap, Iterable, Layer, Schema, Stream } from "effect" +import { serviceUse } from "@opencode-ai/core/effect/service-use" +import { FetchHttpClient, HttpBody, HttpClient, HttpClientError, HttpClientRequest } from "effect/unstable/http" +import { Database } from "@opencode-ai/core/database/database" +import { asc } from "drizzle-orm" +import { eq } from "drizzle-orm" +import { inArray } from "drizzle-orm" +import { Project } from "@/project/project" +import { GlobalBus } from "@/bus/global" +import { Auth } from "@/auth" +import { EventV2 } from "@opencode-ai/core/event" +import { EventV2Bridge } from "@/event-v2-bridge" +import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql" +import { FSUtil } from "@opencode-ai/core/fs-util" +import { RuntimeFlags } from "@/effect/runtime-flags" +import { ProjectV2 } from "@opencode-ai/core/project" +import { Slug } from "@opencode-ai/core/util/slug" +import { WorkspaceTable } from "@opencode-ai/core/control-plane/workspace.sql" +import { getAdapter, registeredAdapters } from "./adapters" +import { type Target, type WorkspaceInfo, WorkspaceInfo as WorkspaceInfoSchema } from "./types" +import { WorkspaceV2 } from "@opencode-ai/core/workspace" +import { Session } from "@/session/session" +import { SessionPrompt } from "@/session/prompt" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { SessionID } from "@/session/schema" +import { NotFoundError } from "@/storage/storage" +import { errorData } from "@/util/error" +import { waitEvent } from "./util" +import { WorkspaceRef } from "@/effect/instance-ref" +import { Vcs } from "@/project/vcs" +import { InstanceStore } from "@/project/instance-store" +import { WorkspaceAdapterRuntime } from "./workspace-adapter-runtime" +import { AppNodeBuilderV1 } from "@/effect/app-node-builder-v1" +import { WorkspaceEvent } from "@opencode-ai/schema/workspace-event" + +export const Info = Schema.Struct({ + ...WorkspaceInfoSchema.fields, + timeUsed: Schema.Number, +}).annotate({ identifier: "Workspace" }) +export type Info = WorkspaceInfo & { timeUsed: number } + +export const ConnectionStatus = WorkspaceEvent.ConnectionStatus +export type ConnectionStatus = WorkspaceEvent.ConnectionStatus + +export const Event = WorkspaceEvent + +function fromRow(row: typeof WorkspaceTable.$inferSelect): Info { + return { + id: row.id, + type: row.type, + branch: row.branch, + name: row.name, + directory: row.directory, + extra: row.extra, + projectID: row.project_id, + timeUsed: row.time_used, + } +} + +export const CreateInput = Schema.Struct({ + id: Schema.optional(WorkspaceV2.ID), + type: Info.fields.type, + branch: Info.fields.branch, + projectID: ProjectV2.ID, + extra: Schema.optional(Info.fields.extra), +}) +export type CreateInput = Schema.Schema.Type + +export const SessionWarpInput = Schema.Struct({ + workspaceID: Schema.NullOr(WorkspaceV2.ID), + sessionID: SessionID, + copyChanges: Schema.optional(Schema.Boolean), +}) +export type SessionWarpInput = Schema.Schema.Type + +export class SyncHttpError extends Schema.TaggedErrorClass()("WorkspaceSyncHttpError", { + message: Schema.String, + status: Schema.Number, + body: Schema.optional(Schema.String), +}) {} + +export class WorkspaceNotFoundError extends Schema.TaggedErrorClass()( + "WorkspaceNotFoundError", + { + message: Schema.String, + workspaceID: WorkspaceV2.ID, + }, +) {} + +export class SessionEventsNotFoundError extends Schema.TaggedErrorClass()( + "WorkspaceSessionEventsNotFoundError", + { + message: Schema.String, + sessionID: SessionID, + }, +) {} + +export class SessionWarpHttpError extends Schema.TaggedErrorClass()( + "WorkspaceSessionWarpHttpError", + { + message: Schema.String, + workspaceID: WorkspaceV2.ID, + sessionID: SessionID, + status: Schema.Number, + body: Schema.String, + }, +) {} + +export class SyncTimeoutError extends Schema.TaggedErrorClass()("WorkspaceSyncTimeoutError", { + message: Schema.String, + state: Schema.Record(Schema.String, Schema.Number), +}) {} + +export class SyncAbortedError extends Schema.TaggedErrorClass()("WorkspaceSyncAbortedError", { + message: Schema.String, + cause: Schema.optional(Schema.Defect()), +}) {} + +type CreateError = Auth.AuthError +type SessionWarpError = + | WorkspaceNotFoundError + | SessionEventsNotFoundError + | SessionWarpHttpError + | Vcs.PatchApplyError + | HttpClientError.HttpClientError +type WaitForSyncError = SyncTimeoutError | SyncAbortedError +type SyncLoopError = SyncHttpError | HttpClientError.HttpClientError + +export interface Interface { + readonly create: (input: CreateInput) => Effect.Effect + readonly sessionWarp: (input: SessionWarpInput) => Effect.Effect + readonly list: (project: Project.Info) => Effect.Effect + readonly syncList: (project: Project.Info) => Effect.Effect + readonly get: (id: WorkspaceV2.ID) => Effect.Effect + readonly remove: (id: WorkspaceV2.ID) => Effect.Effect + readonly status: () => Effect.Effect + readonly isSyncing: (workspaceID: WorkspaceV2.ID) => Effect.Effect + readonly waitForSync: ( + workspaceID: WorkspaceV2.ID, + state: Record, + signal?: AbortSignal, + timeout?: number, + ) => Effect.Effect + readonly startWorkspaceSyncing: (projectID: ProjectV2.ID) => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/Workspace") {} + +export const use = serviceUse(Service) + +const layer = Layer.effect( + Service, + Effect.gen(function* () { + const auth = yield* Auth.Service + const session = yield* Session.Service + const prompt = yield* SessionPrompt.Service + const http = yield* HttpClient.HttpClient + const events = yield* EventV2Bridge.Service + const vcs = yield* Vcs.Service + const flags = yield* RuntimeFlags.Service + const fs = yield* FSUtil.Service + const { db } = yield* Database.Service + const connections = new Map() + const syncFibers = yield* FiberMap.make() + + const setStatus = (id: WorkspaceV2.ID, status: ConnectionStatus["status"]) => { + const prev = connections.get(id) + if (prev?.status === status) return + const next = { workspaceID: id, status } + connections.set(id, next) + + GlobalBus.emit("event", { + directory: "global", + workspace: id, + payload: { + type: Event.Status.type, + properties: next, + }, + }) + } + + const connectSSE = Effect.fn("Workspace.connectSSE")(function* ( + url: URL | string, + headers: HeadersInit | undefined, + ) { + // Sync copies are opt-in per global-stream subscription (#1778, #1806 + // P1a): this loop is the subscriber that replays them (events.replay + // below is the whole point of the connection), so it must ask for them + // — the header the ported engine reads. + const syncHeaders = new Headers(headers) + syncHeaders.set("x-opencode-sync-copies", "true") + const response = yield* http.execute( + HttpClientRequest.get(route(url, "/global/event"), { + headers: syncHeaders, + accept: "text/event-stream", + }), + ) + if (response.status < 200 || response.status >= 300) { + return yield* new SyncHttpError({ + message: `Workspace sync HTTP failure: ${response.status}`, + status: response.status, + }) + } + return response.stream + }) + + const parseSSE = Effect.fn("Workspace.parseSSE")(function* ( + stream: Stream.Stream, + onEvent: (event: unknown) => Effect.Effect, + ) { + yield* stream.pipe( + Stream.decodeText(), + Stream.splitLines, + Stream.mapAccum( + () => ({ data: [] as string[], id: undefined as string | undefined, retry: 1000 }), + (state, line) => { + if (line === "") { + if (!state.data.length) return [state, []] + return [{ ...state, data: [] }, [{ data: state.data.join("\n"), id: state.id, retry: state.retry }]] + } + + const index = line.indexOf(":") + const field = index === -1 ? line : line.slice(0, index) + const value = index === -1 ? "" : line.slice(index + (line[index + 1] === " " ? 2 : 1)) + + if (field === "data") return [{ ...state, data: [...state.data, value] }, []] + if (field === "id") return [{ ...state, id: value }, []] + if (field === "retry") { + const retry = Number.parseInt(value, 10) + return [Number.isNaN(retry) ? state : { ...state, retry }, []] + } + return [state, []] + }, + { + onHalt: (state) => + state.data.length ? [{ data: state.data.join("\n"), id: state.id, retry: state.retry }] : [], + }, + ), + Stream.map((event) => { + try { + return JSON.parse(event.data) as unknown + } catch { + return { + type: "sse.message", + properties: { + data: event.data, + id: event.id || undefined, + retry: event.retry, + }, + } + } + }), + Stream.runForEach(onEvent), + ) + }) + + const runInWorkspace = (input: { + workspaceID?: WorkspaceV2.ID + local: () => Effect.Effect + remote: (input: { + workspace: Info + target: Extract + }) => HttpClientRequest.HttpClientRequest + fallback: A + response?: "json" | "text" + }) => + Effect.gen(function* () { + if (!input.workspaceID) return yield* input.local() + + const workspace = yield* get(input.workspaceID) + if (!workspace) return input.fallback + + const target = yield* WorkspaceAdapterRuntime.target(workspace) + + if (target.type === "local") { + const store = yield* InstanceStore.Service + return yield* store.provide({ directory: target.directory }, input.local()) + } + + const response = yield* http.execute(input.remote({ workspace, target })).pipe( + Effect.catch((error) => + Effect.logWarning("workspace target request failed", { + workspaceID: workspace.id, + error: errorData(error), + }).pipe(Effect.as(undefined)), + ), + ) + if (!response) return input.fallback + if (response.status < 200 || response.status >= 300) { + const body = yield* response.text.pipe(Effect.catch(() => Effect.succeed(""))) + yield* Effect.logWarning("workspace target request failed", { + workspaceID: workspace.id, + status: response.status, + body, + }) + return input.fallback + } + + const body = input.response === "text" ? response.text : response.json + return yield* body.pipe( + Effect.map((result) => result as A), + Effect.catch((error) => + Effect.logWarning("workspace target response decode failed", { + workspaceID: workspace.id, + error: errorData(error), + }).pipe(Effect.as(input.fallback)), + ), + ) + }) + + const syncHistory = Effect.fn("Workspace.syncHistory")(function* ( + space: Info, + url: URL | string, + headers: HeadersInit | undefined, + ) { + const sessionIDs = (yield* db + .select({ id: SessionTable.id }) + .from(SessionTable) + .where(eq(SessionTable.workspace_id, space.id)) + .all() + .pipe(Effect.orDie)).map((row) => row.id) + const state = sessionIDs.length + ? Object.fromEntries( + (yield* db + .select() + .from(EventSequenceTable) + .where(inArray(EventSequenceTable.aggregate_id, sessionIDs)) + .all() + .pipe(Effect.orDie)).map((row) => [row.aggregate_id, row.seq]), + ) + : {} + + const response = yield* http.execute( + HttpClientRequest.post(route(url, "/sync/history"), { + headers: new Headers(headers), + body: HttpBody.jsonUnsafe(state), + }), + ) + + if (response.status < 200 || response.status >= 300) { + const body = yield* response.text + return yield* new SyncHttpError({ + message: `Workspace history HTTP failure: ${response.status} ${body}`, + status: response.status, + body, + }) + } + + const history = (yield* response.json) as HistoryEvent[] + + yield* Effect.forEach( + history, + (event) => + events + .replay( + { + id: EventV2.ID.make(event.id), + aggregateID: event.aggregate_id, + seq: event.seq, + type: event.type, + data: event.data, + }, + { publish: true, ownerID: space.id }, + ) + .pipe(Effect.provideService(WorkspaceRef, space.id)), + { discard: true }, + ) + }) + + const syncWorkspaceLoop = Effect.fn("Workspace.syncWorkspaceLoop")(function* (space: Info) { + const target = yield* WorkspaceAdapterRuntime.target(space) + + if (target.type === "local") return + + let attempt = 0 + + while (true) { + setStatus(space.id, "connecting") + + const stream = yield* connectSSE(target.url, target.headers).pipe( + Effect.tap(() => syncHistory(space, target.url, target.headers)), + Effect.catch((err) => + Effect.gen(function* () { + setStatus(space.id, "error") + yield* Effect.logWarning("failed to connect to global sync", { + workspace: space.name, + error: errorData(err), + }) + return null + }), + ), + ) + + if (stream) { + attempt = 0 + + setStatus(space.id, "connected") + + yield* parseSSE(stream, (evt) => + Effect.gen(function* () { + if (!evt || typeof evt !== "object" || !("payload" in evt)) return + const payload = evt.payload as { type?: string; syncEvent?: EventV2.SerializedEvent } + if (payload.type === "server.heartbeat") return + + if (payload.type === "sync" && payload.syncEvent) { + const failed = yield* events.replay(payload.syncEvent, { publish: true, ownerID: space.id }).pipe( + Effect.as(false), + Effect.catchCause((error) => + Effect.logWarning("failed to replay global event", error).pipe( + Effect.annotateLogs({ workspaceID: space.id }), + Effect.as(true), + ), + ), + ) + if (failed) return + } + + try { + const event = evt as { directory?: string; project?: string; payload: unknown } + GlobalBus.emit("event", { + directory: event.directory, + project: event.project, + workspace: space.id, + payload: event.payload, + }) + } catch (error) { + yield* Effect.logWarning("failed to emit global event", { + workspaceID: space.id, + error: errorData(error), + }) + } + }), + ) + + setStatus(space.id, "disconnected") + } + + // Back off reconnect attempts up to 2 minutes while the workspace + // stays unavailable. + yield* Effect.sleep(`${Math.min(120_000, 1_000 * 2 ** attempt)} millis`) + attempt += 1 + } + }) + + const startSync = Effect.fn("Workspace.startSync")(function* (space: Info) { + if (!flags.experimentalWorkspaces) return + + const target = yield* WorkspaceAdapterRuntime.target(space).pipe( + Effect.catch((error) => + Effect.gen(function* () { + setStatus(space.id, "error") + yield* Effect.logWarning("workspace target failed", { + workspaceID: space.id, + error: errorData(error), + }) + return null + }), + ), + ) + if (!target) return + + if (target.type === "local") { + setStatus(space.id, (yield* fs.existsSafe(target.directory)) ? "connected" : "error") + return + } + + const exists = yield* FiberMap.has(syncFibers, space.id) + if (exists && connections.get(space.id)?.status !== "error") return + + setStatus(space.id, "disconnected") + + yield* FiberMap.run( + syncFibers, + space.id, + // TODO: look into `tapError` to set the status but still + // allow the fiber to fail and automatically get removed + syncWorkspaceLoop(space).pipe( + Effect.catch((error) => + Effect.gen(function* () { + setStatus(space.id, "error") + yield* Effect.logWarning("workspace listener failed", { + workspaceID: space.id, + error: errorData(error), + }) + }), + ), + ), + ) + }) + + const stopSync = Effect.fn("Workspace.stopSync")(function* (id: WorkspaceV2.ID) { + yield* FiberMap.remove(syncFibers, id) + connections.delete(id) + }) + + const create = Effect.fn("Workspace.create")(function* (input: CreateInput) { + const id = WorkspaceV2.ID.ascending(input.id) + const adapter = getAdapter(input.projectID, input.type) + const config = yield* WorkspaceAdapterRuntime.configure(adapter, { + ...input, + id, + name: Slug.create(), + directory: null, + extra: input.extra ?? null, + }) + + const info: Info = { + id, + type: config.type, + branch: config.branch ?? null, + name: config.name ?? null, + directory: config.directory ?? null, + extra: config.extra ?? null, + projectID: input.projectID, + timeUsed: Date.now(), + } + + yield* db + .insert(WorkspaceTable) + .values({ + id: info.id, + type: info.type, + branch: info.branch, + name: info.name, + directory: info.directory, + extra: info.extra, + project_id: info.projectID, + time_used: info.timeUsed, + }) + .run() + .pipe(Effect.orDie) + + const env = { + OPENCODE_AUTH_CONTENT: JSON.stringify(yield* auth.all()), + OPENCODE_WORKSPACE_ID: config.id, + OPENCODE_EXPERIMENTAL_WORKSPACES: "true", + OTEL_EXPORTER_OTLP_HEADERS: process.env.OTEL_EXPORTER_OTLP_HEADERS, + OTEL_EXPORTER_OTLP_ENDPOINT: process.env.OTEL_EXPORTER_OTLP_ENDPOINT, + OTEL_RESOURCE_ATTRIBUTES: process.env.OTEL_RESOURCE_ATTRIBUTES, + } + + yield* WorkspaceAdapterRuntime.create(adapter, config, env) + yield* Effect.all( + [ + waitEvent({ + timeout: TIMEOUT, + fn(event) { + if (event.workspace === info.id && event.payload.type === Event.Status.type) { + const { status } = event.payload.properties + return status === "error" || status === "connected" + } + return false + }, + }), + startSync(info), + ], + { concurrency: 2, discard: true }, + ) + + return info + }) + + const sessionWarp = Effect.fn("Workspace.sessionWarp")(function* (input: SessionWarpInput) { + return yield* Effect.gen(function* () { + const current = yield* db + .select({ workspaceID: SessionTable.workspace_id }) + .from(SessionTable) + .where(eq(SessionTable.id, input.sessionID)) + .get() + .pipe(Effect.orDie) + + if (current?.workspaceID) { + const previous = yield* get(current.workspaceID) + if (previous) { + const target = yield* WorkspaceAdapterRuntime.target(previous) + + if (target.type === "remote") { + yield* syncHistory(previous, target.url, target.headers).pipe( + Effect.catch((error) => + Effect.logWarning("session warp final source sync failed", { + workspaceID: previous.id, + sessionID: input.sessionID, + error: errorData(error), + }), + ), + ) + } else { + yield* prompt.cancel(input.sessionID) + } + + // "claim" this session so any future events coming from + // the old workspace are ignored + yield* events.claim(input.sessionID, input.workspaceID ?? previous.projectID) + } + } + + const sourcePatch = + input.copyChanges && current?.workspaceID + ? yield* runInWorkspace({ + workspaceID: current?.workspaceID ?? undefined, + local: () => vcs.diffRaw(), + remote: ({ target }) => + HttpClientRequest.get(route(target.url, "/vcs/diff/raw"), { + headers: new Headers(target.headers), + }), + fallback: "", + response: "text", + }).pipe(Effect.provide(AppNodeBuilderV1.build(InstanceStore.node))) + : "" + + if (sourcePatch) { + // Attempt to apply the file changes to the new workspace. + // We intentionally do first so if it fails we don't warp + // the session. + yield* runInWorkspace({ + workspaceID: input.workspaceID ?? undefined, + local: () => vcs.apply({ patch: sourcePatch }), + remote: ({ target }) => + HttpClientRequest.post(route(target.url, "/vcs/apply"), { + headers: new Headers(target.headers), + body: HttpBody.jsonUnsafe({ patch: sourcePatch }), + }), + fallback: { applied: false }, + }).pipe(Effect.provide(AppNodeBuilderV1.build(InstanceStore.node))) + } + + if (input.workspaceID === null) { + yield* session.setWorkspace({ sessionID: input.sessionID, workspaceID: undefined }) + + return + } + + const workspaceID = input.workspaceID + const space = yield* get(workspaceID) + if (!space) + return yield* new WorkspaceNotFoundError({ + message: `Workspace not found: ${workspaceID}`, + workspaceID, + }) + + const target = yield* WorkspaceAdapterRuntime.target(space) + + if (target.type === "local") { + yield* session.setWorkspace({ sessionID: input.sessionID, workspaceID: input.workspaceID }) + + return + } + + const rows = yield* db + .select({ + id: EventTable.id, + aggregateID: EventTable.aggregate_id, + seq: EventTable.seq, + type: EventTable.type, + data: EventTable.data, + }) + .from(EventTable) + .where(eq(EventTable.aggregate_id, input.sessionID)) + .orderBy(asc(EventTable.seq)) + .all() + .pipe(Effect.orDie) + if (rows.length === 0) + return yield* new SessionEventsNotFoundError({ + message: `No events found for session: ${input.sessionID}`, + sessionID: input.sessionID, + }) + + const batches = Iterable.chunksOf(rows, 10) + const total = Iterable.size(batches) + + yield* Effect.forEach( + batches, + (events, i) => + Effect.gen(function* () { + const response = yield* http.execute( + HttpClientRequest.post(route(target.url, "/sync/replay"), { + headers: new Headers(target.headers), + body: HttpBody.jsonUnsafe({ + directory: space.directory ?? "", + events, + }), + }), + ) + + if (response.status < 200 || response.status >= 300) { + const body = yield* response.text + return yield* new SessionWarpHttpError({ + message: `Failed to warp session ${input.sessionID} into workspace ${workspaceID}: HTTP ${response.status} ${body}`, + workspaceID, + sessionID: input.sessionID, + status: response.status, + body, + }) + } + }), + { discard: true }, + ) + + const response = yield* http.execute( + HttpClientRequest.post(route(target.url, "/sync/steal"), { + headers: new Headers(target.headers), + body: HttpBody.jsonUnsafe({ sessionID: input.sessionID }), + }), + ) + if (response.status < 200 || response.status >= 300) { + const body = yield* response.text + return yield* new SessionWarpHttpError({ + message: `Failed to steal session ${input.sessionID} into workspace ${workspaceID}: HTTP ${response.status} ${body}`, + workspaceID, + sessionID: input.sessionID, + status: response.status, + body, + }) + } + + yield* session.setWorkspace({ sessionID: input.sessionID, workspaceID: input.workspaceID }) + }) + }) + + const list = Effect.fn("Workspace.list")(function* (project: Project.Info) { + if (!flags.experimentalWorkspaces) return [] + return (yield* db + .select() + .from(WorkspaceTable) + .where(eq(WorkspaceTable.project_id, project.id)) + .all() + .pipe(Effect.orDie)) + .map(fromRow) + .sort((a, b) => a.id.localeCompare(b.id)) + }) + + const syncList = Effect.fn("Workspace.syncList")(function* (project: Project.Info) { + const names = new Set((yield* list(project)).map((workspace) => workspace.name)) + const discovered = yield* Effect.forEach( + registeredAdapters(project.id), + ([type, adapter]) => + WorkspaceAdapterRuntime.list(adapter).pipe( + Effect.catchCause((error) => + Effect.logWarning("workspace adapter list failed", { type, error }).pipe(Effect.as([])), + ), + ), + { concurrency: "unbounded" }, + ).pipe(Effect.map((items) => items.flat())) + + yield* Effect.forEach( + discovered, + (item) => + Effect.gen(function* () { + if (names.has(item.name)) return + names.add(item.name) + + const info: Info = { + id: WorkspaceV2.ID.ascending(), + type: item.type, + branch: item.branch, + name: item.name, + directory: item.directory, + extra: item.extra, + projectID: item.projectID, + timeUsed: Date.now(), + } + + yield* db + .insert(WorkspaceTable) + .values({ + id: info.id, + type: info.type, + branch: info.branch, + name: info.name, + directory: info.directory, + extra: info.extra, + project_id: info.projectID, + time_used: info.timeUsed, + }) + .run() + .pipe(Effect.orDie) + + yield* startSync(info) + }), + { concurrency: 1 }, + ) + }) + + const get = Effect.fn("Workspace.get")(function* (id: WorkspaceV2.ID) { + const row = yield* db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, id)).get().pipe(Effect.orDie) + if (!row) return + return fromRow(row) + }) + + const remove = Effect.fn("Workspace.remove")(function* (id: WorkspaceV2.ID) { + const sessions = yield* db + .select({ id: SessionTable.id, parentID: SessionTable.parent_id }) + .from(SessionTable) + .where(eq(SessionTable.workspace_id, id)) + .all() + .pipe(Effect.orDie) + const sessionIDs = new Set(sessions.map((sessionInfo) => sessionInfo.id)) + yield* Effect.forEach( + sessions.filter((sessionInfo) => !sessionInfo.parentID || !sessionIDs.has(sessionInfo.parentID)), + (sessionInfo) => + session.remove(sessionInfo.id).pipe(Effect.catchIf(NotFoundError.isInstance, () => Effect.void)), + { discard: true }, + ) + + const row = yield* db.select().from(WorkspaceTable).where(eq(WorkspaceTable.id, id)).get().pipe(Effect.orDie) + if (!row) return + + yield* stopSync(id) + + const info = fromRow(row) + yield* Effect.catchCause( + Effect.gen(function* () { + yield* WorkspaceAdapterRuntime.remove(info) + }), + () => Effect.logError("adapter not available when removing workspace", { type: row.type }), + ) + + yield* db.delete(WorkspaceTable).where(eq(WorkspaceTable.id, id)).run().pipe(Effect.orDie) + return info + }) + + const status = Effect.fn("Workspace.status")(function* () { + return [...connections.values()] + }) + + const isSyncing = Effect.fn("Workspace.isSyncing")(function* (workspaceID: WorkspaceV2.ID) { + const exists = yield* FiberMap.has(syncFibers, workspaceID) + return exists && connections.get(workspaceID)?.status !== "error" + }) + + const waitForSync = Effect.fn("Workspace.waitForSync")(function* ( + workspaceID: WorkspaceV2.ID, + state: Record, + signal?: AbortSignal, + timeout = TIMEOUT, + ) { + if (yield* synced(db, state)) return + + yield* Effect.catch( + waitUntilSynced({ db, workspaceID, state, signal, timeout }), + (): Effect.Effect => + signal?.aborted + ? Effect.fail( + new SyncAbortedError({ + message: signal.reason instanceof Error ? signal.reason.message : "Request aborted", + cause: signal.reason, + }), + ) + : Effect.fail( + new SyncTimeoutError({ + message: `Timed out waiting for sync fence: ${JSON.stringify(state)}`, + state, + }), + ), + ) + }) + + const startWorkspaceSyncing = Effect.fn("Workspace.startWorkspaceSyncing")(function* (projectID: ProjectV2.ID) { + if (!flags.experimentalWorkspaces) return + const rows = yield* db + .selectDistinct({ workspace: WorkspaceTable }) + .from(WorkspaceTable) + .where(eq(WorkspaceTable.project_id, projectID)) + .all() + .pipe(Effect.orDie) + + for (const { workspace } of rows) { + yield* startSync(fromRow(workspace)).pipe( + Effect.catch((error) => + Effect.sync(() => { + setStatus(workspace.id, "error") + }), + ), + Effect.forkDetach, + ) + } + }) + + return Service.of({ + create, + sessionWarp, + list, + syncList, + get, + remove, + status, + isSyncing, + waitForSync, + startWorkspaceSyncing, + }) + }), +) + +const TIMEOUT = 5000 + +type HistoryEvent = { + id: string + aggregate_id: string + seq: number + type: string + data: Record +} + +function waitUntilSynced(input: { + db: Database.Interface["db"] + workspaceID: WorkspaceV2.ID + state: Record + signal?: AbortSignal + timeout: number +}): Effect.Effect { + return Effect.suspend(() => + waitEvent({ + timeout: input.timeout, + signal: input.signal, + fn(event) { + return event.workspace === input.workspaceID || event.payload.type === "sync" + }, + }).pipe( + Effect.andThen(synced(input.db, input.state)), + Effect.flatMap((done): Effect.Effect => (done ? Effect.void : waitUntilSynced(input))), + ), + ) +} + +function synced(db: Database.Interface["db"], state: Record): Effect.Effect { + const ids = Object.keys(state) + if (ids.length === 0) return Effect.succeed(true) + + return db + .select({ + id: EventSequenceTable.aggregate_id, + seq: EventSequenceTable.seq, + }) + .from(EventSequenceTable) + .where(inArray(EventSequenceTable.aggregate_id, ids)) + .all() + .pipe( + Effect.orDie, + Effect.map((rows) => { + const done = Object.fromEntries(rows.map((row) => [row.id, row.seq])) as Record + return ids.every((id) => (done[id] ?? -1) >= state[id]) + }), + ) +} + +function route(url: string | URL, path: string) { + const next = new URL(url) + next.pathname = `${next.pathname.replace(/\/$/, "")}${path}` + next.search = "" + next.hash = "" + return next +} + +export const node = LayerNode.make({ + service: Service, + layer: layer, + deps: [ + Auth.node, + Session.node, + SessionPrompt.node, + httpClient, + EventV2Bridge.node, + Vcs.node, + RuntimeFlags.node, + FSUtil.node, + Database.node, + ], +}) + +export * as Workspace from "./workspace" diff --git a/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts index d366bdd4..b4d95ce7 100644 --- a/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts +++ b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts @@ -1,94 +1,43 @@ -import { EventV2Bridge } from "@/event-v2-bridge" import { InstanceState } from "@/effect/instance-state" -import { GlobalBus } from "@/bus/global" +import { GlobalEventReplay, formatID, type Frame } from "@/bus/global-replay" import { sseTickTelemetry } from "@/server/shared/sse-tick-telemetry" +import { makeBoundedSseWriter } from "@/server/shared/sse-bounded-writer" import { EventV2 } from "@opencode-ai/core/event" -import { NodeHttpServerRequest } from "@effect/platform-node" import { Effect, Queue } from "effect" import * as Stream from "effect/Stream" -import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http" +import { HttpServerRequest } from "effect/unstable/http" import { HttpApiBuilder } from "effect/unstable/httpapi" import * as Sse from "effect/unstable/encoding/Sse" import { EventApi, EventPaths } from "../groups/event" type SsePayload = { id: string; type: string; properties: unknown } +// One wire frame: the SSE `id` line plus the payload the client parses. The +// ported sequencing (#1806 P1a, main's #1776) puts the bus's `:` +// on the id line; the payload keeps its `evt_` id — the client's dedup keys +// and (#1264) cursor vocabulary live there, and nothing about the data line +// changes. +type WireFrame = { id: string; payload: SsePayload } + // SSE reconnects are silent: EventSource retries automatically, heartbeats keep // the stream looking alive, and every event published during the disconnect is // lost forever — tabs wedge mid-turn ("thinking" with no updates) until a -// manual reload. The client already carries a cursor (#1264: sessionStorage -// `lastEventID` query param; browser EventSource does the same via the -// Last-Event-ID header once the server emits SSE ids). This side of the -// contract: tag every SSE event with its id, keep a bounded process-wide ring -// of recent payloads, and replay the gap on reconnect. When the cursor -// predates the ring's retention (or a fresh process cannot cover it at all), -// emit a `state.resync` sentinel instead — clients should refetch fully. -const RING_CAPACITY = 8192 -const RING_BYTE_BUDGET = 48 * 1024 * 1024 - -// ── #1798 phase C: the bounded fan-out write path ───────────────────────────── -// -// The platform's response writer parks FOREVER on a FIN-less corpse: -// nodeResponse.write() returns false once the socket buffer is full, and the -// fiber then waits for a `drain` event the corpse never produces (the -// frontdoor recycler kills its upstreams with close() while they are blocked -// in recv — no FIN, the frontdoor's own #1757 comment). Unbounded, that park -// wedges the connection's whole recurring emission: Stream.tick and the bus -// fan-out back up behind the write, heartbeats die while HTTP stays healthy, -// the frontdoor's dead-man recycler kills MORE upstreams, and the loop -// propagates fleet-wide. Bound it instead: emission hands every chunk to a -// small bounded per-connection queue with an offer timeout. When the offer -// times out, the subscriber is a corpse — drop it (registry unregister + one -// SSE-DROP telemetry line + engine-side socket close) and keep emitting to -// every other subscriber. A dead subscriber wedges NOTHING. -const WRITE_QUEUE_CAPACITY = 64 -// The writer's end-of-stream sentinel handed through the write queue: a -// zero-length chunk is never a real SSE frame. -const END_OF_STREAM = new Uint8Array(0) - -class WriteStallError { - readonly _tag = "WriteStallError" -} - -interface RingEntry extends SsePayload { - readonly directory: string | undefined - readonly workspaceID: string | undefined - readonly size: number -} +// manual reload. This side of the contract, ported from main's #1776 (#1806 +// P1a): the global bus stamps every frame `:`, ONE shared replay +// buffer (byte+age bounded, env-tunable) serves cursor resumption, and an +// unservable cursor (evicted past the bound, a foreign epoch, ahead of the +// stream) gets the explicit `state.resync` sentinel instead — the client +// refetches fully, never a silent gap. The client's existing payload-id +// cursors (#1264: the last payload id it applied, query param or +// Last-Event-ID header) keep resolving against retained frames — wire +// compatible, no client change rides with the port. -const ring: Array = [] -let ringBytes = 0 -let ringEvictedThrough: string | undefined - -const ringAppend = (event: EventV2.Payload): Effect.Effect => - Effect.sync(() => { - const last = ring[ring.length - 1] - if (last !== undefined && last.id === event.id) return - const data = JSON.stringify(event.data) - const size = event.id.length + event.type.length + data.length - ringBytes += size - ring.push({ - id: event.id, - type: event.type, - properties: event.data, - directory: event.location?.directory, - workspaceID: event.location?.workspaceID, - size, - }) - while (ring.length > RING_CAPACITY || (ring.length > 0 && ringBytes > RING_BYTE_BUDGET)) { - const evicted = ring.shift() - if (evicted === undefined) break - ringBytes -= evicted.size - ringEvictedThrough = evicted.id - } - }) - -function eventData(data: SsePayload): Sse.Event { +function eventData(frame: WireFrame): Sse.Event { return { _tag: "Event", event: "message", - id: data.id, - data: JSON.stringify(data), + id: frame.id, + data: JSON.stringify(frame.payload), } } @@ -107,54 +56,78 @@ function cursorFromRequest(request: HttpServerRequest.HttpServerRequest): string return request.headers["last-event-id"] ?? undefined } -function eventResponse(events: EventV2.Interface) { +function eventResponse() { return Effect.gen(function* () { const instance = yield* InstanceState.context const workspaceID = yield* InstanceState.workspaceID - const cursor = cursorFromRequest(yield* HttpServerRequest.HttpServerRequest) - // Listener registration is eager, so events published after this point cannot - // be lost while the HTTP body fiber is starting or emitting server.connected. - const queue = yield* Queue.unbounded() - const unsubscribe = yield* events.listen((event) => Effect.sync(() => Queue.offerUnsafe(queue, event))) - yield* Effect.addFinalizer(() => unsubscribe) - const stream = Stream.fromQueue(queue).pipe( - Stream.filter( - (event) => - event.location?.directory === instance.directory && - (event.location.workspaceID === undefined || event.location.workspaceID === workspaceID), - ), - Stream.map((event) => ({ id: event.id, type: event.type, properties: event.data })), - ) - const disposed = Stream.callback<{ id: string; type: string; properties: unknown }>((queue) => { - const listener = (event: { - directory?: string - payload: { id?: string; type?: string; properties?: unknown } - }) => { - if (event.directory !== instance.directory || event.payload.type !== "server.instance.disposed") return - Queue.offerUnsafe(queue, { - id: event.payload.id ?? eventID(), - type: "server.instance.disposed", - properties: event.payload.properties ?? {}, - }) - } - return Effect.acquireRelease( - Effect.sync(() => GlobalBus.on("event", listener)), - () => Effect.sync(() => GlobalBus.off("event", listener)), - ) - }) - const output = stream.pipe( - Stream.merge(disposed, { haltStrategy: "left" }), - Stream.takeUntil((event) => event.type === "server.instance.disposed"), - ) + const request = yield* HttpServerRequest.HttpServerRequest + const cursor = cursorFromRequest(request) // #1798 phase A: this connection's tick telemetry identity. The registry - // is eager (like the listener above) and the stream finalizer unregisters; - // the heartbeat cadence is single-sourced from the telemetry singleton — - // the contract in sse-tick-telemetry.ts is the one place both must agree. + // is eager and the stream finalizer unregisters; the heartbeat cadence is + // single-sourced from the telemetry singleton — the contract in + // sse-tick-telemetry.ts is the one place both must agree. const connId = eventID() sseTickTelemetry.registerConnection(EventPaths.event, connId) + // The connection's event queue: everything the shared buffer hands this + // instance's view of the stream, in strict bus order. Registration is + // eager, so events published after this point cannot be lost while the + // HTTP body fiber is starting or emitting server.connected. + const queue = yield* Queue.unbounded() + const matches = (frame: Frame) => { + const event = frame.event + if (event.directory !== instance.directory) return false + if (event.workspace !== undefined && event.workspace !== workspaceID) return false + return true + } + const payloadOf = (frame: Frame): SsePayload => { + const payload = frame.event.payload as SsePayload + return { id: payload.id, type: payload.type, properties: payload.properties } + } + // The shared replay buffer is the connection's live source: one buffer per + // bus, shared by every subscriber (the global stream route and this + // instance-filtered view alike), with the bus's pre-dispatch tap doing the + // sequencing. An unservable cursor surfaces as `subscription.reset` below. + const subscription = yield* Effect.acquireRelease( + Effect.sync(() => { + const subscription = GlobalEventReplay.subscribe(cursor) + subscription.start((frame) => { + if (!matches(frame)) return + Queue.offerUnsafe(queue, { id: frame.id, payload: payloadOf(frame) }) + }) + return subscription + }), + (subscription) => Effect.sync(() => subscription.close()), + ) + const replay = subscription.replay.filter(matches).map((frame) => ({ id: frame.id, payload: payloadOf(frame) })) + const preamble: Array = [ + // Control frames never consume a sequence; their id line carries the + // connection's current position, so ids on one connection never decrease. + { + id: formatID(GlobalEventReplay.epoch, subscription.position), + payload: { id: eventID(), type: "server.connected", properties: {} }, + }, + ...(subscription.reset + ? [ + { + id: formatID(GlobalEventReplay.epoch, subscription.position), + payload: { + id: eventID(), + type: "state.resync", + properties: { reason: subscription.reset } satisfies unknown, + }, + }, + ] + : []), + ...replay, + ] + + yield* Effect.logInfo("event connected") const heartbeat = Stream.tick(sseTickTelemetry.cadenceMs).pipe( Stream.drop(1), - Stream.map(() => ({ id: eventID(), type: "server.heartbeat", properties: {} })), + Stream.map(() => ({ + id: formatID(GlobalEventReplay.epoch, GlobalEventReplay.current), + payload: { id: eventID(), type: "server.heartbeat", properties: {} }, + })), // One telemetry line per heartbeat: the emitting fiber's identity and // the emitter's queue depth ride every tick (queue depth at stall onset // discriminates the blocked-shared-emit-loop hypothesis), alongside the @@ -170,153 +143,43 @@ function eventResponse(events: EventV2.Interface) { ), ) - const replay = new Array() - let resync = false - if (cursor !== undefined) { - const oldest = ring[0] - if ( - oldest === undefined || - cursor < oldest.id || - (ringEvictedThrough !== undefined && cursor <= ringEvictedThrough) - ) { - resync = true - } else { - for (const entry of ring) { - if (entry.id <= cursor) continue - if (entry.directory !== instance.directory) continue - if (entry.workspaceID !== undefined && entry.workspaceID !== workspaceID) continue - replay.push(entry) - } - } - } - const replayedIds = new Set(replay.map((event) => event.id)) - const replayWatermark = replay.length > 0 ? replay[replay.length - 1]!.id : undefined - const live = output.pipe( - Stream.filter( - (event) => - !replayedIds.has(event.id) && (replayWatermark === undefined || event.id > replayWatermark), - ), + const live = Stream.fromQueue(queue).pipe( + // The instance's disposal arrives as an ordinary frame (the disposed + // event rides the same bus): end the body at it so the response FINs + // cleanly instead of wedging a dead instance's stream open. + Stream.takeUntil((frame) => frame.payload.type === "server.instance.disposed"), ) - const preamble: Array = [ - { id: eventID(), type: "server.connected", properties: {} }, - ...(resync - ? [{ id: eventID(), type: "state.resync", properties: { reason: "retention" } satisfies unknown }] - : []), - ...replay, - ] - yield* Effect.logInfo("event connected") - // #1798 phase C: emit through a bounded, non-blocking write path instead - // of handing the merged stream to the platform writer (which parks - // forever on a full corpse socket — see the WRITE_QUEUE_CAPACITY block - // above). This connection's writer fiber encodes every frame and hands - // each chunk to a small bounded queue; the response body is just that - // queue, so backpressure from a dead socket can only ever fill 64 chunks - // before the offer bound trips and the corpse is dropped. - const serverRequest = yield* HttpServerRequest.HttpServerRequest + // #1798 phase C: emit through the shared bounded, non-blocking write path + // (sse-bounded-writer.ts) instead of handing the merged stream to the + // platform writer (which parks forever on a full corpse socket). The + // writer fiber encodes every frame and hands each chunk to a small bounded + // queue; a stalled offer trips the bound, names the corpse in telemetry + // (SSE-DROP) and tears its socket down — a dead subscriber wedges NOTHING. const scope = yield* Effect.scope - const writeQueue = yield* Queue.bounded(WRITE_QUEUE_CAPACITY) - // Half a watchdog window: a stalled subscriber is dropped before its - // silence could ever be declared a process-wide tick stall — the engine - // heals itself instead of naming its own death. - const writeStallMs = Math.max(1, Math.floor(sseTickTelemetry.watchdogIntervalMs / 2)) - let streamEnded = false - let dropped = false - const dropConnection = () => - Effect.uninterruptible( - Effect.gen(function* () { - if (dropped) return - dropped = true - // The drop line beats the eviction: the registry forgets the corpse - // and the telemetry names it, with the depths that show the write - // path parked (events kept piling in, the write queue pinned full). - sseTickTelemetry.noteDrop(EventPaths.event, connId, { - reason: "stalled_write", - eventQueueDepth: Queue.sizeUnsafe(queue), - writeQueueDepth: Queue.sizeUnsafe(writeQueue), - }) - yield* Queue.shutdown(writeQueue) - // Tear the connection down on our side: destroy the response and - // the socket it rides, which interrupts the parked platform writer - // through the response 'close' handler. (Bun runtime gap, measured - // 2026-10-14: none of res.destroy / socket.destroy / - // socket.resetAndDestroy actually send a FIN or RST to the peer — - // the wire-level close is the frontdoor slice's end of the - // contract; the engine-side teardown is what unwedges the fiber and - // frees the connection bookkeeping.) - yield* Effect.sync(() => { - const rawResponse = NodeHttpServerRequest.toServerResponse(serverRequest) - // Only the platform-node server carries the raw response; other - // platforms skip the socket close (the interrupt-only exit still - // unwedges the fiber). - if (typeof rawResponse?.destroy === "function") rawResponse.destroy() - const socket = rawResponse?.socket - if (typeof socket?.destroy === "function") socket.destroy() - }) - }), - ) - const offerBounded = (chunk: Uint8Array) => - Queue.offer(writeQueue, chunk).pipe( - Effect.timeoutOrElse({ - duration: writeStallMs, - orElse: () => dropConnection().pipe(Effect.andThen(Effect.fail(new WriteStallError()))), - }), - ) - yield* Stream.fromArray(preamble) - .pipe( + return yield* makeBoundedSseWriter({ + serverRequest: request, + route: EventPaths.event, + connId, + source: Stream.fromArray(preamble).pipe( Stream.concat(live.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }))), Stream.map(eventData), - Stream.pipeThroughChannel(Sse.encode()), - Stream.encodeText, - Stream.runForEach(offerBounded), - Effect.flatMap(() => { - // Normal end (instance disposed): hand the sentinel through so the - // response drains every buffered chunk and FINs cleanly. - streamEnded = true - return offerBounded(END_OF_STREAM) - }), - Effect.catchTag("WriteStallError", () => Effect.void), - Effect.ensuring( - Effect.sync(() => { - // Interrupt (client hangup) or unexpected failure: end the body - // stream rather than leaving a parked consumer. The drop path has - // already shut the queue down itself. - if (!streamEnded && !dropped) Queue.shutdown(writeQueue) - }), - ), - Effect.forkIn(scope), - ) - return HttpServerResponse.stream( - Stream.fromQueue(writeQueue).pipe( - Stream.takeUntil((chunk) => chunk.length === 0), - Stream.ensuring( - Effect.gen(function* () { - yield* Effect.logInfo("event disconnected") - yield* Effect.sync(() => sseTickTelemetry.unregisterConnection(connId)) - }), - ), ), - { - contentType: "text/event-stream", - headers: { - "Cache-Control": "no-cache, no-transform", - "X-Accel-Buffering": "no", - "X-Content-Type-Options": "nosniff", - }, - }, - ) + eventQueueDepth: () => Queue.sizeUnsafe(queue), + onEnded: Effect.gen(function* () { + yield* Effect.logInfo("event disconnected") + yield* Effect.sync(() => sseTickTelemetry.unregisterConnection(connId)) + }), + scope, + }) }) } export const eventHandlers = HttpApiBuilder.group(EventApi, "event", (handlers) => - Effect.gen(function* () { - const events = yield* EventV2Bridge.Service - yield* events.listen(ringAppend) - return handlers.handleRaw( - "subscribe", - Effect.fn("EventHttpApi.subscribe")(function* () { - return yield* eventResponse(events) - }), - ) - }), + handlers.handleRaw( + "subscribe", + Effect.fn("EventHttpApi.subscribe")(function* () { + return yield* eventResponse() + }), + ), ) diff --git a/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts new file mode 100644 index 00000000..d3addbe3 --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/src/server/routes/instance/httpapi/handlers/global.ts @@ -0,0 +1,230 @@ +import { Config } from "@/config/config" +import { GlobalBus } from "@/bus/global" +import { GlobalEventReplay, formatID, type Frame, type Subscription } from "@/bus/global-replay" +import { EffectBridge } from "@/effect/bridge" +import { EventV2 } from "@opencode-ai/core/event" +import { Installation } from "@/installation" +import { disposeAllInstancesAndEmitGlobalDisposed } from "@/server/global-lifecycle" +import { InstallationVersion } from "@opencode-ai/core/installation/version" +import { sseTickTelemetry } from "@/server/shared/sse-tick-telemetry" +import { makeBoundedSseWriter } from "@/server/shared/sse-bounded-writer" +import { Clock, Deferred, Effect, Exit, Queue } from "effect" +import * as Stream from "effect/Stream" +import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http" +import { HttpApiBuilder } from "effect/unstable/httpapi" +import * as Sse from "effect/unstable/encoding/Sse" +import type { Socket } from "node:net" +import { RootHttpApi } from "../api" +import { GlobalPaths, GlobalUpgradeInput } from "../groups/global" + +// The resume cursor: the standard `Last-Event-ID` header (what an SSE client +// sends once it has seen an `id:` line), else the `lastEventID` query +// parameter clients already send. +function cursorOf(request: HttpServerRequest.HttpServerRequest) { + const header = request.headers["last-event-id"]?.trim() + if (header) return header + return new URL(request.url, "http://localhost").searchParams.get("lastEventID")?.trim() || undefined +} + +// Sync copies (the event-v2 bridge's `type: "sync"` twin of every durable +// event) are opt-in per subscription (#1778): only a subscriber that replays +// them (the workspace control plane, control-plane/workspace.ts) asks — with +// the `x-opencode-sync-copies` header, else the `sync` query parameter; every +// other subscriber discards them, so they stay off its wire. +function syncRequested(request: HttpServerRequest.HttpServerRequest) { + const raw = + request.headers["x-opencode-sync-copies"] ?? new URL(request.url, "http://localhost").searchParams.get("sync") + const value = raw?.trim().toLowerCase() + return value === "true" || value === "1" +} + +function frameEvent(frame: Pick): Sse.Event { + return { _tag: "Event", event: "message", id: frame.id, data: frame.data } +} + +// A per-connection control frame (connected, heartbeat, reset). It never +// consumes a sequence — that would punch a hole in every other subscriber's +// stream — and carries the connection's current position as its id instead, +// so ids on one connection never decrease. +function controlEvent(seq: number, type: string, properties: Record = {}): Sse.Event { + return frameEvent({ + id: formatID(GlobalEventReplay.epoch, seq), + data: JSON.stringify({ payload: { id: EventV2.ID.create(), type, properties } }), + }) +} + +// The node:http adapter interrupts a request fiber only on the response's +// "close" event, which Bun's node:http shim (the engine's runtime) never emits +// when a client aborts a streaming response — it emits the request's +// "aborted"/"close" and the socket's "close", and writes to the dead socket keep +// returning true. Without this the stream, its bus subscription and its queue +// outlive the client forever. Watch the socket itself; the response owns it +// for the stream's lifetime. +function clientGone(request: HttpServerRequest.HttpServerRequest) { + return Effect.gen(function* () { + const gone = yield* Deferred.make() + const done = () => Deferred.doneUnsafe(gone, Exit.void) + const socket = (request.source as { socket?: Partial } | undefined)?.socket + if (!socket || typeof socket.once !== "function" || typeof socket.off !== "function") return gone + if (socket.destroyed) { + done() + return gone + } + socket.once("close", done) + yield* Effect.addFinalizer(() => Effect.sync(() => socket.off?.("close", done))) + return gone + }) +} + +function eventResponse() { + return Effect.gen(function* () { + const request = yield* HttpServerRequest.HttpServerRequest + const cursor = cursorOf(request) + const sync = syncRequested(request) + const gone = yield* clientGone(request) + // Registration is eager and synchronous: a bus event emitted while the + // response body is starting is queued, never lost. + const queue = yield* Queue.unbounded() + const release = (subscription: Subscription) => Effect.sync(() => subscription.close()) + const subscription = yield* Effect.acquireRelease( + Effect.sync(() => { + const subscription = GlobalEventReplay.subscribe(cursor, { sync }) + // An unservable cursor is never silent: a reset frame comes first, + // carrying the current epoch and sequence, and the client resyncs. + if (subscription.reset) + Queue.offerUnsafe( + queue, + controlEvent(subscription.position, "server.cursor.unservable", { + epoch: GlobalEventReplay.epoch, + seq: subscription.position, + }), + ) + Queue.offerUnsafe(queue, controlEvent(subscription.position, "server.connected")) + for (const frame of subscription.replay) Queue.offerUnsafe(queue, frameEvent(frame)) + subscription.start((frame) => Queue.offerUnsafe(queue, frameEvent(frame))) + return subscription + }), + release, + ) + // #1798/#1804: the connection's tick telemetry identity, single-sourced + // cadence, and registry entry — the same contract the instance /event + // route carries, so the watchdog covers every live SSE connection. + const connId = EventV2.ID.create() + sseTickTelemetry.registerConnection(GlobalPaths.event, connId) + yield* Effect.logInfo("global event connected").pipe( + Effect.annotateLogs({ + subscribers: GlobalEventReplay.subscribers, + resume: subscription.reset ? `reset:${subscription.reset}` : cursor ? "replay" : "live", + replayed: subscription.replay.length, + sync, + }), + ) + // The heartbeat enters the connection's own queue, behind every frame up to + // the current sequence, so a backlogged connection never sees a heartbeat + // claim a sequence it has not been sent yet (that would read as a hole). + const heartbeat = Stream.tick(sseTickTelemetry.cadenceMs).pipe( + Stream.drop(1), + Stream.mapEffect(() => Effect.map(Clock.currentTimeMillis, (time) => { + const seq = GlobalEventReplay.current + console.log("[hb] tick fired, offering beat"); + Queue.offerUnsafe(queue, controlEvent(seq, "server.heartbeat", { time, seq })) + })), + Stream.tap(() => + Effect.gen(function* () { + const fiberId = yield* Effect.fiberId + sseTickTelemetry.noteTick(GlobalPaths.event, connId, { + queueDepth: Queue.sizeUnsafe(queue), + fiberId, + }) + }), + ), + Stream.drain, + ) + + const scope = yield* Effect.scope + return yield* makeBoundedSseWriter({ + serverRequest: request, + route: GlobalPaths.event, + connId, + source: Stream.fromQueue(queue).pipe(Stream.merge(heartbeat, { haltStrategy: "left" })), + haltWhen: gone, + eventQueueDepth: () => Queue.sizeUnsafe(queue), + onEnded: Effect.gen(function* () { + yield* release(subscription) + yield* Effect.sync(() => sseTickTelemetry.unregisterConnection(connId)) + yield* Effect.logInfo("global event disconnected").pipe( + Effect.annotateLogs({ subscribers: GlobalEventReplay.subscribers, dropped: Queue.sizeUnsafe(queue) }), + ) + }), + scope, + }) + }) +} + +export const globalHandlers = HttpApiBuilder.group(RootHttpApi, "global", (handlers) => + Effect.gen(function* () { + const config = yield* Config.Service + const installation = yield* Installation.Service + const bridge = yield* EffectBridge.make() + + const health = Effect.fn("GlobalHttpApi.health")(function* () { + return { healthy: true as const, version: InstallationVersion } + }) + + const event = Effect.fn("GlobalHttpApi.event")(function* () { + return yield* eventResponse() + }) + + const configGet = Effect.fn("GlobalHttpApi.configGet")(function* () { + return yield* config.getGlobal() + }) + + const configUpdate = Effect.fn("GlobalHttpApi.configUpdate")(function* (ctx) { + const result = yield* config.updateGlobal(ctx.payload) + if (result.changed) bridge.fork(disposeAllInstancesAndEmitGlobalDisposed({ swallowErrors: true })) + return result.info + }) + + const dispose = Effect.fn("GlobalHttpApi.dispose")(function* () { + yield* disposeAllInstancesAndEmitGlobalDisposed() + return true + }) + + const upgrade = Effect.fn("GlobalHttpApi.upgrade")(function* (ctx: { payload: typeof GlobalUpgradeInput.Type }) { + const method = yield* installation.method() + if (method === "unknown") { + return HttpServerResponse.jsonUnsafe( + { success: false as const, error: "Unknown installation method" }, + { status: 400 }, + ) + } + const target = ctx.payload.target + const result = yield* installation.upgrade(method, target).pipe( + Effect.as({ success: true as const, version: target }), + Effect.catch((err) => + Effect.succeed({ + success: false as const, + error: err instanceof Error ? err.message : String(err), + }), + ), + ) + if (!result.success) return HttpServerResponse.jsonUnsafe(result, { status: 500 }) + GlobalBus.emit("event", { + directory: "global", + payload: { + type: Installation.Event.Updated.type, + properties: { version: target }, + }, + }) + return HttpServerResponse.jsonUnsafe(result) + }) + + return handlers + .handle("health", health) + .handleRaw("event", event) + .handle("configGet", configGet) + .handle("configUpdate", configUpdate) + .handle("dispose", dispose) + .handle("upgrade", upgrade) + }), +) diff --git a/packages/app-bundle/overlay/packages/opencode/src/server/shared/sse-bounded-writer.ts b/packages/app-bundle/overlay/packages/opencode/src/server/shared/sse-bounded-writer.ts new file mode 100644 index 00000000..fc477c84 --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/src/server/shared/sse-bounded-writer.ts @@ -0,0 +1,153 @@ +// ── #1798 phase C: the bounded fan-out write path ───────────────────────────── +// +// The platform's response writer parks FOREVER on a FIN-less corpse: +// nodeResponse.write() returns false once the socket buffer is full, and the +// fiber then waits for a `drain` event the corpse never produces (the +// frontdoor recycler kills its upstreams with close() while they are blocked +// in recv — no FIN, the frontdoor's own #1757 comment). Unbounded, that park +// wedges the connection's whole recurring emission: Stream.tick and the bus +// fan-out back up behind the write, heartbeats die while HTTP stays healthy, +// the frontdoor's dead-man recycler kills MORE upstreams, and the loop +// propagates fleet-wide. Bound it instead: emission hands every chunk to a +// small bounded per-connection queue with an offer timeout. When the offer +// times out, the subscriber is a corpse — drop it (registry unregister + one +// SSE-DROP telemetry line + engine-side socket close) and keep emitting to +// every other subscriber. A dead subscriber wedges NOTHING. +// +// Shared by every SSE route (the instance /event route and the global +// /global/event route — #1806 P1a composes the ported global stream onto this +// same write path, so one corpse-proof discipline covers both). +import { NodeHttpServerRequest } from "@effect/platform-node" +import { Deferred, Effect, Queue, Stream } from "effect" +import type { Scope } from "effect" +import * as Sse from "effect/unstable/encoding/Sse" +import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http" +import { sseTickTelemetry } from "./sse-tick-telemetry" + +// The per-connection bounded write queue: 64 chunks of headroom before the +// offer bound trips and the subscriber is declared a corpse. +const WRITE_QUEUE_CAPACITY = 64 +// The writer's end-of-stream sentinel handed through the write queue: a +// zero-length chunk is never a real SSE frame. +const END_OF_STREAM = new Uint8Array(0) + +class WriteStallError { + readonly _tag = "WriteStallError" +} + +/** + * Run an SSE source through the bounded, non-blocking write path instead of + * handing the stream to the platform writer (which parks forever on a full + * corpse socket — see the header). This connection's writer fiber encodes + * every frame and hands each chunk to a small bounded queue; the response body + * is just that queue, so backpressure from a dead socket can only ever fill + * 64 chunks before the offer bound trips and the corpse is dropped. + */ +export function makeBoundedSseWriter(input: { + serverRequest: HttpServerRequest.HttpServerRequest + /** Telemetry route tag (rides the SSE-TICK/SSE-DROP lines). */ + route: string + /** Telemetry connection identity. */ + connId: string + /** The frames to write, already merged with the heartbeat. */ + source: Stream.Stream + /** The connection's pre-encode event queue depth (the SSE-DROP discriminator). */ + eventQueueDepth: () => number + /** Runs once when the response body ends (drop, client gone, or clean end). */ + onEnded: Effect.Effect + /** Optional abort source (a resolved Deferred ends the write). */ + haltWhen?: Deferred.Deferred + scope: Scope.Scope +}): Effect.Effect { + return Effect.gen(function* () { + const writeQueue = yield* Queue.bounded(WRITE_QUEUE_CAPACITY) + // Half a watchdog window: a stalled subscriber is dropped before its + // silence could ever be declared a process-wide tick stall — the engine + // heals itself instead of naming its own death. + const writeStallMs = Math.max(1, Math.floor(sseTickTelemetry.watchdogIntervalMs / 2)) + let streamEnded = false + let dropped = false + const dropConnection = () => + Effect.uninterruptible( + Effect.gen(function* () { + if (dropped) return + dropped = true + // The drop line beats the eviction: the registry forgets the corpse + // and the telemetry names it, with the depths that show the write + // path parked (events kept piling in, the write queue pinned full). + sseTickTelemetry.noteDrop(input.route, input.connId, { + reason: "stalled_write", + eventQueueDepth: input.eventQueueDepth(), + writeQueueDepth: Queue.sizeUnsafe(writeQueue), + }) + yield* Queue.shutdown(writeQueue) + // Tear the connection down on our side: destroy the response and + // the socket it rides, which interrupts the parked platform writer + // through the response 'close' handler. (Bun runtime gap, measured + // 2026-10-14: none of res.destroy / socket.destroy / + // socket.resetAndDestroy actually send a FIN or RST to the peer — + // the wire-level close is the frontdoor slice's end of the + // contract; the engine-side teardown is what unwedges the fiber and + // frees the connection bookkeeping.) + yield* Effect.sync(() => { + const rawResponse = NodeHttpServerRequest.toServerResponse(input.serverRequest) + // Only the platform-node server carries the raw response; other + // platforms skip the socket close (the interrupt-only exit still + // unwedges the fiber). + if (typeof rawResponse?.destroy === "function") rawResponse.destroy() + const socket = rawResponse?.socket + if (typeof socket?.destroy === "function") socket.destroy() + }) + }), + ) + const offerBounded = (chunk: Uint8Array) => + Queue.offer(writeQueue, chunk).pipe( + Effect.timeoutOrElse({ + duration: writeStallMs, + orElse: () => dropConnection().pipe(Effect.andThen(Effect.fail(new WriteStallError()))), + }), + ) + const source = + input.haltWhen === undefined + ? input.source + : input.source.pipe(Stream.interruptWhen(Deferred.await(input.haltWhen))) + yield* source + .pipe( + Stream.pipeThroughChannel(Sse.encode()), + Stream.encodeText, + Stream.runForEach(offerBounded), + Effect.flatMap(() => { + // Normal end (instance disposed): hand the sentinel through so the + // response drains every buffered chunk and FINs cleanly. + streamEnded = true + return offerBounded(END_OF_STREAM) + }), + Effect.catchTag("WriteStallError", () => Effect.void), + Effect.ensuring( + Effect.sync(() => { + // Interrupt (client hangup) or unexpected failure: end the body + // stream rather than leaving a parked consumer. The drop path has + // already shut the queue down itself. + if (!streamEnded && !dropped) Queue.shutdown(writeQueue) + }), + ), + Effect.forkIn(input.scope), + ) + return HttpServerResponse.stream( + Stream.fromQueue(writeQueue).pipe( + Stream.takeUntil((chunk) => chunk.length === 0), + Stream.ensuring(input.onEnded), + ), + { + contentType: "text/event-stream", + headers: { + "Cache-Control": "no-cache, no-transform", + "X-Accel-Buffering": "no", + "X-Content-Type-Options": "nosniff", + }, + }, + ) + }) +} + +export * as SseBoundedWriter from "./sse-bounded-writer" diff --git a/packages/app-bundle/overlay/packages/opencode/test/server/global-bus-sequencing.test.ts b/packages/app-bundle/overlay/packages/opencode/test/server/global-bus-sequencing.test.ts new file mode 100644 index 00000000..f023483a --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/test/server/global-bus-sequencing.test.ts @@ -0,0 +1,129 @@ +// #1776 (ported, #1806 P1a) — the global bus stamps source-side: a random +// per-process epoch, one strictly increasing seq assigned once per event at +// emit, taps that observe events BEFORE any ordinary listener (so nested emits +// stay strictly ordered), and the mirror window that lets a sync copy share +// its source's sequence. An engine restart is a fresh epoch — the resync +// boundary the reset contract rides on. +import { describe, expect, test } from "bun:test" +import path from "node:path" +import { GlobalBus, type GlobalEvent } from "../../src/bus/global" + +const emit = (type: string, properties: Record = {}) => + GlobalBus.emit("event", { directory: "/tmp/p1a-bus", payload: { type, properties } }) + +describe("global bus — source-side sequencing (#1776 port)", () => { + test("stamps every emitted event with a strictly increasing seq under a random epoch-shaped token", () => { + expect(GlobalBus.epoch).toMatch(/^[A-Za-z0-9_-]+$/) + const base = GlobalBus.seq + const seqs: number[] = [] + const untap = GlobalBus.tap((_event, seq) => seqs.push(seq)) + try { + emit("test.seq.a") + emit("test.seq.b") + emit("test.seq.c") + } finally { + untap() + } + expect(seqs).toEqual([base + 1, base + 2, base + 3]) + expect(GlobalBus.seq).toBe(base + 3) + }) + + test("a pre-dispatch tap observes nested emits strictly after their outer event", () => { + const observed: Array<{ type: string; seq: number }> = [] + const untap = GlobalBus.tap((event, seq) => observed.push({ type: event.payload.type, seq })) + const listener = (event: GlobalEvent) => { + if (event.payload.type === "test.outer") emit("test.inner") + } + GlobalBus.on("event", listener) + try { + emit("test.outer") + } finally { + GlobalBus.off("event", listener) + untap() + } + // Strict sequence order at the tap even though the inner emit happened + // inside a listener's handler: outer was sequenced and tapped first. + expect(observed.map((entry) => entry.type)).toEqual(["test.outer", "test.inner"]) + expect(observed[1]!.seq).toBe(observed[0]!.seq + 1) + }) + + test("an ordinary listener never sees an event before its tap does", () => { + const order: string[] = [] + const untap = GlobalBus.tap((event) => order.push(`tap:${event.payload.type}`)) + const listener = (event: GlobalEvent) => order.push(`listener:${event.payload.type}`) + GlobalBus.on("event", listener) + try { + emit("test.precedence") + } finally { + GlobalBus.off("event", listener) + untap() + } + expect(order).toEqual(["tap:test.precedence", "listener:test.precedence"]) + }) + + test("a sync copy mirrors its source's seq within the window; past the window it takes a fresh one", () => { + const seqOf = (type: string, id?: string) => { + let seq: number | undefined + const untap = GlobalBus.tap((event, s) => { + if (event.payload.type === type) seq = s + }) + try { + GlobalBus.emit("event", { + directory: "/tmp/p1a-bus", + payload: id === undefined ? { type } : { id, type }, + }) + } finally { + untap() + } + return seq! + } + const copyAfter = (id: string) => + GlobalBus.emit("event", { + directory: "/tmp/p1a-bus", + payload: { type: "sync", syncEvent: { id, type: "test.durable.1", seq: 1, aggregateID: "agg", data: {} } }, + }) + + // within the window: the copy follows its source with nothing between + const source = seqOf("test.durable", "evt_p1a_window") + let mirrored: number | undefined + const untapMirror = GlobalBus.tap((event, seq) => { + if (event.payload.type === "sync") mirrored = seq + }) + try { + copyAfter("evt_p1a_window") + } finally { + untapMirror() + } + expect(mirrored).toBe(source) + + // past the window (MIRROR_WINDOW = 256 remembered source ids): the source + // is forgotten, so the copy is sequenced as a frame of its own + for (let i = 0; i < 256; i++) seqOf("test.filler", `evt_p1a_fill_${i}`) + let fresh: number | undefined + const untapFresh = GlobalBus.tap((event, seq) => { + if (event.payload.type === "sync") fresh = seq + }) + try { + copyAfter("evt_p1a_window") + } finally { + untapFresh() + } + expect(fresh).not.toBe(source) + expect(fresh!).toBeGreaterThan(source) + }) + + test("a fresh process gets a fresh epoch — an engine restart is a resync boundary", () => { + const result = Bun.spawnSync({ + cmd: [ + process.execPath, + "-e", + 'const { GlobalBus } = await import("./src/bus/global.ts"); console.log(GlobalBus.epoch)', + ], + cwd: path.join(import.meta.dir, "..", ".."), + }) + expect(result.stderr.toString()).toBe("") + const other = result.stdout.toString().trim() + expect(other).toMatch(/^[A-Za-z0-9_-]+$/) + expect(other).not.toBe(GlobalBus.epoch) + }) +}) diff --git a/packages/app-bundle/overlay/packages/opencode/test/server/global-event-replay.test.ts b/packages/app-bundle/overlay/packages/opencode/test/server/global-event-replay.test.ts new file mode 100644 index 00000000..45b92acc --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/test/server/global-event-replay.test.ts @@ -0,0 +1,173 @@ +// #1776 (ported, #1806 P1a) — the global stream's shared replay buffer: one +// buffer per bus bounded by bytes and age (env-tunable, main's defaults), +// oldest-first eviction, cursor resolution, and the reset contract. The lane +// addition: a cursor that is not `:` (the lane's client tracks +// payload ids) resolves against retained frames — it is served when the +// history covers it and reset when it does not. +import { describe, expect, test } from "bun:test" +import path from "node:path" +import { DEFAULT_BOUNDS, ReplayBuffer, boundsFromEnv, formatID, parseCursor } from "../../src/bus/global-replay" + +const frameOf = (bytes: number) => "x".repeat(bytes) +const eventOf = (type: string, id?: string) => ({ + directory: "/tmp/p1a-replay", + payload: id === undefined ? { type, properties: {} } : { id, type, properties: {} }, +}) + +function buffer(bounds: { maxBytes: number; maxAgeMs: number }, clock = { now: 0 }, epoch = "ep0ch") { + return new ReplayBuffer({ epoch, bounds, now: () => clock.now }) +} + +const seqs = (b: ReplayBuffer, cursor: string) => { + const sub = b.subscribe(cursor) + sub.close() + return sub.reset ?? sub.replay.map((frame) => frame.seq) +} + +describe("global event replay buffer — bounds (#1776 port)", () => { + test("evicts oldest-first once retained bytes exceed the byte bound", () => { + const b = buffer({ maxBytes: 30, maxAgeMs: 60_000 }) + for (const seq of [1, 2, 3]) b.push(seq, eventOf("test.a"), frameOf(10)) + expect(seqs(b, "ep0ch:0")).toEqual([1, 2, 3]) + + b.push(4, eventOf("test.a"), frameOf(10)) // 40 bytes > 30: the oldest frame goes + expect(seqs(b, "ep0ch:0")).toBe("evicted") + expect(seqs(b, "ep0ch:1")).toEqual([2, 3, 4]) + + b.push(5, eventOf("test.a"), frameOf(25)) // 55 bytes: evict 2, 3, 4 (oldest first) until within bound + expect(seqs(b, "ep0ch:3")).toBe("evicted") + expect(seqs(b, "ep0ch:4")).toEqual([5]) + }) + + test("evicts oldest-first once frames are older than the age bound", () => { + const clock = { now: 0 } + const b = buffer({ maxBytes: 1 << 20, maxAgeMs: 1_500 }, clock) + b.push(1, eventOf("test.a"), frameOf(10)) + clock.now = 1_000 + b.push(2, eventOf("test.a"), frameOf(10)) + clock.now = 1_400 + expect(seqs(b, "ep0ch:0")).toEqual([1, 2]) + + clock.now = 2_000 // frame 1 is 2000 ms old (> 1500), frame 2 is 1000 ms old + expect(seqs(b, "ep0ch:0")).toBe("evicted") + expect(seqs(b, "ep0ch:1")).toEqual([2]) + + clock.now = 3_000 // both aged out; a cursor at the current sequence is still servable + b.push(3, eventOf("test.a"), frameOf(10)) + expect(seqs(b, "ep0ch:1")).toBe("evicted") + expect(seqs(b, "ep0ch:2")).toEqual([3]) + expect(seqs(b, "ep0ch:3")).toEqual([]) + }) + + test("both bounds are read from the environment, defaulting to 32 MiB and 5 minutes", () => { + expect(DEFAULT_BOUNDS).toEqual({ maxBytes: 32 * 1024 * 1024, maxAgeMs: 5 * 60_000 }) + expect( + boundsFromEnv({ + OPENCODE_GLOBAL_EVENT_REPLAY_MAX_BYTES: "1024", + OPENCODE_GLOBAL_EVENT_REPLAY_MAX_AGE_MS: "60000", + }), + ).toEqual({ maxBytes: 1024, maxAgeMs: 60_000 }) + expect(boundsFromEnv({})).toEqual(DEFAULT_BOUNDS) + expect( + boundsFromEnv({ OPENCODE_GLOBAL_EVENT_REPLAY_MAX_BYTES: "lots", OPENCODE_GLOBAL_EVENT_REPLAY_MAX_AGE_MS: "-5" }), + ).toEqual(DEFAULT_BOUNDS) + }) + + test("the engine's shared buffer takes its bounds from the process environment", () => { + const result = Bun.spawnSync({ + cmd: [ + process.execPath, + "-e", + 'const { GlobalEventReplay } = await import("./src/bus/global-replay.ts"); console.log(JSON.stringify(GlobalEventReplay.bounds))', + ], + cwd: path.join(import.meta.dir, "..", ".."), + env: { + ...process.env, + OPENCODE_GLOBAL_EVENT_REPLAY_MAX_BYTES: "4096", + OPENCODE_GLOBAL_EVENT_REPLAY_MAX_AGE_MS: "2500", + }, + }) + expect(result.stderr.toString()).toBe("") + expect(JSON.parse(result.stdout.toString())).toEqual({ maxBytes: 4096, maxAgeMs: 2500 }) + }) +}) + +describe("global event replay buffer — the reset contract (#1776 port)", () => { + test("a cursor from a foreign epoch resets (an engine restart is a resync boundary)", () => { + const a = buffer({ maxBytes: 1 << 20, maxAgeMs: 60_000 }, { now: 0 }, "ep0ch") + a.push(1, eventOf("test.a"), frameOf(10)) + const b = buffer({ maxBytes: 1 << 20, maxAgeMs: 60_000 }, { now: 0 }, "other") + const sub = b.subscribe(formatID("ep0ch", 1)) + expect(sub.reset).toBe("epoch") + sub.close() + }) + + test("a cursor ahead of the stream resets, and one at the current sequence serves live", () => { + const b = buffer({ maxBytes: 1 << 20, maxAgeMs: 60_000 }) + b.push(1, eventOf("test.a"), frameOf(10)) + const ahead = b.subscribe("ep0ch:5") + expect(ahead.reset).toBe("ahead") + ahead.close() + const atCurrent = b.subscribe("ep0ch:1") + expect(atCurrent.reset).toBeUndefined() + expect(atCurrent.position).toBe(1) + atCurrent.close() + }) + + test("a garbage cursor below retained history resets; one above it resolves live-only", () => { + const b = buffer({ maxBytes: 1 << 20, maxAgeMs: 60_000 }) + b.push(1, eventOf("test.a", "evt_p1a_malformed_1"), frameOf(10)) + // Below every retained payload id: nothing covers it — the reset contract. + const below = b.subscribe("aaa") + expect(below.reset).toBe("evicted") + below.close() + // Above every retained payload id: the legacy vocabulary is order-based, + // and a cursor newer than every retained frame is live-only — the old + // ring's exact semantics (a heartbeat cursor behaves the same way). + const above = b.subscribe("not-a-cursor") + expect(above.reset).toBeUndefined() + expect(above.position).toBe(1) + expect(above.replay).toEqual([]) + above.close() + }) + + test("formatID/parseCursor round-trip and reject malformed ids", () => { + expect(formatID("ep0ch", 7)).toBe("ep0ch:7") + expect(parseCursor(" ep0ch:7 ")).toEqual({ epoch: "ep0ch", seq: 7 }) + expect(parseCursor("evt_131b06142001C9nKEJ5DvEnx4G")).toBeUndefined() + expect(parseCursor("a:b")).toBeUndefined() + expect(parseCursor("ep0ch:1:2")).toBeUndefined() + }) +}) + +describe("global event replay buffer — legacy payload-id cursors (the lane's wire compat)", () => { + test("a payload-id cursor resolves to the last retained frame at or behind it and serves the frames after", () => { + const b = buffer({ maxBytes: 1 << 20, maxAgeMs: 60_000 }) + b.push(1, eventOf("test.a", "evt_p1a_leg_1"), frameOf(10)) + b.push(2, eventOf("test.b", "evt_p1a_leg_2"), frameOf(10)) + b.push(3, eventOf("test.c", "evt_p1a_leg_3"), frameOf(10)) + const sub = b.subscribe("evt_p1a_leg_2") + expect(sub.reset).toBeUndefined() + expect(sub.position).toBe(2) + expect(sub.replay.map((frame) => frame.seq)).toEqual([3]) + sub.close() + }) + + test("a payload-id cursor newer than every retained frame resolves live-only (a heartbeat cursor)", () => { + const b = buffer({ maxBytes: 1 << 20, maxAgeMs: 60_000 }) + b.push(1, eventOf("test.a", "evt_p1a_leg_1"), frameOf(10)) + const sub = b.subscribe("evt_p1a_leg_9") + expect(sub.reset).toBeUndefined() + expect(sub.position).toBe(1) + expect(sub.replay).toEqual([]) + sub.close() + }) + + test("a payload-id cursor older than every retained frame resets as evicted", () => { + const b = buffer({ maxBytes: 1 << 20, maxAgeMs: 60_000 }) + b.push(1, eventOf("test.a", "evt_p1a_leg_1"), frameOf(10)) + const sub = b.subscribe("evt_0".padEnd(35, "0")) + expect(sub.reset).toBe("evicted") + sub.close() + }) +}) diff --git a/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-event-sequencing.test.ts b/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-event-sequencing.test.ts new file mode 100644 index 00000000..57ce7bf6 --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-event-sequencing.test.ts @@ -0,0 +1,268 @@ +// #1806 P1a (ported onto #1264/#1798 machinery) — the instance /event route +// now stamps its SSE id line with the bus's `:`: a new-vocabulary +// cursor replays the gap through the shared buffer, an unservable cursor gets +// the lane's explicit reset sentinel (`state.resync`), and the client's +// existing payload-id cursors keep being served (no client change rides with +// the port). The data payload keeps its `evt_` id — only the id line changes. +import { afterEach, describe, expect } from "bun:test" +import { Effect, Queue, Stream } from "effect" +import { HttpClient, HttpClientRequest } from "effect/unstable/http" +import { EventPaths } from "../../src/server/routes/instance/httpapi/groups/event" +import { GlobalBus } from "../../src/bus/global" +import { GlobalEventReplay } from "../../src/bus/global-replay" +import { makeSseTickTelemetry, resetSseTickTelemetry, setSseTickTelemetry } from "../../src/server/shared/sse-tick-telemetry" +import { resetDatabase } from "../fixture/db" +import { disposeAllInstances, TestInstance } from "../fixture/fixture" +import { testEffect } from "../lib/effect" +import { httpApiLayer, requestInDirectory } from "./httpapi-layer" + +const it = testEffect(httpApiLayer) + +const ID = /^([A-Za-z0-9_-]+):(\d+)$/ + +type Frame = { id: string | undefined; type: string; payloadID: string | undefined } + +// The shared httpapi-layer request helper drops the URL's query (setUrl keeps +// only the pathname), so the query-carrying carrier dials through its own +// request — the directory rides the proven x-opencode-directory header. +const openFrames = (input: { directory: string; query?: string; headers?: Record }) => + Effect.gen(function* () { + const url = new URL(`${EventPaths.event}${input.query ?? ""}`, "http://localhost") + const response = yield* HttpClientRequest.fromWeb( + new Request(url, { headers: { "x-opencode-directory": input.directory, ...input.headers } }), + ).pipe( + HttpClientRequest.setUrl(url.pathname + url.search), + HttpClient.execute, + ) + const frames = yield* Queue.unbounded() + // SSE frames can straddle body chunks — accumulate and split on the + // frame boundary, keeping the trailing partial block buffered. + const state = { buffer: "" } + yield* response.stream.pipe( + Stream.decodeText(), + Stream.map((chunk: string) => { + state.buffer += chunk + const blocks = state.buffer.split("\n\n") + state.buffer = blocks.pop() ?? "" + return blocks + }), + Stream.flatMap((blocks: Array) => Stream.fromArray(blocks)), + Stream.filter((block: string) => block.includes("data:")), + Stream.mapEffect((block: string) => + Effect.sync(() => { + const id = block.split("\n").find((line) => line.startsWith("id: "))?.slice(4) + const data = block + .split("\n") + .find((line) => line.startsWith("data: ")) + ?.slice(6) + if (data === undefined) return undefined + const payload = JSON.parse(data) as { id?: string; type: string } + return { id, type: payload.type, payloadID: payload.id } satisfies Frame + }), + ), + Stream.filter((frame): frame is Frame => frame !== undefined), + Stream.runForEach((frame) => Queue.offer(frames, frame)), + Effect.forkScoped, + ) + return frames + }) + +const createSession = (directory: string, title: string) => + Effect.gen(function* () { + const created = yield* requestInDirectory("/session", directory, { + method: "POST", + body: JSON.stringify({ title }), + }) + if (created.status !== 200) return yield* Effect.die(new Error(`session create failed: ${created.status}`)) + return title + }) + +const nextFrame = (frames: Queue.Dequeue, type: string) => + Effect.gen(function* () { + const deadline = Date.now() + 10_000 + for (;;) { + const frame = yield* Queue.take(frames).pipe( + Effect.timeoutOrElse({ + duration: Math.max(50, deadline - Date.now()), + orElse: () => Effect.succeed(undefined), + }), + ) + if (frame === undefined) return yield* Effect.die(new Error(`timed out waiting for ${type}`)) + if (frame.type === type) return frame + } + }) + +afterEach(async () => { + await disposeAllInstances() + await resetDatabase() + resetSseTickTelemetry() +}) + +describe("instance event stream — epoch:seq stamping (#1806 P1a)", () => { + it.instance( + "data frames carry an : id line while the payload keeps its evt id", + () => + Effect.gen(function* () { + const { directory } = yield* TestInstance + const frames = yield* openFrames({ directory }) + const connected = yield* nextFrame(frames, "server.connected") + // Control frames never consume a sequence; their id line is the + // connection's current position, still :-shaped. + expect(connected.id).toMatch(ID) + + yield* createSession(directory, "p1a-stamp") + const created = yield* nextFrame(frames, "session.created") + expect(created.id).toMatch(ID) + const parsed = ID.exec(created.id!)! + expect(parsed[1]).toBe(GlobalBus.epoch) + expect(Number(parsed[2])).toBeGreaterThan(0) + // The payload's id is untouched — the client's dedup and cursor + // vocabulary live there, and nothing about the data line changes. + expect(created.payloadID).toMatch(/^evt_/) + expect(created.payloadID).not.toBe(created.id) + }), + { git: true, config: { formatter: false, lsp: false } }, + ) +}) + +describe("instance event stream — cursor resume against the new ids (#1806 P1a)", () => { + // The query carrier (?lastEventID=) cannot be exercised against the bare + // /event mount: it 400s at the schema layer on its own DECLARED query + // fields (directory/lastEventID), while an undeclared param passes — + // measured byte-identically against this branch's HEAD handler and the + // ported one (the route's groups/middleware are untouched base files), so + // it is a pre-existing mount quirk, not a port regression. Both carriers + // ARE proven against the ported cursor machinery on the /global/event + // route (httpapi-global-event.test.ts); the standard Last-Event-ID header + // — the carrier a browser EventSource sends — is proven here. + it.instance( + "replays the gap for a Last-Event-ID header carrying the new : id", + () => + Effect.gen(function* () { + const { directory } = yield* TestInstance + const live = yield* openFrames({ directory }) + yield* nextFrame(live, "server.connected") + yield* createSession(directory, "p1a-resume-1") + const first = yield* nextFrame(live, "session.created") + yield* createSession(directory, "p1a-resume-2") + yield* nextFrame(live, "session.created") + + // reconnect at the first session.created's position + const resumed = yield* openFrames({ directory, headers: { "last-event-id": first.id! } }) + const connected = yield* nextFrame(resumed, "server.connected") + expect(connected.id).toBe(first.id) + + // exactly the frames after the cursor: the second session, not the first + const resumedFrames: Array = [] + for (let i = 0; i < 9; i++) { + const frame = yield* Queue.take(resumed).pipe( + Effect.timeoutOrElse({ duration: "3 seconds", orElse: () => Effect.succeed(undefined) }), + ) + resumedFrames.push(frame) + if (frame === undefined) break + } + let sawFirst = false + for (const frame of resumedFrames) { + if (frame === undefined) continue + if (frame.type === "session.created" && frame.payloadID === first.payloadID) sawFirst = true + } + expect(sawFirst).toBe(false) + expect(resumedFrames.filter((f) => f?.type === "session.created").length).toBe(1) + + // a later live event flows exactly once through the resumed connection + yield* createSession(directory, "p1a-resume-3") + const later = yield* nextFrame(resumed, "session.created") + expect(later.payloadID).not.toBe(first.payloadID) + }), + { git: true, config: { formatter: false, lsp: false } }, + ) + + it.instance( + "a legacy payload-id cursor (the lane's client vocabulary) keeps being served", + () => + Effect.gen(function* () { + const { directory } = yield* TestInstance + const live = yield* openFrames({ directory }) + yield* nextFrame(live, "server.connected") + yield* createSession(directory, "p1a-legacy-1") + const first = yield* nextFrame(live, "session.created") + yield* createSession(directory, "p1a-legacy-2") + + // the existing client sends the last payload id it applied + const resumed = yield* openFrames({ directory, headers: { "last-event-id": first.payloadID! } }) + yield* nextFrame(resumed, "server.connected") + let resync = false + let second = false + for (let i = 0; i < 6 && !(resync || second); i++) { + const frame = yield* Queue.take(resumed).pipe( + Effect.timeoutOrElse({ duration: "3 seconds", orElse: () => Effect.succeed(undefined) }), + ) + if (frame === undefined) break + if (frame.type === "state.resync") resync = true + if (frame.type === "session.created" && frame.payloadID !== first.payloadID) second = true + } + expect(resync).toBe(false) + expect(second).toBe(true) + }), + { git: true, config: { formatter: false, lsp: false } }, + ) +}) + +describe("instance event stream — the reset contract (#1806 P1a)", () => { + const unservable: Record string> = { + "a cursor from a foreign epoch (an engine restart)": () => `x${GlobalBus.epoch}:1`, + "a cursor ahead of the stream": () => `${GlobalBus.epoch}:${GlobalEventReplay.current + 1000}`, + "a payload id older than the buffer's retention": () => "evt_0".padEnd(35, "0"), + } + for (const [name, cursorFor] of Object.entries(unservable)) { + it.instance( + `${name} yields the explicit state.resync sentinel — never a silent gap`, + () => + Effect.gen(function* () { + const { directory } = yield* TestInstance + yield* createSession(directory, "p1a-reset-before") + const resumed = yield* openFrames({ directory, headers: { "last-event-id": cursorFor() } }) + yield* nextFrame(resumed, "server.connected") + const frame = yield* Queue.take(resumed).pipe( + Effect.timeoutOrElse({ duration: "3 seconds", orElse: () => Effect.succeed(undefined) }), + ) + expect(frame?.type).toBe("state.resync") + + // the connection is live after the reset: later events still arrive + yield* createSession(directory, "p1a-reset-after") + const after = yield* nextFrame(resumed, "session.created") + expect(after.id).toMatch(ID) + }), + { git: true, config: { formatter: false, lsp: false } }, + ) + } +}) + +describe("instance event stream — liveness machinery through the port (#1798/#1804)", () => { + it.instance( + "heartbeats keep their cadence through the ported route and SSE-TICK lines still flow", + () => + Effect.gen(function* () { + const lines: Array = [] + const telemetry = makeSseTickTelemetry({ + cadenceMs: 100, + watchdogIntervalMs: 2_000, + probeIntervalMs: 500, + sink: (line) => lines.push(line), + }) + setSseTickTelemetry(telemetry) + + const { directory } = yield* TestInstance + const frames = yield* openFrames({ directory }) + yield* nextFrame(frames, "server.connected") + const beat = yield* nextFrame(frames, "server.heartbeat") + expect(beat.id).toMatch(ID) + + const tickLines = lines.filter((line) => line.startsWith("SSE-TICK")) + expect(tickLines.length).toBeGreaterThanOrEqual(1) + expect(tickLines[0]).toMatch(/^SSE-TICK route=\/event conn=\S+ cadence_ms=100 /) + expect(lines.some((line) => line.startsWith("SELF-STALL"))).toBe(false) + }), + { git: true, config: { formatter: false, lsp: false } }, + ) +}) diff --git a/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-global-event.test.ts b/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-global-event.test.ts new file mode 100644 index 00000000..cf02173a --- /dev/null +++ b/packages/app-bundle/overlay/packages/opencode/test/server/httpapi-global-event.test.ts @@ -0,0 +1,548 @@ +// #1776 (ported, #1806 P1a) — the global event stream is sequenced and +// replayable: every frame carries an `:` SSE id, a resume cursor +// replays the gap, an unservable cursor gets an explicit reset, heartbeats +// carry time + seq, a client abort releases its subscription, and the lane's +// bounded-write + tick-telemetry machinery covers the route (SSE-TICK lines +// flow, a stalled write drops, never SELF-STALLs). Sync copies are opt-in per +// subscription (#1778's subscription flag rides with the port; main's +// patch-body wire cuts are P3 and stay out). +// +// Lane adaptation vs main's suite: a cursor that is not `:` (the +// lane's client tracks payload ids) resolves against retained frames instead +// of resetting — wire compatibility requires the existing client's cursors +// to keep being served losslessly. +import { NodeHttpServer } from "@effect/platform-node" +import Http from "node:http" +import { afterEach, describe, expect } from "bun:test" +import { Clock, Context, Effect, Fiber, Layer, Logger, Option, Queue, References, Scope } from "effect" +import * as TestClock from "effect/testing/TestClock" +import { HttpRouter, HttpServer } from "effect/unstable/http" +import { HttpApiBuilder } from "effect/unstable/httpapi" +import { Auth } from "../../src/auth" +import { GlobalBus, type GlobalEvent } from "../../src/bus/global" +import { GlobalEventReplay } from "../../src/bus/global-replay" +import { Config } from "../../src/config/config" +import { Installation } from "../../src/installation" +import { MoveSession } from "@opencode-ai/core/control-plane/move-session" +import { ServerAuth } from "../../src/server/auth" +import { RootHttpApi } from "../../src/server/routes/instance/httpapi/api" +import { GlobalPaths } from "../../src/server/routes/instance/httpapi/groups/global" +import { controlHandlers } from "../../src/server/routes/instance/httpapi/handlers/control" +import { controlPlaneHandlers } from "../../src/server/routes/instance/httpapi/handlers/control-plane" +import { globalHandlers } from "../../src/server/routes/instance/httpapi/handlers/global" +import { authorizationLayer } from "../../src/server/routes/instance/httpapi/middleware/authorization" +import { schemaErrorLayer } from "../../src/server/routes/instance/httpapi/middleware/schema-error" +import { resetSseTickTelemetry, setSseTickTelemetry, makeSseTickTelemetry } from "../../src/server/shared/sse-tick-telemetry" +import { testEffect } from "../lib/effect" + +const apiLayer = HttpRouter.serve( + HttpApiBuilder.layer(RootHttpApi).pipe( + Layer.provide([controlHandlers, controlPlaneHandlers, globalHandlers]), + Layer.provide([authorizationLayer, schemaErrorLayer]), + // Raw HttpApi routes expose an opaque handler context at the request boundary. + // oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion + HttpRouter.provideRequest(Layer.succeedContext(Context.empty() as Context.Context)), + ), + { disableListenLog: true, disableLogger: true }, +).pipe( + // Bound to 127.0.0.1 — the address the raw fetch client dials — not + // `NodeHttpServer.layerTest`'s wildcard. The kernel only skips ephemeral ports + // already bound on the SAME address, so a wildcard bind can share its port + // with another process's 127.0.0.1 listener, which then answers our requests. + Layer.provideMerge(NodeHttpServer.layer(Http.createServer, { host: "127.0.0.1", port: 0 })), + Layer.provide(Layer.mock(Auth.Service)({})), + Layer.provide(Layer.mock(Config.Service)({})), + Layer.provide(Layer.mock(MoveSession.Service)({})), + Layer.provide(Layer.mock(Installation.Service)({})), + Layer.provide(ServerAuth.Config.configLayer({ password: Option.none(), username: "opencode" })), + Layer.provide( + Logger.layer([ + Logger.make((options) => { + logs.push({ + message: ([] as unknown[]).concat(options.message).join(" "), + annotations: { ...options.fiber.getRef(References.CurrentLogAnnotations) }, + }) + }), + ]), + ), +) +const it = testEffect(apiLayer) + +// The server's own log lines (the stream's connect/disconnect census). +const logs: { message: string; annotations: Record }[] = [] +const logged = (message: string) => logs.filter((line) => line.message === message) + +/** Poll until `probe` holds (true), or give up after `ms` (false). */ +const eventually = (probe: () => boolean, ms = 3_000) => + Effect.gen(function* () { + while (!probe()) yield* Effect.sleep("20 millis") + return true + }).pipe(Effect.timeoutOrElse({ duration: ms, orElse: () => Effect.succeed(false) })) + +const ID = /^([A-Za-z0-9_-]+):(\d+)$/ + +/** One SSE frame; `raw` is its `data` exactly as sent, `bytes` the whole frame on the wire. */ +type Frame = { id: string | undefined; event: string | undefined; data: any; raw: string; bytes: number } + +/** Split an SSE id into its epoch and sequence (fails the test if malformed). */ +function idOf(frame: Frame) { + const match = ID.exec(frame.id ?? "") + if (!match) + throw new Error(`frame id ${JSON.stringify(frame.id)} is not : (${JSON.stringify(frame.data)})`) + return { epoch: match[1], seq: Number(match[2]) } +} + +function parseFrame(block: string): Frame { + const frame: Frame = { + id: undefined, + event: undefined, + data: undefined, + raw: "", + bytes: Buffer.byteLength(block) + 2, + } + const data: string[] = [] + for (const line of block.split("\n")) { + if (line.startsWith("id: ")) frame.id = line.slice(4) + else if (line.startsWith("event: ")) frame.event = line.slice(7) + else if (line.startsWith("data: ")) data.push(line.slice(6)) + } + frame.raw = data.join("\n") + frame.data = data.length ? JSON.parse(frame.raw) : undefined + return frame +} + +/** Open a raw SSE connection to the in-process server's global stream. */ +const connect = (input: { headers?: Record; query?: string } = {}) => + Effect.gen(function* () { + // The router is attached while the layer builds, so the server routes from + // the first request; dial exactly the address it is bound to. + const url = HttpServer.formatAddress((yield* HttpServer.HttpServer).address) + const frames = yield* Queue.unbounded() + const seen: string[] = [] + let ended = "" + const controller = new AbortController() + const response = yield* Effect.promise(() => + fetch(`${url}${GlobalPaths.event}${input.query ?? ""}`, { + headers: { accept: "text/event-stream", ...input.headers }, + signal: controller.signal, + }), + ) + const status = `${response.status} ${response.headers.get("content-type") ?? "-"}` + // Read the body in the background, splitting it into SSE frames. + void (async () => { + const reader = response.body!.pipeThrough(new TextDecoderStream()).getReader() + let buffer = "" + for (;;) { + const chunk = await reader.read() + if (chunk.done) return + buffer += chunk.value + for (let end = buffer.indexOf("\n\n"); end !== -1; end = buffer.indexOf("\n\n")) { + const block = buffer.slice(0, end) + buffer = buffer.slice(end + 2) + if (block.trim() === "") continue + const frame = parseFrame(block) + seen.push(`${frame.id}=${frame.data?.payload?.type}`) + Queue.offerUnsafe(frames, frame) + } + } + })().then( + () => (ended ||= `end (${status})`), + (error) => (ended ||= `${error?.name ?? "error"} (${status})`), + ) + const abort = () => controller.abort() + yield* Effect.addFinalizer(() => Effect.sync(abort)) + const next = Queue.take(frames).pipe( + Effect.timeoutOrElse({ + duration: "4 seconds", + orElse: () => + Effect.fail(new Error(`timed out waiting for frame; seen [${seen.join(", ")}], response ${ended || "open"}`)), + }), + ) + const take = (n: number) => Effect.forEach(Array.from({ length: n }), () => next) + return { next, take, abort } + }) + +const emit = (type: string, properties: Record = {}) => + GlobalBus.emit("event", { directory: "/tmp/i1776", payload: { type, properties } }) + +/** The event-v2 bridge's twin emission: a durable event, then its sync copy. */ +const twin = (id: string, type = "test.durable") => { + GlobalBus.emit("event", { directory: "/tmp/i1778", payload: { id, type, properties: {} } }) + GlobalBus.emit("event", { + directory: "/tmp/i1778", + payload: { type: "sync", syncEvent: { id, type: `${type}.1`, seq: 1, aggregateID: "agg", data: {} } }, + }) +} + +const typesOf = (frames: Frame[]) => frames.map((frame) => frame.data.payload.type) +const seqsOf = (frames: Frame[]) => frames.map((frame) => idOf(frame).seq) +/** Resolves `undefined` when no further frame arrives within `ms`. */ +const quiet = (stream: { next: Effect.Effect }, ms = 200) => + stream.next.pipe(Effect.timeoutOrElse({ duration: ms, orElse: () => Effect.succeed(undefined) })) + +describe("global event stream — sequenced ids (#1776 port)", () => { + it.live( + "every frame carries an : id shared by all subscribers; a sync copy carries its source's id", + () => + Effect.gen(function* () { + // sync copies are opt-in (#1778): both subscribers request them + const a = yield* connect({ query: "?sync=true" }) + const b = yield* connect({ query: "?sync=true" }) + const connectedA = yield* a.next + const connectedB = yield* b.next + expect(connectedA.data.payload.type).toBe("server.connected") + const start = idOf(connectedA) + expect(idOf(connectedB)).toEqual(start) + + emit("test.first") + // the event-v2 bridge's twin emission: the durable event, then its sync copy + GlobalBus.emit("event", { + directory: "/tmp/i1776", + payload: { id: "evt_i1776_durable", type: "test.durable", properties: {} }, + }) + GlobalBus.emit("event", { + directory: "/tmp/i1776", + payload: { + type: "sync", + syncEvent: { id: "evt_i1776_durable", type: "test.durable.1", seq: 1, aggregateID: "agg", data: {} }, + }, + }) + emit("test.last") + + const framesA = yield* a.take(4) + const framesB = yield* b.take(4) + expect(framesA.map((f) => f.data.payload.type)).toEqual(["test.first", "test.durable", "sync", "test.last"]) + expect(framesA.map((f) => f.id)).toEqual(framesB.map((f) => f.id)) + const seqs = framesA.map((f) => idOf(f).seq) + expect(framesA.every((f) => idOf(f).epoch === start.epoch)).toBe(true) + expect(seqs).toEqual([start.seq + 1, start.seq + 2, start.seq + 2, start.seq + 3]) + }), + ) +}) + +describe("global event stream — sync copies are opt-in (#1778 subscription flag)", () => { + it.live( + "a default subscriber gets no sync copies and its sequence stays gap-free; an opt-in subscriber gets them all", + () => + Effect.gen(function* () { + const plain = yield* connect() + // the header carrier (what the workspace control plane sends); `?sync=true` is the query twin + const synced = yield* connect({ headers: { "x-opencode-sync-copies": "true" } }) + const start = idOf(yield* plain.next) + expect(idOf(yield* synced.next)).toEqual(start) + + emit("test.first") + twin("evt_i1778_optin_a") + twin("evt_i1778_optin_b") + emit("test.last") + + const plainFrames = yield* plain.take(4) + expect(typesOf(plainFrames)).toEqual(["test.first", "test.durable", "test.durable", "test.last"]) + // filtering the copies punches no hole: a copy shares its source's sequence + expect(seqsOf(plainFrames)).toEqual([1, 2, 3, 4].map((n) => start.seq + n)) + expect(yield* quiet(plain)).toBeUndefined() + + const syncedFrames = yield* synced.take(6) + expect(typesOf(syncedFrames)).toEqual([ + "test.first", + "test.durable", + "sync", + "test.durable", + "sync", + "test.last", + ]) + expect(seqsOf(syncedFrames)).toEqual([1, 2, 2, 3, 3, 4].map((n) => start.seq + n)) + // the frames both subscribers get are the same frames + expect(syncedFrames.filter((f) => f.data.payload.type !== "sync").map((f) => f.id)).toEqual( + plainFrames.map((f) => f.id), + ) + }), + ) + + it.live("a resumed connection replays sync copies only when it requests them", () => + Effect.gen(function* () { + const a = yield* connect() + const cursor = (yield* a.next).id! + a.abort() + twin("evt_i1778_replay_a") + emit("test.between") + twin("evt_i1778_replay_b") + + const plain = yield* connect({ headers: { "last-event-id": cursor } }) + const synced = yield* connect({ headers: { "last-event-id": cursor }, query: "?sync=true" }) + const base = idOf(yield* plain.next).seq + yield* synced.next + emit("test.live") + + const plainFrames = yield* plain.take(4) + expect(typesOf(plainFrames)).toEqual(["test.durable", "test.between", "test.durable", "test.live"]) + expect(seqsOf(plainFrames)).toEqual([1, 2, 3, 4].map((n) => base + n)) + expect(yield* quiet(plain)).toBeUndefined() + + const syncedFrames = yield* synced.take(6) + expect(typesOf(syncedFrames)).toEqual([ + "test.durable", + "sync", + "test.between", + "test.durable", + "sync", + "test.live", + ]) + expect(seqsOf(syncedFrames)).toEqual([1, 1, 2, 3, 3, 4].map((n) => base + n)) + }), + ) + + it.live( + "a sync copy with no framed source keeps its own sequence and reaches every subscriber, so no hole opens", + () => + Effect.gen(function* () { + const plain = yield* connect() + const start = idOf(yield* plain.next) + emit("test.before") + GlobalBus.emit("event", { + directory: "/tmp/i1778", + payload: { + type: "sync", + syncEvent: { id: "evt_i1778_orphan", type: "test.orphan.1", seq: 1, aggregateID: "agg", data: {} }, + }, + }) + emit("test.after") + const frames = yield* plain.take(3) + expect(typesOf(frames)).toEqual(["test.before", "sync", "test.after"]) + expect(seqsOf(frames)).toEqual([1, 2, 3].map((n) => start.seq + n)) + }), + ) +}) + +describe("global event stream — resume from a cursor (#1776 port)", () => { + for (const carrier of ["Last-Event-ID header", "lastEventID query"] as const) { + it.live(`replays exactly the frames after a ${carrier} cursor, then continues live`, () => + Effect.gen(function* () { + const a = yield* connect() + yield* a.next + emit("test.one") + emit("test.two") + emit("test.three") + const [one] = yield* a.take(3) + const cursor = idOf(one) + a.abort() + + emit("test.gap") + const resume = + carrier === "Last-Event-ID header" + ? { headers: { "last-event-id": one.id! } } + : { query: `?lastEventID=${encodeURIComponent(one.id!)}` } + // events keep arriving while the reconnect is in flight + const reconnect = yield* Effect.forkChild(connect(resume)) + for (let i = 0; i < 20; i++) { + emit("test.inflight", { i }) + yield* Effect.sleep("1 millis") + } + const b = yield* Fiber.join(reconnect) + emit("test.after") + + const connected = yield* b.next + expect(connected.data.payload.type).toBe("server.connected") + expect(idOf(connected)).toEqual(cursor) + const frames = yield* b.take(2 + 1 + 20 + 1) + const types = frames.map((f) => f.data.payload.type) + expect(types).toEqual(["test.two", "test.three", "test.gap", ...Array(20).fill("test.inflight"), "test.after"]) + expect(frames.map((f) => idOf(f).seq)).toEqual(Array.from({ length: 24 }, (_, i) => cursor.seq + 1 + i)) + expect( + frames.filter((f) => f.data.payload.type === "test.inflight").map((f) => f.data.payload.properties.i), + ).toEqual(Array.from({ length: 20 }, (_, i) => i)) + // nothing further: no duplicate trails the live frames + const extra = yield* b.next.pipe( + Effect.timeoutOrElse({ duration: "200 millis", orElse: () => Effect.succeed(undefined) }), + ) + expect(extra).toBeUndefined() + }), + ) + } + + it.live( + "a legacy payload-id cursor (the lane's client vocabulary) resolves leniently and replays the gap", + () => + Effect.gen(function* () { + const a = yield* connect() + const connected = yield* a.next + expect(connected.data.payload.type).toBe("server.connected") + GlobalBus.emit("event", { directory: "/tmp/i1776", payload: { id: "evt_p1a_leg_1", type: "test.leg.a", properties: {} } }) + GlobalBus.emit("event", { directory: "/tmp/i1776", payload: { id: "evt_p1a_leg_2", type: "test.leg.b", properties: {} } }) + const second = yield* a.next + expect(second.data.payload.type).toBe("test.leg.a") + + // the lane's client sends the last payload id it applied — not the SSE id + const b = yield* connect({ query: `?lastEventID=${encodeURIComponent("evt_p1a_leg_1")}` }) + const bConnected = yield* b.next + expect(bConnected.data.payload.type).toBe("server.connected") + expect(bConnected.data.payload.type).not.toBe("server.cursor.unservable") + const replayed = yield* b.next + expect(replayed.data.payload.type).toBe("test.leg.b") + expect(yield* quiet(b)).toBeUndefined() + }), + ) +}) + +describe("global event stream — unservable cursor resets (#1776 port)", () => { + const unservable: Record Effect.Effect> = { + "a cursor from another epoch": () => Effect.succeed(`x${GlobalBus.epoch}:1`), + "a cursor ahead of the stream": () => Effect.succeed(`${GlobalBus.epoch}:${GlobalBus.seq + 1000}`), + "a cursor below every retained payload id (garbage the history cannot cover)": () => + Effect.succeed("aaa"), + "a cursor older than the buffer's oldest entry": () => + Effect.gen(function* () { + const a = yield* connect() + const cursor = (yield* a.next).id! + a.abort() + // more than the default 32 MiB byte bound: the frames after the cursor are evicted + const big = "x".repeat(1024 * 1024) + for (let i = 0; i < 33; i++) emit("test.big", { big }) + return cursor + }), + } + for (const [name, cursorFor] of Object.entries(unservable)) { + it.live(`${name} yields a reset frame first, then live frames`, () => + Effect.gen(function* () { + emit("test.before") + const cursor = yield* cursorFor() + const b = yield* connect({ headers: { "last-event-id": cursor } }) + const current = { epoch: GlobalBus.epoch, seq: GlobalBus.seq } + + const reset = yield* b.next + expect(reset.data.payload.type).toBe("server.cursor.unservable") + expect(reset.data.payload.properties).toEqual(current) + expect(idOf(reset)).toEqual(current) + const connected = yield* b.next + expect(connected.data.payload.type).toBe("server.connected") + expect(idOf(connected)).toEqual(current) + + emit("test.live") + const live = yield* b.next + expect(live.data.payload.type).toBe("test.live") + expect(idOf(live)).toEqual({ epoch: current.epoch, seq: current.seq + 1 }) + }), + ) + } +}) + +describe("global event stream — heartbeats (#1776 port)", () => { + it.effect("a heartbeat carries server time in milliseconds and the current sequence, and ids itself with it", () => + Effect.gen(function* () { + const a = yield* connect() + yield* a.next + emit("test.beat") + const live = yield* a.next + + yield* TestClock.adjust("10 seconds") + const beat = yield* a.next + expect(beat.data.payload.type).toBe("server.heartbeat") + expect(beat.data.payload.properties).toEqual({ time: yield* Clock.currentTimeMillis, seq: idOf(live).seq }) + expect(idOf(beat)).toEqual(idOf(live)) + + emit("test.after-beat") + yield* TestClock.adjust("10 seconds") + const [after, second] = yield* a.take(2) + expect(after.data.payload.type).toBe("test.after-beat") + expect(second.data.payload.type).toBe("server.heartbeat") + expect(second.data.payload.properties).toEqual({ time: yield* Clock.currentTimeMillis, seq: idOf(after).seq }) + expect(idOf(second)).toEqual(idOf(after)) + }), + ) +}) + +describe("global event stream — client abort (#1776 port)", () => { + it.live("a client that disconnects mid-stream releases its subscription and logs the disconnect", () => + Effect.gen(function* () { + // earlier tests' connections (destroyed at their end) finish releasing first + const settled = () => logged("global event connected").length === logged("global event disconnected").length + expect(yield* eventually(settled)).toBe(true) + const baseline = GlobalEventReplay.subscribers + const disconnects = logged("global event disconnected").length + const a = yield* connect() + yield* a.next + expect(GlobalEventReplay.subscribers).toBe(baseline + 1) + expect(logged("global event connected").at(-1)?.annotations).toMatchObject({ subscribers: baseline + 1 }) + + emit("test.unread") + emit("test.unread") + a.abort() + + expect(yield* eventually(() => GlobalEventReplay.subscribers === baseline)).toBe(true) + const lines = logged("global event disconnected").slice(disconnects) + expect(lines).toHaveLength(1) + expect(lines[0].annotations).toMatchObject({ subscribers: baseline }) + + // released for good: later events are queued for no one + for (let i = 0; i < 100; i++) emit("test.after-abort", { i }) + expect(GlobalEventReplay.subscribers).toBe(baseline) + }), + ) +}) + +describe("global event stream — lane liveness machinery composes (#1798/#1804)", () => { + afterEach(() => { + resetSseTickTelemetry() + }) + + it.live("the route heartbeats at the telemetry cadence and logs an SSE-TICK line per tick; never SELF-STALLs", () => + Effect.gen(function* () { + const lines: Array = [] + const telemetry = makeSseTickTelemetry({ + cadenceMs: 100, + watchdogIntervalMs: 250, + probeIntervalMs: 500, + sink: (line) => lines.push(line), + }) + setSseTickTelemetry(telemetry) + + const a = yield* connect() + yield* a.next + emit("test.tick-source") + + // At the fast test cadence the next frames are the event, then heartbeats. + const live = yield* a.next + expect(live.data.payload.type).toBe("test.tick-source") + const beat = yield* a.next + expect(beat.data.payload.type).toBe("server.heartbeat") + + const tickLines = lines.filter((line) => line.startsWith("SSE-TICK")) + expect(tickLines.length).toBeGreaterThanOrEqual(1) + // queue_depth here is the honest backlog at tick time: the global + // heartbeat offers its control frame into the event queue, so the tick + // may see one undelivered beat — the /event route emits the frame as a + // stream element and reads 0. The line's structure is the contract. + expect(tickLines[0]).toMatch( + /^SSE-TICK route=\/global\/event conn=\S+ cadence_ms=100 queue_depth=\d+ fiber=\d+ lag_ms=\d+ rss_bytes=\d+ ts_ms=\d+$/, + ) + expect(telemetry.snapshot().connections).toBe(1) + + // The stream ends (test scope) → the registry drains → the watchdog stays quiet. + a.abort() + expect(yield* eventually(() => telemetry.snapshot().connections === 0)).toBe(true) + yield* Effect.sleep("600 millis") + expect(lines.some((line) => line.startsWith("SELF-STALL"))).toBe(false) + }), + ) +}) + +// The lane keeps one observable the port must not lose: a `GlobalEvent` read +// through the buffer is the emitted object (the instance route projects the +// payload view from it without re-parsing the wire serialization). +describe("global event replay buffer — frame carries the event (#1806 lane adaptation)", () => { + it.live("a pushed frame keeps the original event for in-process subscribers", () => + Effect.gen(function* () { + const a = yield* connect() + yield* a.next + const event: GlobalEvent = { + directory: "/tmp/i1776", + payload: { id: "evt_p1a_frame_event", type: "test.frame.event", properties: { marker: true } }, + } + GlobalBus.emit("event", event) + const frame = yield* a.next + expect(frame.data.payload.type).toBe("test.frame.event") + expect(frame.data).toEqual(event) + a.abort() + }), + ) +})