From 0c9f00f8b6a4da42f229f74f884f89160ed65105 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 11:23:15 +0800 Subject: [PATCH 01/12] feat(opencode): add bounded event replay store --- packages/opencode/src/server/event-replay.ts | 162 ++++++++++++++++ .../opencode/test/server/event-replay.test.ts | 175 ++++++++++++++++++ 2 files changed, 337 insertions(+) create mode 100644 packages/opencode/src/server/event-replay.ts create mode 100644 packages/opencode/test/server/event-replay.test.ts diff --git a/packages/opencode/src/server/event-replay.ts b/packages/opencode/src/server/event-replay.ts new file mode 100644 index 000000000..a03342236 --- /dev/null +++ b/packages/opencode/src/server/event-replay.ts @@ -0,0 +1,162 @@ +import { randomUUID } from "node:crypto" + +export type GlobalEventEnvelope = { + directory?: string + project?: string + workspace?: string + payload: { + type: string + properties: unknown + } +} + +export type ReplayCursor = { + bootID: string + seq: number +} + +export type ReplayRecord = { + id: string + seq: number + createdAt: number + envelope: GlobalEventEnvelope +} + +export type ReplaySnapshot = { + bootID: string + fenceSeq: number + fenceID: string + replay: ReplayRecord[] + gap: boolean + invalidCursor: boolean + unsubscribe: () => void + releaseLiveQueue: (push: (record: ReplayRecord) => void) => void +} + +const DEFAULT_MAX_RECORDS = 2048 +const DEFAULT_MAX_AGE_MS = 5 * 60 * 1000 + +const REPLAYABLE_EVENT_TYPES = new Set([ + "question.asked", + "question.replied", + "question.rejected", + "permission.asked", + "permission.replied", + "session.created", + "session.updated", + "session.deleted", + "session.status", +]) + +export function parseReplayCursor(input: string | undefined): ReplayCursor | undefined { + if (!input) return undefined + const index = input.lastIndexOf(":") + if (index <= 0 || index === input.length - 1) return undefined + + const bootID = input.slice(0, index) + const seq = Number(input.slice(index + 1)) + if (!Number.isSafeInteger(seq) || seq < 0) return undefined + + return { bootID, seq } +} + +export function isReplayableGlobalEvent(envelope: GlobalEventEnvelope): boolean { + return REPLAYABLE_EVENT_TYPES.has(envelope.payload?.type) +} + +export class EventReplayStore { + private readonly bootID: string + private readonly maxRecords: number + private readonly maxAgeMs: number + private readonly now: () => number + private seq = 0 + private records: ReplayRecord[] = [] + private listeners = new Set<(record: ReplayRecord) => void>() + + constructor(input?: { bootID?: string; maxRecords?: number; maxAgeMs?: number; now?: () => number }) { + this.bootID = input?.bootID ?? `${Date.now().toString(36)}-${randomUUID().slice(0, 8)}` + this.maxRecords = input?.maxRecords ?? DEFAULT_MAX_RECORDS + this.maxAgeMs = input?.maxAgeMs ?? DEFAULT_MAX_AGE_MS + this.now = input?.now ?? Date.now + } + + latestID(): string { + return this.formatID(this.seq) + } + + append(envelope: GlobalEventEnvelope): ReplayRecord | undefined { + if (!isReplayableGlobalEvent(envelope)) return undefined + + const seq = ++this.seq + const record: ReplayRecord = { + id: this.formatID(seq), + seq, + createdAt: this.now(), + envelope: structuredClone(envelope), + } + + this.records.push(record) + this.prune() + + for (const listener of this.listeners) listener(record) + return record + } + + open(lastEventID: string | undefined, onLive: (record: ReplayRecord) => void): ReplaySnapshot { + const liveBuffer: ReplayRecord[] = [] + let replaying = true + + const listener = (record: ReplayRecord) => { + if (replaying) { + liveBuffer.push(record) + return + } + onLive(record) + } + + this.listeners.add(listener) + + const fenceSeq = this.seq + const fenceID = this.formatID(fenceSeq) + const parsed = parseReplayCursor(lastEventID) + const invalidCursor = !!lastEventID && (!parsed || parsed.bootID !== this.bootID || parsed.seq > fenceSeq) + const cursorSeq = invalidCursor || !parsed ? fenceSeq : parsed.seq + const earliestSeq = this.records[0]?.seq + const gap = !invalidCursor && !!parsed && earliestSeq !== undefined && parsed.seq < earliestSeq - 1 + const replay = + invalidCursor || !parsed + ? [] + : this.records.filter((record) => record.seq > cursorSeq && record.seq <= fenceSeq) + + return { + bootID: this.bootID, + fenceSeq, + fenceID, + replay, + gap, + invalidCursor, + unsubscribe: () => this.listeners.delete(listener), + releaseLiveQueue: (push) => { + replaying = false + for (const record of liveBuffer) { + if (record.seq > fenceSeq) push(record) + } + liveBuffer.length = 0 + }, + } + } + + recordsForTest(): ReplayRecord[] { + return this.records + } + + private formatID(seq: number): string { + return `${this.bootID}:${seq}` + } + + private prune() { + const cutoff = this.now() - this.maxAgeMs + while (this.records.length > this.maxRecords) this.records.shift() + while (this.records[0] && this.records[0].createdAt < cutoff) this.records.shift() + } +} diff --git a/packages/opencode/test/server/event-replay.test.ts b/packages/opencode/test/server/event-replay.test.ts new file mode 100644 index 000000000..4a6241586 --- /dev/null +++ b/packages/opencode/test/server/event-replay.test.ts @@ -0,0 +1,175 @@ +import { describe, expect, test } from "bun:test" +import { EventReplayStore, isReplayableGlobalEvent, parseReplayCursor } from "../../src/server/event-replay" + +const question = (id: string, sessionID = "ses_1") => ({ + directory: "/repo", + project: "proj", + workspace: "work", + payload: { + type: "question.asked", + properties: { id, sessionID, questions: [{ header: "H", question: "Q", options: [] }] }, + }, +}) + +const event = (type: string) => ({ + directory: "/repo", + payload: { + type, + properties: {}, + }, +}) + +describe("parseReplayCursor", () => { + test("parses boot id and numeric sequence", () => { + expect(parseReplayCursor("boot-abc:42")).toEqual({ bootID: "boot-abc", seq: 42 }) + }) + + test("rejects malformed cursors", () => { + expect(parseReplayCursor(undefined)).toBeUndefined() + expect(parseReplayCursor("")).toBeUndefined() + expect(parseReplayCursor("boot")).toBeUndefined() + expect(parseReplayCursor("boot:abc")).toBeUndefined() + expect(parseReplayCursor("boot:-1")).toBeUndefined() + }) +}) + +describe("isReplayableGlobalEvent", () => { + test("allows blocker and session state events", () => { + expect(isReplayableGlobalEvent(event("question.asked"))).toBe(true) + expect(isReplayableGlobalEvent(event("question.replied"))).toBe(true) + expect(isReplayableGlobalEvent(event("question.rejected"))).toBe(true) + expect(isReplayableGlobalEvent(event("permission.asked"))).toBe(true) + expect(isReplayableGlobalEvent(event("permission.replied"))).toBe(true) + expect(isReplayableGlobalEvent(event("session.created"))).toBe(true) + expect(isReplayableGlobalEvent(event("session.updated"))).toBe(true) + expect(isReplayableGlobalEvent(event("session.deleted"))).toBe(true) + expect(isReplayableGlobalEvent(event("session.status"))).toBe(true) + }) + + test("rejects high-volume and unrelated events", () => { + expect(isReplayableGlobalEvent(event("message.part.delta"))).toBe(false) + expect(isReplayableGlobalEvent(event("message.part.updated"))).toBe(false) + expect(isReplayableGlobalEvent(event("message.updated"))).toBe(false) + expect(isReplayableGlobalEvent(event("todo.updated"))).toBe(false) + expect(isReplayableGlobalEvent(event("lsp.updated"))).toBe(false) + expect(isReplayableGlobalEvent(event("vcs.branch.updated"))).toBe(false) + }) +}) + +describe("EventReplayStore", () => { + test("assigns monotonic ids under one boot id", () => { + const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) + const first = store.append(question("q1")) + const second = store.append(question("q2")) + + expect(first?.id).toBe("boot:1") + expect(second?.id).toBe("boot:2") + expect(store.latestID()).toBe("boot:2") + }) + + test("does not store non-replayable events", () => { + const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) + expect(store.append(event("message.part.delta"))).toBeUndefined() + expect(store.recordsForTest()).toEqual([]) + expect(store.latestID()).toBe("boot:0") + }) + + test("stores cloned envelopes", () => { + const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) + const input = question("q1") + const record = store.append(input) + ;(input.payload.properties as { id: string }).id = "mutated" + + expect((record?.envelope.payload.properties as { id: string }).id).toBe("q1") + expect((store.recordsForTest()[0].envelope.payload.properties as { id: string }).id).toBe("q1") + }) + + test("prunes by max record count", () => { + const store = new EventReplayStore({ bootID: "boot", maxRecords: 2, now: () => 1000 }) + store.append(question("q1")) + store.append(question("q2")) + store.append(question("q3")) + + expect(store.recordsForTest().map((record) => record.seq)).toEqual([2, 3]) + }) + + test("prunes by max age", () => { + let now = 1000 + const store = new EventReplayStore({ bootID: "boot", maxAgeMs: 100, now: () => now }) + store.append(question("q1")) + now = 1200 + store.append(question("q2")) + + expect(store.recordsForTest().map((record) => record.seq)).toEqual([2]) + }) + + test("replays records after cursor in ascending order", () => { + const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) + store.append(question("q1")) + store.append(question("q2")) + store.append(question("q3")) + + const opened = store.open("boot:1", () => {}) + + expect(opened.invalidCursor).toBe(false) + expect(opened.gap).toBe(false) + expect(opened.replay.map((record) => record.id)).toEqual(["boot:2", "boot:3"]) + opened.unsubscribe() + }) + + test("returns no replay for invalid boot id", () => { + const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) + store.append(question("q1")) + + const opened = store.open("old:1", () => {}) + + expect(opened.invalidCursor).toBe(true) + expect(opened.gap).toBe(false) + expect(opened.replay).toEqual([]) + expect(opened.fenceID).toBe("boot:1") + opened.unsubscribe() + }) + + test("treats a same-boot future cursor as invalid", () => { + const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) + store.append(question("q1")) + + const opened = store.open("boot:99", () => {}) + + expect(opened.invalidCursor).toBe(true) + expect(opened.replay).toEqual([]) + expect(opened.fenceID).toBe("boot:1") + opened.unsubscribe() + }) + + test("detects a gap when cursor is older than retained records", () => { + const store = new EventReplayStore({ bootID: "boot", maxRecords: 2, now: () => 1000 }) + store.append(question("q1")) + store.append(question("q2")) + store.append(question("q3")) + + const opened = store.open("boot:0", () => {}) + + expect(opened.gap).toBe(true) + expect(opened.replay.map((record) => record.id)).toEqual(["boot:2", "boot:3"]) + opened.unsubscribe() + }) + + test("buffers live records during replay and releases only records after the fence", () => { + const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) + store.append(question("q1")) + + const live: string[] = [] + const opened = store.open("boot:0", (record) => live.push(record.id)) + store.append(question("q2")) + + expect(opened.fenceID).toBe("boot:1") + expect(opened.replay.map((record) => record.id)).toEqual(["boot:1"]) + expect(live).toEqual([]) + + opened.releaseLiveQueue((record) => live.push(record.id)) + + expect(live).toEqual(["boot:2"]) + opened.unsubscribe() + }) +}) From 8c2fe8d64b8d6838ef4a0e0f06c529cfb464379b Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 11:25:01 +0800 Subject: [PATCH 02/12] feat(opencode): replay global SSE event cursors --- .../opencode/src/server/instance/global.ts | 161 +++++++++++++++++- .../test/server/global-event-replay.test.ts | 78 +++++++++ 2 files changed, 232 insertions(+), 7 deletions(-) create mode 100644 packages/opencode/test/server/global-event-replay.test.ts diff --git a/packages/opencode/src/server/instance/global.ts b/packages/opencode/src/server/instance/global.ts index 7da22212c..002277ded 100644 --- a/packages/opencode/src/server/instance/global.ts +++ b/packages/opencode/src/server/instance/global.ts @@ -14,11 +14,118 @@ import { Log } from "@opencode-ai/core/util/log" import { lazy } from "../../util/lazy" import { Config } from "../../config/config" import { errors } from "../error" +import { EventReplayStore, type GlobalEventEnvelope, type ReplayRecord } from "../event-replay" const log = Log.create({ service: "server" }) export const GlobalDisposedEvent = BusEvent.define("global.disposed", z.object({})) +type SsePacket = { + id?: string + data: string + replaySeq?: number +} + +export type GlobalEventReplayPacket = SsePacket + +function packetForEnvelope(envelope: GlobalEventEnvelope, id?: string): SsePacket { + return { + id, + data: JSON.stringify(envelope), + } +} + +function packetForRecord(record: ReplayRecord): SsePacket { + return { + id: record.id, + replaySeq: record.seq, + data: JSON.stringify(record.envelope), + } +} + +export function createGlobalEventReplayBridge(input?: { replayStore?: EventReplayStore }) { + const replayStore = input?.replayStore ?? new EventReplayStore() + const listeners = new Set<(packet: GlobalEventReplayPacket) => void>() + + return { + replayStore, + append(event: GlobalEventEnvelope) { + const record = replayStore.append(event) + const packet = record ? packetForRecord(record) : packetForEnvelope(event) + for (const listener of listeners) listener(packet) + return packet + }, + subscribe(listener: (packet: GlobalEventReplayPacket) => void) { + listeners.add(listener) + return () => listeners.delete(listener) + }, + } +} + +export function openGlobalEventReplayConnection(input: { + bridge: ReturnType + lastEventID?: string + push: (packet: GlobalEventReplayPacket) => void +}) { + const liveBuffer: GlobalEventReplayPacket[] = [] + let replaying = true + const pushLive = (packet: GlobalEventReplayPacket) => { + if (replaying) { + liveBuffer.push(packet) + return + } + input.push(packet) + } + + const unsubscribeLive = input.bridge.subscribe(pushLive) + const opened = input.bridge.replayStore.open(input.lastEventID, () => {}) + + input.push({ + id: input.lastEventID ? undefined : opened.fenceID, + data: JSON.stringify({ + payload: { + type: "server.connected", + properties: {}, + }, + }), + }) + + for (const record of opened.replay) { + input.push(packetForRecord(record)) + } + + if (opened.invalidCursor || opened.gap) { + input.push({ + id: opened.fenceID, + data: JSON.stringify({ + payload: { + type: "server.connected", + properties: {}, + }, + }), + }) + } + + opened.releaseLiveQueue(() => {}) + replaying = false + for (const packet of liveBuffer) { + if (packet.replaySeq !== undefined && packet.replaySeq <= opened.fenceSeq) continue + input.push(packet) + } + liveBuffer.length = 0 + + return () => { + opened.unsubscribe() + unsubscribeLive() + } +} + +const globalEventReplay = createGlobalEventReplayBridge() + +GlobalBus.on("event", (event) => { + globalEventReplay.append(event as GlobalEventEnvelope) +}) + async function streamEvents(c: Context, subscribe: (q: AsyncQueue) => () => void) { return streamSSE(c, async (stream) => { const q = new AsyncQueue() @@ -69,6 +176,52 @@ async function streamEvents(c: Context, subscribe: (q: AsyncQueue }) } +async function streamGlobalEvents(c: Context) { + const lastEventID = c.req.header("Last-Event-ID") ?? c.req.header("last-event-id") ?? undefined + + return streamSSE(c, async (stream) => { + const q = new AsyncQueue() + let done = false + const unsubscribe = openGlobalEventReplayConnection({ + bridge: globalEventReplay, + lastEventID, + push: (packet) => q.push(packet), + }) + + const heartbeat = setInterval(() => { + q.push({ + data: JSON.stringify({ + payload: { + type: "server.heartbeat", + properties: {}, + }, + }), + }) + }, 10_000) + + const stop = () => { + if (done) return + done = true + clearInterval(heartbeat) + unsubscribe() + q.push(null) + log.info("global event disconnected") + } + + stream.onAbort(stop) + + try { + for await (const packet of q) { + if (packet === null) return + const { replaySeq: _replaySeq, ...sse } = packet + await stream.writeSSE(sse) + } + } finally { + stop() + } + }) +} + export const GlobalRoutes = lazy(() => new Hono() .get( @@ -126,13 +279,7 @@ export const GlobalRoutes = lazy(() => c.header("X-Accel-Buffering", "no") c.header("X-Content-Type-Options", "nosniff") - return streamEvents(c, (q) => { - async function handler(event: any) { - q.push(JSON.stringify(event)) - } - GlobalBus.on("event", handler) - return () => GlobalBus.off("event", handler) - }) + return streamGlobalEvents(c) }, ) .get( diff --git a/packages/opencode/test/server/global-event-replay.test.ts b/packages/opencode/test/server/global-event-replay.test.ts new file mode 100644 index 000000000..e0520e965 --- /dev/null +++ b/packages/opencode/test/server/global-event-replay.test.ts @@ -0,0 +1,78 @@ +import { describe, expect, test } from "bun:test" +import { EventReplayStore } from "../../src/server/event-replay" +import { createGlobalEventReplayBridge, openGlobalEventReplayConnection } from "../../src/server/instance/global" + +const envelope = (type: string, id = type) => ({ + directory: "/repo", + payload: { + type, + properties: type === "question.asked" ? { id, sessionID: "ses_1", questions: [] } : { id }, + }, +}) + +describe("createGlobalEventReplayBridge", () => { + test("live broadcasts replayable events with ids and non-replayable events without ids", () => { + const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot" }) }) + const packets: Array<{ id?: string; replaySeq?: number; data: string }> = [] + const unsubscribe = bridge.subscribe((packet) => packets.push(packet)) + + bridge.append(envelope("question.asked", "q1")) + bridge.append(envelope("message.part.delta", "delta1")) + + expect(packets[0].id).toBe("boot:1") + expect(packets[0].replaySeq).toBe(1) + expect(JSON.parse(packets[0].data).payload.type).toBe("question.asked") + expect(packets[1].id).toBeUndefined() + expect(packets[1].replaySeq).toBeUndefined() + expect(JSON.parse(packets[1].data).payload.type).toBe("message.part.delta") + unsubscribe() + }) + + test("connection seeds a fresh cursor and replays missed records after Last-Event-ID", () => { + const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot" }) }) + bridge.append(envelope("question.asked", "q1")) + const fresh: Array<{ id?: string; replaySeq?: number; data: string }> = [] + + openGlobalEventReplayConnection({ bridge, push: (packet) => fresh.push(packet) })() + + expect(fresh[0].id).toBe("boot:1") + + bridge.append(envelope("question.replied", "q1")) + const reconnect: Array<{ id?: string; replaySeq?: number; data: string }> = [] + + openGlobalEventReplayConnection({ + bridge, + lastEventID: "boot:1", + push: (packet) => reconnect.push(packet), + })() + + expect(reconnect.map((packet) => JSON.parse(packet.data).payload.type)).toEqual([ + "server.connected", + "question.replied", + ]) + expect(reconnect[0].id).toBeUndefined() + expect(reconnect[1].id).toBe("boot:2") + }) + + test("connection signals a gap even when partial replay records are available", () => { + const bridge = createGlobalEventReplayBridge({ + replayStore: new EventReplayStore({ bootID: "boot", maxRecords: 1 }), + }) + bridge.append(envelope("question.asked", "q1")) + bridge.append(envelope("question.replied", "q1")) + const packets: Array<{ id?: string; replaySeq?: number; data: string }> = [] + + openGlobalEventReplayConnection({ + bridge, + lastEventID: "boot:0", + push: (packet) => packets.push(packet), + })() + + expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual([ + "server.connected", + "question.replied", + "server.connected", + ]) + expect(packets.at(-1)?.id).toBe("boot:2") + }) +}) From 021e86d2d89cd7058ca3cf4236c23fe9b86786cf Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 11:26:09 +0800 Subject: [PATCH 03/12] fix(app): persist SSE replay cursors --- packages/app/src/context/global-sdk.tsx | 6 ++++ .../src/context/global-sdk/sse-cursor.test.ts | 30 +++++++++++++++++++ .../app/src/context/global-sdk/sse-cursor.ts | 19 ++++++++++++ 3 files changed, 55 insertions(+) create mode 100644 packages/app/src/context/global-sdk/sse-cursor.test.ts create mode 100644 packages/app/src/context/global-sdk/sse-cursor.ts diff --git a/packages/app/src/context/global-sdk.tsx b/packages/app/src/context/global-sdk.tsx index 5935e3819..eff9b504c 100644 --- a/packages/app/src/context/global-sdk.tsx +++ b/packages/app/src/context/global-sdk.tsx @@ -8,6 +8,7 @@ import { createSdkForServer } from "@/utils/server" import { useLanguage } from "./language" import { usePlatform } from "./platform" import { useServer } from "./server" +import { createSseCursor } from "./global-sdk/sse-cursor" const abortError = z.object({ name: z.literal("AbortError"), @@ -111,6 +112,7 @@ export const { use: useGlobalSDK, provider: GlobalSDKProvider } = createSimpleCo const HEARTBEAT_TIMEOUT_MS = 15_000 let lastEventAt = Date.now() let heartbeat: ReturnType | undefined + const replayCursor = createSseCursor() const resetHeartbeat = () => { lastEventAt = Date.now() if (heartbeat) clearTimeout(heartbeat) @@ -139,6 +141,10 @@ export const { use: useGlobalSDK, provider: GlobalSDKProvider } = createSimpleCo try { const events = await eventSdk.global.event({ signal: attempt.signal, + headers: replayCursor.headers(), + onSseEvent: (event) => { + replayCursor.update(event.id) + }, onSseError: (error) => { if (aborted(error)) return if (streamErrorLogged) return diff --git a/packages/app/src/context/global-sdk/sse-cursor.test.ts b/packages/app/src/context/global-sdk/sse-cursor.test.ts new file mode 100644 index 000000000..8a9a5de10 --- /dev/null +++ b/packages/app/src/context/global-sdk/sse-cursor.test.ts @@ -0,0 +1,30 @@ +import { describe, expect, test } from "bun:test" +import { createSseCursor } from "./sse-cursor" + +describe("createSseCursor", () => { + test("starts without a cursor", () => { + const cursor = createSseCursor() + expect(cursor.current()).toBeUndefined() + expect(cursor.headers()).toBeUndefined() + }) + + test("stores the latest non-empty event id", () => { + const cursor = createSseCursor() + cursor.update(undefined) + cursor.update("") + cursor.update("boot:1") + cursor.update("boot:2") + + expect(cursor.current()).toBe("boot:2") + }) + + test("builds Last-Event-ID headers when a cursor exists", () => { + const cursor = createSseCursor() + cursor.update("boot:7") + + const headers = cursor.headers() + + expect(headers).toBeInstanceOf(Headers) + expect(headers?.get("Last-Event-ID")).toBe("boot:7") + }) +}) diff --git a/packages/app/src/context/global-sdk/sse-cursor.ts b/packages/app/src/context/global-sdk/sse-cursor.ts new file mode 100644 index 000000000..698c0e86e --- /dev/null +++ b/packages/app/src/context/global-sdk/sse-cursor.ts @@ -0,0 +1,19 @@ +export function createSseCursor() { + let value: string | undefined + + return { + current() { + return value + }, + update(id: string | undefined) { + if (!id) return + value = id + }, + headers() { + if (!value) return undefined + const headers = new Headers() + headers.set("Last-Event-ID", value) + return headers + }, + } +} From 0002784f33df6dc83cf2663e98afff236d38ee93 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 11:28:09 +0800 Subject: [PATCH 04/12] fix(app): ignore stale blocker requests --- packages/app/src/context/global-sync.tsx | 4 ++ .../blocker-terminal-cache.test.ts | 51 ++++++++++++++++ .../global-sync/blocker-terminal-cache.ts | 50 +++++++++++++++ .../context/global-sync/event-reducer.test.ts | 61 +++++++++++++++++++ .../src/context/global-sync/event-reducer.ts | 6 ++ 5 files changed, 172 insertions(+) create mode 100644 packages/app/src/context/global-sync/blocker-terminal-cache.test.ts create mode 100644 packages/app/src/context/global-sync/blocker-terminal-cache.ts diff --git a/packages/app/src/context/global-sync.tsx b/packages/app/src/context/global-sync.tsx index 2528ffcdf..9396b4f8a 100644 --- a/packages/app/src/context/global-sync.tsx +++ b/packages/app/src/context/global-sync.tsx @@ -16,6 +16,7 @@ import { Persist, persisted } from "@/utils/persist" import type { InitError } from "../pages/error" import { useGlobalSDK } from "./global-sdk" import { bootstrapDirectory, bootstrapGlobal, clearProviderRev } from "./global-sync/bootstrap" +import { createBlockerTerminalCache } from "./global-sync/blocker-terminal-cache" import { createChildStoreManager } from "./global-sync/child-store" import { applyDirectoryEvent, applyGlobalEvent, cleanupDroppedSessionCaches } from "./global-sync/event-reducer" import { createRefreshQueue } from "./global-sync/queue" @@ -57,6 +58,7 @@ function createGlobalSync() { const booting = new Map>() const sessionLoads = new Map>() const sessionMeta = new Map() + const blockerTerminals = createBlockerTerminalCache() const [projectCache, setProjectCache, projectInit] = persisted( Persist.global("globalSync.project", ["globalSync.project.v1"]), @@ -166,6 +168,7 @@ function createGlobalSync() { onDispose: (directory) => { queue.clear(directory) sessionMeta.delete(directory) + blockerTerminals.clearDirectory(directory) sdkCache.delete(directory) clearProviderRev(directory) clearSessionPrefetchDirectory(directory) @@ -330,6 +333,7 @@ function createGlobalSync() { setStore, push: queue.push, setSessionTodo, + blockerTerminals, vcsCache: children.vcsCache.get(directory), loadLsp: () => { void sdkFor(directory) diff --git a/packages/app/src/context/global-sync/blocker-terminal-cache.test.ts b/packages/app/src/context/global-sync/blocker-terminal-cache.test.ts new file mode 100644 index 000000000..91b1fb057 --- /dev/null +++ b/packages/app/src/context/global-sync/blocker-terminal-cache.test.ts @@ -0,0 +1,51 @@ +import { describe, expect, test } from "bun:test" +import { createBlockerTerminalCache } from "./blocker-terminal-cache" + +describe("createBlockerTerminalCache", () => { + test("marks and finds terminal blocker ids by kind, directory, session, and request", () => { + const cache = createBlockerTerminalCache({ now: () => 1000 }) + + cache.mark("question", "/repo", "ses_1", "q1") + + expect(cache.has("question", "/repo", "ses_1", "q1")).toBe(true) + expect(cache.has("permission", "/repo", "ses_1", "q1")).toBe(false) + expect(cache.has("question", "/other", "ses_1", "q1")).toBe(false) + expect(cache.has("question", "/repo", "ses_2", "q1")).toBe(false) + }) + + test("expires old entries by ttl", () => { + let now = 1000 + const cache = createBlockerTerminalCache({ ttlMs: 100, now: () => now }) + + cache.mark("question", "/repo", "ses_1", "q1") + now = 1200 + + expect(cache.has("question", "/repo", "ses_1", "q1")).toBe(false) + }) + + test("prunes oldest entries by max size", () => { + let now = 1000 + const cache = createBlockerTerminalCache({ max: 2, now: () => now }) + + cache.mark("question", "/repo", "ses_1", "q1") + now += 1 + cache.mark("question", "/repo", "ses_1", "q2") + now += 1 + cache.mark("question", "/repo", "ses_1", "q3") + + expect(cache.has("question", "/repo", "ses_1", "q1")).toBe(false) + expect(cache.has("question", "/repo", "ses_1", "q2")).toBe(true) + expect(cache.has("question", "/repo", "ses_1", "q3")).toBe(true) + }) + + test("clears all entries for a directory", () => { + const cache = createBlockerTerminalCache({ now: () => 1000 }) + + cache.mark("question", "/repo", "ses_1", "q1") + cache.mark("question", "/other", "ses_1", "q1") + cache.clearDirectory("/repo") + + expect(cache.has("question", "/repo", "ses_1", "q1")).toBe(false) + expect(cache.has("question", "/other", "ses_1", "q1")).toBe(true) + }) +}) diff --git a/packages/app/src/context/global-sync/blocker-terminal-cache.ts b/packages/app/src/context/global-sync/blocker-terminal-cache.ts new file mode 100644 index 000000000..8a287320f --- /dev/null +++ b/packages/app/src/context/global-sync/blocker-terminal-cache.ts @@ -0,0 +1,50 @@ +export type BlockerKind = "question" | "permission" + +type Entry = { + directory: string + createdAt: number +} + +const DEFAULT_MAX = 2048 +const DEFAULT_TTL_MS = 5 * 60 * 1000 + +export function createBlockerTerminalCache(input?: { max?: number; ttlMs?: number; now?: () => number }) { + const max = input?.max ?? DEFAULT_MAX + const ttlMs = input?.ttlMs ?? DEFAULT_TTL_MS + const now = input?.now ?? Date.now + const entries = new Map() + + const keyFor = (kind: BlockerKind, directory: string, sessionID: string, requestID: string) => + `${kind}\n${directory}\n${sessionID}\n${requestID}` + + const prune = () => { + const cutoff = now() - ttlMs + for (const [key, entry] of entries) { + if (entry.createdAt >= cutoff) continue + entries.delete(key) + } + while (entries.size > max) { + const oldest = entries.keys().next().value + if (!oldest) break + entries.delete(oldest) + } + } + + return { + mark(kind: BlockerKind, directory: string, sessionID: string, requestID: string) { + const key = keyFor(kind, directory, sessionID, requestID) + entries.delete(key) + entries.set(key, { directory, createdAt: now() }) + prune() + }, + has(kind: BlockerKind, directory: string, sessionID: string, requestID: string) { + prune() + return entries.has(keyFor(kind, directory, sessionID, requestID)) + }, + clearDirectory(directory: string) { + for (const [key, entry] of entries) { + if (entry.directory === directory) entries.delete(key) + } + }, + } +} diff --git a/packages/app/src/context/global-sync/event-reducer.test.ts b/packages/app/src/context/global-sync/event-reducer.test.ts index cfff37d2b..bacc0146c 100644 --- a/packages/app/src/context/global-sync/event-reducer.test.ts +++ b/packages/app/src/context/global-sync/event-reducer.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "bun:test" import type { Message, Part, PermissionRequest, Project, QuestionRequest, Session } from "@opencode-ai/sdk/v2/client" import { createStore } from "solid-js/store" import type { State } from "./types" +import { createBlockerTerminalCache } from "./blocker-terminal-cache" import { applyDirectoryEvent, applyGlobalEvent, cleanupDroppedSessionCaches } from "./event-reducer" const rootSession = (input: { id: string; parentID?: string; archived?: number; created?: number; updated?: number }) => @@ -493,6 +494,66 @@ describe("applyDirectoryEvent", () => { expect(store.question[sessionID]?.map((x) => x.id)).toEqual(["q_1", "q_3"]) }) + test("question.replied before question.asked prevents stale ask from reopening", () => { + const blockerTerminals = createBlockerTerminalCache({ now: () => 1000 }) + const [store, setStore] = createStore(baseState()) + + applyDirectoryEvent({ + event: { type: "question.replied", properties: { sessionID: "ses_1", requestID: "q1" } }, + directory: "/repo", + store, + setStore, + push() {}, + loadLsp() {}, + blockerTerminals, + }) + + applyDirectoryEvent({ + event: { + type: "question.asked", + properties: questionRequest("q1", "ses_1"), + }, + directory: "/repo", + store, + setStore, + push() {}, + loadLsp() {}, + blockerTerminals, + }) + + expect(store.question.ses_1).toBeUndefined() + }) + + test("permission.replied before permission.asked prevents stale ask from reopening", () => { + const blockerTerminals = createBlockerTerminalCache({ now: () => 1000 }) + const [store, setStore] = createStore(baseState()) + + applyDirectoryEvent({ + event: { type: "permission.replied", properties: { sessionID: "ses_1", requestID: "perm_1" } }, + directory: "/repo", + store, + setStore, + push() {}, + loadLsp() {}, + blockerTerminals, + }) + + applyDirectoryEvent({ + event: { + type: "permission.asked", + properties: permissionRequest("perm_1", "ses_1"), + }, + directory: "/repo", + store, + setStore, + push() {}, + loadLsp() {}, + blockerTerminals, + }) + + expect(store.permission.ses_1).toBeUndefined() + }) + test("updates vcs branch in store and cache", () => { const [store, setStore] = createStore(baseState({ vcs: { branch: "main", default_branch: "main" } })) const [cacheStore, setCacheStore] = createStore({ diff --git a/packages/app/src/context/global-sync/event-reducer.ts b/packages/app/src/context/global-sync/event-reducer.ts index 351d3cdf0..09dfb9939 100644 --- a/packages/app/src/context/global-sync/event-reducer.ts +++ b/packages/app/src/context/global-sync/event-reducer.ts @@ -15,6 +15,7 @@ import type { State, VcsCache } from "./types" import { trimSessions } from "./session-trim" import { dropSessionCaches } from "./session-cache" import { diffs as list, message as clean } from "@/utils/diffs" +import type { createBlockerTerminalCache } from "./blocker-terminal-cache" const SKIP_PARTS = new Set(["patch", "step-start", "step-finish"]) @@ -95,6 +96,7 @@ export function applyDirectoryEvent(input: { loadLsp: () => void vcsCache?: VcsCache setSessionTodo?: (sessionID: string, todos: Todo[] | undefined) => void + blockerTerminals?: ReturnType }) { const event = input.event switch (event.type) { @@ -277,6 +279,7 @@ export function applyDirectoryEvent(input: { } case "permission.asked": { const permission = event.properties as PermissionRequest + if (input.blockerTerminals?.has("permission", input.directory, permission.sessionID, permission.id)) break const permissions = input.store.permission[permission.sessionID] if (!permissions) { input.setStore("permission", permission.sessionID, [permission]) @@ -298,6 +301,7 @@ export function applyDirectoryEvent(input: { } case "permission.replied": { const props = event.properties as { sessionID: string; requestID: string } + input.blockerTerminals?.mark("permission", input.directory, props.sessionID, props.requestID) const permissions = input.store.permission[props.sessionID] if (!permissions) break const result = Binary.search(permissions, props.requestID, (p) => p.id) @@ -313,6 +317,7 @@ export function applyDirectoryEvent(input: { } case "question.asked": { const question = event.properties as QuestionRequest + if (input.blockerTerminals?.has("question", input.directory, question.sessionID, question.id)) break const questions = input.store.question[question.sessionID] if (!questions) { input.setStore("question", question.sessionID, [question]) @@ -335,6 +340,7 @@ export function applyDirectoryEvent(input: { case "question.replied": case "question.rejected": { const props = event.properties as { sessionID: string; requestID: string } + input.blockerTerminals?.mark("question", input.directory, props.sessionID, props.requestID) const questions = input.store.question[props.sessionID] if (!questions) break const result = Binary.search(questions, props.requestID, (q) => q.id) From edb5b09927599d771266602af253525212ca2513 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 11:31:44 +0800 Subject: [PATCH 05/12] test(app): improve question seed diagnostics --- packages/app/e2e/actions.ts | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/packages/app/e2e/actions.ts b/packages/app/e2e/actions.ts index 90f71f2c6..f3f0019b4 100644 --- a/packages/app/e2e/actions.ts +++ b/packages/app/e2e/actions.ts @@ -790,7 +790,20 @@ export async function seedSessionQuestion( }, }) - if (!result) throw new Error("Timed out seeding question request") + if (!result) { + const [questions, status] = await Promise.all([ + sdk.question.list().then((x) => x.data ?? []).catch((error) => ({ error: String(error) })), + sdk.session.status().then((x) => x.data ?? {}).catch((error) => ({ error: String(error) })), + ]) + throw new Error( + `Timed out seeding question request: ${JSON.stringify({ + sessionID: input.sessionID, + wantedHeader: first.header, + questions, + status, + })}`, + ) + } return { id: result.id } } From 8cb62d87785052672a5ded0d4a038ed8313ebcd4 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 11:52:51 +0800 Subject: [PATCH 06/12] fix(opencode): avoid refresh on valid SSE replay --- .../app/src/context/global-sdk/sse-cursor.ts | 1 + .../global-sync/blocker-terminal-cache.ts | 2 +- packages/opencode/src/server/event-replay.ts | 52 +++++------ .../opencode/src/server/instance/global.ts | 47 +++++----- .../opencode/test/server/event-replay.test.ts | 53 +++++++---- .../test/server/global-event-replay.test.ts | 91 ++++++++++++++++--- 6 files changed, 162 insertions(+), 84 deletions(-) diff --git a/packages/app/src/context/global-sdk/sse-cursor.ts b/packages/app/src/context/global-sdk/sse-cursor.ts index 698c0e86e..c33668c51 100644 --- a/packages/app/src/context/global-sdk/sse-cursor.ts +++ b/packages/app/src/context/global-sdk/sse-cursor.ts @@ -7,6 +7,7 @@ export function createSseCursor() { }, update(id: string | undefined) { if (!id) return + // SSE ids come from the server replay layer; keep this helper transport-only. value = id }, headers() { diff --git a/packages/app/src/context/global-sync/blocker-terminal-cache.ts b/packages/app/src/context/global-sync/blocker-terminal-cache.ts index 8a287320f..994185980 100644 --- a/packages/app/src/context/global-sync/blocker-terminal-cache.ts +++ b/packages/app/src/context/global-sync/blocker-terminal-cache.ts @@ -15,7 +15,7 @@ export function createBlockerTerminalCache(input?: { max?: number; ttlMs?: numbe const entries = new Map() const keyFor = (kind: BlockerKind, directory: string, sessionID: string, requestID: string) => - `${kind}\n${directory}\n${sessionID}\n${requestID}` + JSON.stringify([kind, directory, sessionID, requestID]) const prune = () => { const cutoff = now() - ttlMs diff --git a/packages/opencode/src/server/event-replay.ts b/packages/opencode/src/server/event-replay.ts index a03342236..2830db6db 100644 --- a/packages/opencode/src/server/event-replay.ts +++ b/packages/opencode/src/server/event-replay.ts @@ -29,13 +29,13 @@ export type ReplaySnapshot = { replay: ReplayRecord[] gap: boolean invalidCursor: boolean - unsubscribe: () => void - releaseLiveQueue: (push: (record: ReplayRecord) => void) => void } const DEFAULT_MAX_RECORDS = 2048 const DEFAULT_MAX_AGE_MS = 5 * 60 * 1000 +// Keep this list intentionally small. These events are low-volume and session +// or blocker critical; high-volume streaming events recover through bootstrap. const REPLAYABLE_EVENT_TYPES = new Set([ "question.asked", "question.replied", @@ -65,13 +65,12 @@ export function isReplayableGlobalEvent(envelope: GlobalEventEnvelope): boolean } export class EventReplayStore { - private readonly bootID: string + private bootID: string private readonly maxRecords: number private readonly maxAgeMs: number private readonly now: () => number private seq = 0 private records: ReplayRecord[] = [] - private listeners = new Set<(record: ReplayRecord) => void>() constructor(input?: { bootID?: string; maxRecords?: number; maxAgeMs?: number; now?: () => number }) { this.bootID = input?.bootID ?? `${Date.now().toString(36)}-${randomUUID().slice(0, 8)}` @@ -98,31 +97,22 @@ export class EventReplayStore { this.records.push(record) this.prune() - for (const listener of this.listeners) listener(record) return record } - open(lastEventID: string | undefined, onLive: (record: ReplayRecord) => void): ReplaySnapshot { - const liveBuffer: ReplayRecord[] = [] - let replaying = true - - const listener = (record: ReplayRecord) => { - if (replaying) { - liveBuffer.push(record) - return - } - onLive(record) - } - - this.listeners.add(listener) - + snapshot(lastEventID: string | undefined): ReplaySnapshot { + this.prune() const fenceSeq = this.seq const fenceID = this.formatID(fenceSeq) const parsed = parseReplayCursor(lastEventID) const invalidCursor = !!lastEventID && (!parsed || parsed.bootID !== this.bootID || parsed.seq > fenceSeq) const cursorSeq = invalidCursor || !parsed ? fenceSeq : parsed.seq const earliestSeq = this.records[0]?.seq - const gap = !invalidCursor && !!parsed && earliestSeq !== undefined && parsed.seq < earliestSeq - 1 + const gap = + !invalidCursor && + !!parsed && + parsed.seq < fenceSeq && + (earliestSeq === undefined || parsed.seq < earliestSeq - 1) const replay = invalidCursor || !parsed ? [] @@ -135,19 +125,21 @@ export class EventReplayStore { replay, gap, invalidCursor, - unsubscribe: () => this.listeners.delete(listener), - releaseLiveQueue: (push) => { - replaying = false - for (const record of liveBuffer) { - if (record.seq > fenceSeq) push(record) - } - liveBuffer.length = 0 - }, } } - recordsForTest(): ReplayRecord[] { - return this.records + reset() { + this.bootID = `${this.now().toString(36)}-${randomUUID().slice(0, 8)}` + this.seq = 0 + this.records = [] + } + + clearDirectory(directory: string) { + this.records = this.records.filter((record) => record.envelope.directory !== directory) + } + + recordsForTest(): readonly ReplayRecord[] { + return this.records.slice() } private formatID(seq: number): string { diff --git a/packages/opencode/src/server/instance/global.ts b/packages/opencode/src/server/instance/global.ts index 002277ded..d1b098563 100644 --- a/packages/opencode/src/server/instance/global.ts +++ b/packages/opencode/src/server/instance/global.ts @@ -50,6 +50,12 @@ export function createGlobalEventReplayBridge(input?: { replayStore?: EventRepla return { replayStore, append(event: GlobalEventEnvelope) { + if (event.payload.type === GlobalDisposedEvent.type) { + replayStore.reset() + } + if (event.payload.type === "server.instance.disposed" && event.directory) { + replayStore.clearDirectory(event.directory) + } const record = replayStore.append(event) const packet = record ? packetForRecord(record) : packetForEnvelope(event) for (const listener of listeners) listener(packet) @@ -78,25 +84,11 @@ export function openGlobalEventReplayConnection(input: { } const unsubscribeLive = input.bridge.subscribe(pushLive) - const opened = input.bridge.replayStore.open(input.lastEventID, () => {}) - - input.push({ - id: input.lastEventID ? undefined : opened.fenceID, - data: JSON.stringify({ - payload: { - type: "server.connected", - properties: {}, - }, - }), - }) - - for (const record of opened.replay) { - input.push(packetForRecord(record)) - } + const opened = input.bridge.replayStore.snapshot(input.lastEventID) - if (opened.invalidCursor || opened.gap) { + const pushConnected = (id?: string) => { input.push({ - id: opened.fenceID, + id, data: JSON.stringify({ payload: { type: "server.connected", @@ -106,7 +98,18 @@ export function openGlobalEventReplayConnection(input: { }) } - opened.releaseLiveQueue(() => {}) + if (!input.lastEventID) { + pushConnected(opened.fenceID) + } + + for (const record of opened.replay) { + input.push(packetForRecord(record)) + } + + if (input.lastEventID && (opened.invalidCursor || opened.gap)) { + pushConnected(opened.fenceID) + } + replaying = false for (const packet of liveBuffer) { if (packet.replaySeq !== undefined && packet.replaySeq <= opened.fenceSeq) continue @@ -115,7 +118,6 @@ export function openGlobalEventReplayConnection(input: { liveBuffer.length = 0 return () => { - opened.unsubscribe() unsubscribeLive() } } @@ -176,14 +178,17 @@ async function streamEvents(c: Context, subscribe: (q: AsyncQueue }) } -async function streamGlobalEvents(c: Context) { +export async function streamGlobalEvents( + c: Context, + bridge: ReturnType = globalEventReplay, +) { const lastEventID = c.req.header("Last-Event-ID") ?? c.req.header("last-event-id") ?? undefined return streamSSE(c, async (stream) => { const q = new AsyncQueue() let done = false const unsubscribe = openGlobalEventReplayConnection({ - bridge: globalEventReplay, + bridge, lastEventID, push: (packet) => q.push(packet), }) diff --git a/packages/opencode/test/server/event-replay.test.ts b/packages/opencode/test/server/event-replay.test.ts index 4a6241586..1485409b9 100644 --- a/packages/opencode/test/server/event-replay.test.ts +++ b/packages/opencode/test/server/event-replay.test.ts @@ -109,37 +109,34 @@ describe("EventReplayStore", () => { store.append(question("q2")) store.append(question("q3")) - const opened = store.open("boot:1", () => {}) + const opened = store.snapshot("boot:1") expect(opened.invalidCursor).toBe(false) expect(opened.gap).toBe(false) expect(opened.replay.map((record) => record.id)).toEqual(["boot:2", "boot:3"]) - opened.unsubscribe() }) test("returns no replay for invalid boot id", () => { const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) store.append(question("q1")) - const opened = store.open("old:1", () => {}) + const opened = store.snapshot("old:1") expect(opened.invalidCursor).toBe(true) expect(opened.gap).toBe(false) expect(opened.replay).toEqual([]) expect(opened.fenceID).toBe("boot:1") - opened.unsubscribe() }) test("treats a same-boot future cursor as invalid", () => { const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) store.append(question("q1")) - const opened = store.open("boot:99", () => {}) + const opened = store.snapshot("boot:99") expect(opened.invalidCursor).toBe(true) expect(opened.replay).toEqual([]) expect(opened.fenceID).toBe("boot:1") - opened.unsubscribe() }) test("detects a gap when cursor is older than retained records", () => { @@ -148,28 +145,48 @@ describe("EventReplayStore", () => { store.append(question("q2")) store.append(question("q3")) - const opened = store.open("boot:0", () => {}) + const opened = store.snapshot("boot:0") expect(opened.gap).toBe(true) expect(opened.replay.map((record) => record.id)).toEqual(["boot:2", "boot:3"]) - opened.unsubscribe() }) - test("buffers live records during replay and releases only records after the fence", () => { + test("detects a gap when cursor is behind but all retained records expired", () => { + let now = 1000 + const store = new EventReplayStore({ bootID: "boot", maxAgeMs: 100, now: () => now }) + store.append(question("q1")) + now = 1200 + store.append(question("q2")) + now = 1400 + store.append(question("q3")) + now = 1600 + + const opened = store.snapshot("boot:1") + + expect(store.recordsForTest()).toEqual([]) + expect(opened.gap).toBe(true) + expect(opened.replay).toEqual([]) + expect(opened.fenceID).toBe("boot:3") + }) + + test("clears records for one disposed directory", () => { const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) store.append(question("q1")) + store.append({ ...question("q2"), directory: "/other" }) - const live: string[] = [] - const opened = store.open("boot:0", (record) => live.push(record.id)) - store.append(question("q2")) + store.clearDirectory("/repo") - expect(opened.fenceID).toBe("boot:1") - expect(opened.replay.map((record) => record.id)).toEqual(["boot:1"]) - expect(live).toEqual([]) + expect(store.recordsForTest().map((record) => record.envelope.directory)).toEqual(["/other"]) + expect(store.latestID()).toBe("boot:2") + }) - opened.releaseLiveQueue((record) => live.push(record.id)) + test("reset starts a new empty generation", () => { + const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) + store.append(question("q1")) - expect(live).toEqual(["boot:2"]) - opened.unsubscribe() + store.reset() + + expect(store.recordsForTest()).toEqual([]) + expect(store.latestID()).toMatch(/^[a-z0-9]+-[a-f0-9-]+:0$/) }) }) diff --git a/packages/opencode/test/server/global-event-replay.test.ts b/packages/opencode/test/server/global-event-replay.test.ts index e0520e965..8ac2053ef 100644 --- a/packages/opencode/test/server/global-event-replay.test.ts +++ b/packages/opencode/test/server/global-event-replay.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test" +import { Hono } from "hono" import { EventReplayStore } from "../../src/server/event-replay" -import { createGlobalEventReplayBridge, openGlobalEventReplayConnection } from "../../src/server/instance/global" +import { createGlobalEventReplayBridge, openGlobalEventReplayConnection, streamGlobalEvents } from "../../src/server/instance/global" const envelope = (type: string, id = type) => ({ directory: "/repo", @@ -10,6 +11,20 @@ const envelope = (type: string, id = type) => ({ }, }) +async function readSseUntil(response: Response, predicate: (text: string) => boolean) { + const reader = response.body?.getReader() + if (!reader) throw new Error("Expected SSE response body") + const decoder = new TextDecoder() + let text = "" + while (!predicate(text)) { + const next = await reader.read() + if (next.done) break + text += decoder.decode(next.value, { stream: true }) + } + await reader.cancel() + return text +} + describe("createGlobalEventReplayBridge", () => { test("live broadcasts replayable events with ids and non-replayable events without ids", () => { const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot" }) }) @@ -28,7 +43,7 @@ describe("createGlobalEventReplayBridge", () => { unsubscribe() }) - test("connection seeds a fresh cursor and replays missed records after Last-Event-ID", () => { + test("connection seeds a fresh cursor and valid reconnect replays without server.connected", () => { const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot" }) }) bridge.append(envelope("question.asked", "q1")) const fresh: Array<{ id?: string; replaySeq?: number; data: string }> = [] @@ -46,15 +61,26 @@ describe("createGlobalEventReplayBridge", () => { push: (packet) => reconnect.push(packet), })() - expect(reconnect.map((packet) => JSON.parse(packet.data).payload.type)).toEqual([ - "server.connected", - "question.replied", - ]) - expect(reconnect[0].id).toBeUndefined() - expect(reconnect[1].id).toBe("boot:2") + expect(reconnect.map((packet) => JSON.parse(packet.data).payload.type)).toEqual(["question.replied"]) + expect(reconnect[0].id).toBe("boot:2") }) - test("connection signals a gap even when partial replay records are available", () => { + test("invalid reconnect emits one server.connected with the fence id", () => { + const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot" }) }) + bridge.append(envelope("question.asked", "q1")) + const packets: Array<{ id?: string; replaySeq?: number; data: string }> = [] + + openGlobalEventReplayConnection({ + bridge, + lastEventID: "old:1", + push: (packet) => packets.push(packet), + })() + + expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual(["server.connected"]) + expect(packets[0].id).toBe("boot:1") + }) + + test("connection signals a gap once after partial replay records are available", () => { const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot", maxRecords: 1 }), }) @@ -68,11 +94,48 @@ describe("createGlobalEventReplayBridge", () => { push: (packet) => packets.push(packet), })() - expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual([ - "server.connected", - "question.replied", - "server.connected", - ]) + expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual(["question.replied", "server.connected"]) + expect(packets[0].id).toBe("boot:2") expect(packets.at(-1)?.id).toBe("boot:2") }) + + test("dispose events invalidate retained replay records", () => { + const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot" }) }) + bridge.append(envelope("question.asked", "q1")) + bridge.append({ + directory: "/repo", + payload: { + type: "server.instance.disposed", + properties: { directory: "/repo" }, + }, + }) + const packets: Array<{ id?: string; replaySeq?: number; data: string }> = [] + + openGlobalEventReplayConnection({ + bridge, + lastEventID: "boot:0", + push: (packet) => packets.push(packet), + })() + + expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual(["server.connected"]) + expect(packets[0].id).toBe("boot:1") + }) + + test("route reads Last-Event-ID and writes replay ids into SSE output", async () => { + const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot" }) }) + bridge.append(envelope("question.asked", "q1")) + const app = new Hono().get("/event", (c) => streamGlobalEvents(c, bridge)) + const controller = new AbortController() + + const response = await app.request("/event", { + headers: { "Last-Event-ID": "boot:0" }, + signal: controller.signal, + }) + const text = await readSseUntil(response, (value) => value.includes("question.asked")) + controller.abort() + + expect(text).toContain("id: boot:1") + expect(text).toContain("data: {\"directory\":\"/repo\",\"payload\":{\"type\":\"question.asked\"") + expect(text).not.toContain("server.connected") + }) }) From 844cf4f07974230ffe023bb68a437ffcd07d459d Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 12:15:27 +0800 Subject: [PATCH 07/12] fix(opencode): replay instance dispose events --- packages/opencode/src/server/event-replay.ts | 1 + .../opencode/src/server/instance/global.ts | 2 ++ .../opencode/test/server/event-replay.test.ts | 1 + .../test/server/global-event-replay.test.ts | 30 +++++++++++++++++-- 4 files changed, 32 insertions(+), 2 deletions(-) diff --git a/packages/opencode/src/server/event-replay.ts b/packages/opencode/src/server/event-replay.ts index 2830db6db..f4555bd37 100644 --- a/packages/opencode/src/server/event-replay.ts +++ b/packages/opencode/src/server/event-replay.ts @@ -46,6 +46,7 @@ const REPLAYABLE_EVENT_TYPES = new Set([ "session.updated", "session.deleted", "session.status", + "server.instance.disposed", ]) export function parseReplayCursor(input: string | undefined): ReplayCursor | undefined { diff --git a/packages/opencode/src/server/instance/global.ts b/packages/opencode/src/server/instance/global.ts index d1b098563..a4eed8050 100644 --- a/packages/opencode/src/server/instance/global.ts +++ b/packages/opencode/src/server/instance/global.ts @@ -98,6 +98,8 @@ export function openGlobalEventReplayConnection(input: { }) } + // Fresh connect seeds a cursor. Valid reconnect replays only missed records. + // Invalid or gapped reconnect sends one refresh signal and advances the cursor. if (!input.lastEventID) { pushConnected(opened.fenceID) } diff --git a/packages/opencode/test/server/event-replay.test.ts b/packages/opencode/test/server/event-replay.test.ts index 1485409b9..2a11dc250 100644 --- a/packages/opencode/test/server/event-replay.test.ts +++ b/packages/opencode/test/server/event-replay.test.ts @@ -44,6 +44,7 @@ describe("isReplayableGlobalEvent", () => { expect(isReplayableGlobalEvent(event("session.updated"))).toBe(true) expect(isReplayableGlobalEvent(event("session.deleted"))).toBe(true) expect(isReplayableGlobalEvent(event("session.status"))).toBe(true) + expect(isReplayableGlobalEvent(event("server.instance.disposed"))).toBe(true) }) test("rejects high-volume and unrelated events", () => { diff --git a/packages/opencode/test/server/global-event-replay.test.ts b/packages/opencode/test/server/global-event-replay.test.ts index 8ac2053ef..a513d0c1e 100644 --- a/packages/opencode/test/server/global-event-replay.test.ts +++ b/packages/opencode/test/server/global-event-replay.test.ts @@ -117,8 +117,34 @@ describe("createGlobalEventReplayBridge", () => { push: (packet) => packets.push(packet), })() - expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual(["server.connected"]) - expect(packets[0].id).toBe("boot:1") + expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual([ + "server.instance.disposed", + "server.connected", + ]) + expect(packets[0].id).toBe("boot:2") + expect(packets.at(-1)?.id).toBe("boot:2") + }) + + test("missed instance dispose advances reconnect recovery from the previous fence", () => { + const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot" }) }) + bridge.append(envelope("question.asked", "q1")) + bridge.append({ + directory: "/repo", + payload: { + type: "server.instance.disposed", + properties: { directory: "/repo" }, + }, + }) + const packets: Array<{ id?: string; replaySeq?: number; data: string }> = [] + + openGlobalEventReplayConnection({ + bridge, + lastEventID: "boot:1", + push: (packet) => packets.push(packet), + })() + + expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual(["server.instance.disposed"]) + expect(packets[0].id).toBe("boot:2") }) test("route reads Last-Event-ID and writes replay ids into SSE output", async () => { From 219c1145efe0206721c131f3d17167522d87df7c Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 12:18:30 +0800 Subject: [PATCH 08/12] fix(opencode): validate replay retention config --- packages/opencode/src/server/event-replay.ts | 12 +++++++-- .../opencode/test/server/event-replay.test.ts | 7 +++++ .../test/server/global-event-replay.test.ts | 26 ++++++++++++++----- 3 files changed, 36 insertions(+), 9 deletions(-) diff --git a/packages/opencode/src/server/event-replay.ts b/packages/opencode/src/server/event-replay.ts index f4555bd37..139bab0df 100644 --- a/packages/opencode/src/server/event-replay.ts +++ b/packages/opencode/src/server/event-replay.ts @@ -75,8 +75,16 @@ export class EventReplayStore { constructor(input?: { bootID?: string; maxRecords?: number; maxAgeMs?: number; now?: () => number }) { this.bootID = input?.bootID ?? `${Date.now().toString(36)}-${randomUUID().slice(0, 8)}` - this.maxRecords = input?.maxRecords ?? DEFAULT_MAX_RECORDS - this.maxAgeMs = input?.maxAgeMs ?? DEFAULT_MAX_AGE_MS + const maxRecords = input?.maxRecords ?? DEFAULT_MAX_RECORDS + const maxAgeMs = input?.maxAgeMs ?? DEFAULT_MAX_AGE_MS + if (!Number.isInteger(maxRecords) || maxRecords < 0) { + throw new RangeError("maxRecords must be a non-negative integer") + } + if (!Number.isFinite(maxAgeMs) || maxAgeMs < 0) { + throw new RangeError("maxAgeMs must be a non-negative number") + } + this.maxRecords = maxRecords + this.maxAgeMs = maxAgeMs this.now = input?.now ?? Date.now } diff --git a/packages/opencode/test/server/event-replay.test.ts b/packages/opencode/test/server/event-replay.test.ts index 2a11dc250..6b26372b3 100644 --- a/packages/opencode/test/server/event-replay.test.ts +++ b/packages/opencode/test/server/event-replay.test.ts @@ -58,6 +58,13 @@ describe("isReplayableGlobalEvent", () => { }) describe("EventReplayStore", () => { + test("rejects invalid retention configuration", () => { + expect(() => new EventReplayStore({ maxRecords: -1 })).toThrow(RangeError) + expect(() => new EventReplayStore({ maxRecords: 1.5 })).toThrow(RangeError) + expect(() => new EventReplayStore({ maxAgeMs: -1 })).toThrow(RangeError) + expect(() => new EventReplayStore({ maxAgeMs: Number.NaN })).toThrow(RangeError) + }) + test("assigns monotonic ids under one boot id", () => { const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) const first = store.append(question("q1")) diff --git a/packages/opencode/test/server/global-event-replay.test.ts b/packages/opencode/test/server/global-event-replay.test.ts index a513d0c1e..d354f3c1d 100644 --- a/packages/opencode/test/server/global-event-replay.test.ts +++ b/packages/opencode/test/server/global-event-replay.test.ts @@ -11,18 +11,30 @@ const envelope = (type: string, id = type) => ({ }, }) -async function readSseUntil(response: Response, predicate: (text: string) => boolean) { +async function readSseUntil(response: Response, predicate: (text: string) => boolean, timeoutMs = 2_000) { const reader = response.body?.getReader() if (!reader) throw new Error("Expected SSE response body") const decoder = new TextDecoder() let text = "" - while (!predicate(text)) { - const next = await reader.read() - if (next.done) break - text += decoder.decode(next.value, { stream: true }) + const deadline = Date.now() + timeoutMs + try { + while (!predicate(text)) { + const remaining = deadline - Date.now() + if (remaining <= 0) throw new Error(`Timed out waiting for SSE payload. Received: ${text}`) + const next = await Promise.race([ + reader.read(), + new Promise((_, reject) => + setTimeout(() => reject(new Error(`Timed out reading SSE stream. Received: ${text}`)), remaining), + ), + ]) + if (next.done) break + text += decoder.decode(next.value, { stream: true }) + } + if (!predicate(text)) throw new Error(`SSE stream ended before expected payload. Received: ${text}`) + return text + } finally { + await reader.cancel() } - await reader.cancel() - return text } describe("createGlobalEventReplayBridge", () => { From dbfc31bc08fa8e52022ac4557a917d87f1ef74b5 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 12:27:55 +0800 Subject: [PATCH 09/12] test(opencode): assert replay SSE route status --- packages/opencode/src/server/event-replay.ts | 2 ++ packages/opencode/src/server/instance/global.ts | 7 +++++-- packages/opencode/test/server/global-event-replay.test.ts | 1 + 3 files changed, 8 insertions(+), 2 deletions(-) diff --git a/packages/opencode/src/server/event-replay.ts b/packages/opencode/src/server/event-replay.ts index 139bab0df..ee3392126 100644 --- a/packages/opencode/src/server/event-replay.ts +++ b/packages/opencode/src/server/event-replay.ts @@ -161,3 +161,5 @@ export class EventReplayStore { while (this.records[0] && this.records[0].createdAt < cutoff) this.records.shift() } } + +export * as EventReplay from "./event-replay" diff --git a/packages/opencode/src/server/instance/global.ts b/packages/opencode/src/server/instance/global.ts index a4eed8050..e409aa2a7 100644 --- a/packages/opencode/src/server/instance/global.ts +++ b/packages/opencode/src/server/instance/global.ts @@ -28,9 +28,8 @@ type SsePacket = { export type GlobalEventReplayPacket = SsePacket -function packetForEnvelope(envelope: GlobalEventEnvelope, id?: string): SsePacket { +function packetForEnvelope(envelope: GlobalEventEnvelope): SsePacket { return { - id, data: JSON.stringify(envelope), } } @@ -51,6 +50,8 @@ export function createGlobalEventReplayBridge(input?: { replayStore?: EventRepla replayStore, append(event: GlobalEventEnvelope) { if (event.payload.type === GlobalDisposedEvent.type) { + // Global dispose starts a new replay generation. Online clients still + // receive the live packet; offline clients refresh on bootID mismatch. replayStore.reset() } if (event.payload.type === "server.instance.disposed" && event.directory) { @@ -104,6 +105,8 @@ export function openGlobalEventReplayConnection(input: { pushConnected(opened.fenceID) } + // Gap reconnects still get retained records first as best-effort recovery; + // the following server.connected refresh is the final source of truth. for (const record of opened.replay) { input.push(packetForRecord(record)) } diff --git a/packages/opencode/test/server/global-event-replay.test.ts b/packages/opencode/test/server/global-event-replay.test.ts index d354f3c1d..9caec2996 100644 --- a/packages/opencode/test/server/global-event-replay.test.ts +++ b/packages/opencode/test/server/global-event-replay.test.ts @@ -169,6 +169,7 @@ describe("createGlobalEventReplayBridge", () => { headers: { "Last-Event-ID": "boot:0" }, signal: controller.signal, }) + expect(response.status).toBe(200) const text = await readSseUntil(response, (value) => value.includes("question.asked")) controller.abort() From 74a5a49623bac552a3929fd82fe77902a287d54e Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 13:17:38 +0800 Subject: [PATCH 10/12] test: cover question replay in UI e2e --- packages/app/e2e/fixtures.ts | 4 + .../e2e/session/session-composer-dock.spec.ts | 126 ++++++++++++++++++ packages/app/src/context/global-sdk.tsx | 15 +++ packages/app/src/testing/terminal.ts | 5 + .../opencode/src/server/instance/question.ts | 46 +++++++ 5 files changed, 196 insertions(+) diff --git a/packages/app/e2e/fixtures.ts b/packages/app/e2e/fixtures.ts index 15ceec054..2626221e4 100644 --- a/packages/app/e2e/fixtures.ts +++ b/packages/app/e2e/fixtures.ts @@ -116,6 +116,7 @@ async function promptSend(page: Page) { } type ProjectHandle = { + url: string directory: string slug: string gotoSession: (sessionID?: string) => Promise @@ -517,6 +518,9 @@ function makeProject( gotoSession, trackSession, trackDirectory, + get url() { + return backend.url + }, get directory() { return need().directory }, diff --git a/packages/app/e2e/session/session-composer-dock.spec.ts b/packages/app/e2e/session/session-composer-dock.spec.ts index 64a7735b6..82968e042 100644 --- a/packages/app/e2e/session/session-composer-dock.spec.ts +++ b/packages/app/e2e/session/session-composer-dock.spec.ts @@ -1,4 +1,6 @@ import { mkdir } from "node:fs/promises" +import type { Page } from "@playwright/test" +import type { QuestionRequest } from "@opencode-ai/sdk/v2/client" import { test, expect } from "../fixtures" import { composerEvent, @@ -21,6 +23,11 @@ import { dict as enDict } from "../../src/i18n/en" type Sdk = Parameters[0] type PermissionRule = { permission: string; pattern: string; action: "allow" | "deny" | "ask" } +type ProjectQuestionSeed = { + url: string + directory: string + sdk: Sdk +} async function withDockSession( sdk: Sdk, @@ -80,6 +87,74 @@ async function withDockSeed(sdk: Sdk, sessionID: string, fn: () => Promise } } +function globalEventStream(page: Page) { + return { + cursor: () => + page.evaluate(() => { + const win = window as Window & { + __opencode_e2e?: { globalEventStream?: { cursor: () => string | undefined } } + } + return win.__opencode_e2e?.globalEventStream?.cursor() + }), + stop: () => + page.evaluate(() => { + const win = window as Window & { + __opencode_e2e?: { globalEventStream?: { stop: () => void } } + } + win.__opencode_e2e?.globalEventStream?.stop() + }), + start: () => + page.evaluate(() => { + const win = window as Window & { + __opencode_e2e?: { globalEventStream?: { start: () => void } } + } + win.__opencode_e2e?.globalEventStream?.start() + }), + } +} + +async function e2eAskQuestion( + project: ProjectQuestionSeed, + input: { sessionID: string; questions: typeof defaultQuestions }, +) { + const response = await fetch(`${project.url}/question/__e2e/ask?directory=${encodeURIComponent(project.directory)}`, { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(input), + }) + expect(response.status).toBe(204) +} + +async function e2ePublishQuestionAsked(project: ProjectQuestionSeed, request: QuestionRequest) { + const response = await fetch( + `${project.url}/question/__e2e/publish-asked?directory=${encodeURIComponent(project.directory)}`, + { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ request }), + }, + ) + expect(response.status).toBe(204) +} + +async function waitForQuestionSeed(project: ProjectQuestionSeed, sessionID: string) { + let current: QuestionRequest | undefined + await expect + .poll( + async () => { + const questions = await project.sdk.question.list().then((response) => response.data ?? []) + current = questions.find( + (question) => question.sessionID === sessionID && question.questions[0]?.header === defaultQuestions[0]?.header, + ) + return !!current + }, + { timeout: 30_000 }, + ) + .toBe(true) + if (!current) throw new Error("Question seed was not visible after polling") + return current +} + async function clearPermissionDock(page: any, label: RegExp) { const dock = page.locator(permissionDockSelector) await expect(dock).toBeVisible() @@ -383,6 +458,57 @@ test("blocked question flow unblocks after submit", async ({ page, llm, project ) }) +test("question dock recovers after missed question.asked via SSE replay", async ({ page, project }) => { + await project.open() + await withDockSession( + project.sdk, + "e2e composer dock question replay", + async (session) => { + await withDockSeed(project.sdk, session.id, async () => { + await project.gotoSession(session.id) + + const stream = globalEventStream(page) + + await expect.poll(stream.cursor, { timeout: 10_000 }).toMatch(/:/) + await stream.stop() + await e2eAskQuestion(project, { sessionID: session.id, questions: defaultQuestions }) + await waitForQuestionSeed(project, session.id) + + await expect(page.locator(questionDockSelector)).toHaveCount(0, { timeout: 750 }) + await stream.start() + + await expectQuestionBlocked(page) + await expect(page.locator(questionDockSelector)).toHaveCount(1) + }) + }, + { trackSession: project.trackSession }, + ) +}) + +test("stale question.asked does not reopen after question reply", async ({ page, project }) => { + await project.open() + await withDockSession( + project.sdk, + "e2e composer dock stale question", + async (session) => { + await withDockSeed(project.sdk, session.id, async () => { + await project.gotoSession(session.id) + + await e2eAskQuestion(project, { sessionID: session.id, questions: defaultQuestions }) + const request = await waitForQuestionSeed(project, session.id) + + await expectQuestionBlocked(page) + await project.sdk.question.reply({ requestID: request.id, questionReply: { answers: [["Continue"]] } }) + await expectQuestionOpen(page) + + await e2ePublishQuestionAsked(project, request) + await expect(page.locator(questionDockSelector)).toHaveCount(0, { timeout: 1_000 }) + }) + }, + { trackSession: project.trackSession }, + ) +}) + test("blocked question flow supports skipping one question before submit", async ({ page, llm, project }) => { await project.open() await withDockSession( diff --git a/packages/app/src/context/global-sdk.tsx b/packages/app/src/context/global-sdk.tsx index eff9b504c..9f7b9d7e5 100644 --- a/packages/app/src/context/global-sdk.tsx +++ b/packages/app/src/context/global-sdk.tsx @@ -5,6 +5,7 @@ import { makeEventListener } from "@solid-primitives/event-listener" import { batch, onCleanup, onMount } from "solid-js" import z from "zod" import { createSdkForServer } from "@/utils/server" +import type { E2EWindow } from "@/testing/terminal" import { useLanguage } from "./language" import { usePlatform } from "./platform" import { useServer } from "./server" @@ -217,7 +218,21 @@ export const { use: useGlobalSDK, provider: GlobalSDKProvider } = createSimpleCo clearHeartbeat() } + const e2e = () => { + if (typeof window === "undefined") return + const state = (window as E2EWindow).__opencode_e2e + if (!state) return + state.globalEventStream = { + stop, + start: () => { + void start() + }, + cursor: replayCursor.current, + } + } + onMount(() => { + e2e() makeEventListener(document, "visibilitychange", () => { if (document.visibilityState !== "visible") return if (!started) return diff --git a/packages/app/src/testing/terminal.ts b/packages/app/src/testing/terminal.ts index db8001ddf..86855a94a 100644 --- a/packages/app/src/testing/terminal.ts +++ b/packages/app/src/testing/terminal.ts @@ -30,6 +30,11 @@ export type E2EWindow = Window & { terminals?: Record controls?: Record } + globalEventStream?: { + stop: () => void + start: () => void + cursor: () => string | undefined + } } } diff --git a/packages/opencode/src/server/instance/question.ts b/packages/opencode/src/server/instance/question.ts index bf1dfe284..c639ae888 100644 --- a/packages/opencode/src/server/instance/question.ts +++ b/packages/opencode/src/server/instance/question.ts @@ -4,12 +4,58 @@ import { resolver } from "hono-openapi" import { QuestionID } from "@/question/schema" import { Question } from "../../question" import { AppRuntime } from "@/effect/app-runtime" +import { Bus } from "@/bus" +import { Env } from "@/env" +import { SessionID } from "@/session/schema" import z from "zod" import { errors } from "../error" import { lazy } from "../../util/lazy" export const QuestionRoutes = lazy(() => new Hono() + // E2E-only hooks exercise question transport/UI flows without relying on flaky LLM seeding. + .post( + "/__e2e/ask", + validator( + "json", + z.object({ + sessionID: SessionID.zod, + questions: z.array(Question.Info.zod), + }), + ), + async (c) => { + if (!Env.get("OPENCODE_E2E_LLM_URL")) return c.notFound() + + const json = c.req.valid("json") + AppRuntime.runPromise( + Question.Service.use((svc) => + svc.ask({ + sessionID: json.sessionID, + questions: json.questions, + }), + ), + ).catch(() => undefined) + + return c.body(null, 204) + }, + ) + .post( + "/__e2e/publish-asked", + validator( + "json", + z.object({ + request: Question.Request.zod, + }), + ), + async (c) => { + if (!Env.get("OPENCODE_E2E_LLM_URL")) return c.notFound() + + const json = c.req.valid("json") + await AppRuntime.runPromise(Bus.Service.use((bus) => bus.publish(Question.Event.Asked, json.request))) + + return c.body(null, 204) + }, + ) .get( "/", describeRoute({ From 79b97dd71830e86ef46cf22d32ab2a5f8c461b82 Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 13:30:43 +0800 Subject: [PATCH 11/12] fix: avoid partial replay on gapped cursors --- packages/opencode/src/server/event-replay.ts | 4 +++ .../opencode/src/server/instance/global.ts | 16 +++++------ .../opencode/test/server/event-replay.test.ts | 11 ++++++++ .../test/server/global-event-replay.test.ts | 28 +++++++++++++------ 4 files changed, 43 insertions(+), 16 deletions(-) diff --git a/packages/opencode/src/server/event-replay.ts b/packages/opencode/src/server/event-replay.ts index ee3392126..58b051fc3 100644 --- a/packages/opencode/src/server/event-replay.ts +++ b/packages/opencode/src/server/event-replay.ts @@ -140,6 +140,10 @@ export class EventReplayStore { reset() { this.bootID = `${this.now().toString(36)}-${randomUUID().slice(0, 8)}` this.seq = 0 + this.clear() + } + + clear() { this.records = [] } diff --git a/packages/opencode/src/server/instance/global.ts b/packages/opencode/src/server/instance/global.ts index e409aa2a7..12b3b89d4 100644 --- a/packages/opencode/src/server/instance/global.ts +++ b/packages/opencode/src/server/instance/global.ts @@ -105,14 +105,14 @@ export function openGlobalEventReplayConnection(input: { pushConnected(opened.fenceID) } - // Gap reconnects still get retained records first as best-effort recovery; - // the following server.connected refresh is the final source of truth. - for (const record of opened.replay) { - input.push(packetForRecord(record)) - } - - if (input.lastEventID && (opened.invalidCursor || opened.gap)) { - pushConnected(opened.fenceID) + // Do not send partial replay for invalid/gapped cursors. Missing earlier + // events can make retained blocker records stale, so bootstrap owns recovery. + if (opened.invalidCursor || opened.gap) { + if (input.lastEventID) pushConnected(opened.fenceID) + } else { + for (const record of opened.replay) { + input.push(packetForRecord(record)) + } } replaying = false diff --git a/packages/opencode/test/server/event-replay.test.ts b/packages/opencode/test/server/event-replay.test.ts index 6b26372b3..0163b3b9b 100644 --- a/packages/opencode/test/server/event-replay.test.ts +++ b/packages/opencode/test/server/event-replay.test.ts @@ -188,6 +188,17 @@ describe("EventReplayStore", () => { expect(store.latestID()).toBe("boot:2") }) + test("clears all retained records without changing the cursor generation", () => { + const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) + store.append(question("q1")) + store.append(question("q2")) + + store.clear() + + expect(store.recordsForTest()).toEqual([]) + expect(store.latestID()).toBe("boot:2") + }) + test("reset starts a new empty generation", () => { const store = new EventReplayStore({ bootID: "boot", now: () => 1000 }) store.append(question("q1")) diff --git a/packages/opencode/test/server/global-event-replay.test.ts b/packages/opencode/test/server/global-event-replay.test.ts index 9caec2996..4f29f2102 100644 --- a/packages/opencode/test/server/global-event-replay.test.ts +++ b/packages/opencode/test/server/global-event-replay.test.ts @@ -92,7 +92,7 @@ describe("createGlobalEventReplayBridge", () => { expect(packets[0].id).toBe("boot:1") }) - test("connection signals a gap once after partial replay records are available", () => { + test("gapped reconnect sends only server.connected and skips partial replay", () => { const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot", maxRecords: 1 }), }) @@ -106,9 +106,8 @@ describe("createGlobalEventReplayBridge", () => { push: (packet) => packets.push(packet), })() - expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual(["question.replied", "server.connected"]) + expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual(["server.connected"]) expect(packets[0].id).toBe("boot:2") - expect(packets.at(-1)?.id).toBe("boot:2") }) test("dispose events invalidate retained replay records", () => { @@ -129,12 +128,25 @@ describe("createGlobalEventReplayBridge", () => { push: (packet) => packets.push(packet), })() - expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual([ - "server.instance.disposed", - "server.connected", - ]) + expect(packets.map((packet) => JSON.parse(packet.data).payload.type)).toEqual(["server.connected"]) expect(packets[0].id).toBe("boot:2") - expect(packets.at(-1)?.id).toBe("boot:2") + }) + + test("global dispose resets the replay generation and clears retained records", () => { + const bridge = createGlobalEventReplayBridge({ replayStore: new EventReplayStore({ bootID: "boot" }) }) + bridge.append(envelope("question.asked", "q1")) + + const packet = bridge.append({ + payload: { + type: "global.disposed", + properties: {}, + }, + }) + + expect(JSON.parse(packet.data).payload.type).toBe("global.disposed") + expect(packet.id).toBeUndefined() + expect(bridge.replayStore.recordsForTest()).toEqual([]) + expect(bridge.replayStore.latestID()).not.toBe("boot:1") }) test("missed instance dispose advances reconnect recovery from the previous fence", () => { From 77ca52e08afe40ca8761eae7494cc12e993a851b Mon Sep 17 00:00:00 2001 From: Yuhan Lei Date: Sun, 3 May 2026 15:18:08 +0800 Subject: [PATCH 12/12] test: harden question e2e routes --- packages/app/e2e/backend.ts | 1 + packages/opencode/src/server/instance/question.ts | 8 +++++--- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/packages/app/e2e/backend.ts b/packages/app/e2e/backend.ts index a67a16ea9..3575c4c5c 100644 --- a/packages/app/e2e/backend.ts +++ b/packages/app/e2e/backend.ts @@ -85,6 +85,7 @@ export function createBackendEnv(input: { XDG_STATE_HOME: path.join(input.sandbox, "state"), OPENCODE_CLIENT: "app", OPENCODE_STRICT_CONFIG_DEPS: "true", + OPENCODE_E2E_ENABLED: "true", OPENCODE_E2E_LLM_URL: input.llmUrl, } for (const key of Object.keys(env)) { diff --git a/packages/opencode/src/server/instance/question.ts b/packages/opencode/src/server/instance/question.ts index b9ba3ce91..833ae2c35 100644 --- a/packages/opencode/src/server/instance/question.ts +++ b/packages/opencode/src/server/instance/question.ts @@ -13,6 +13,8 @@ import { errors } from "../error" import { lazy } from "../../util/lazy" const log = Log.create({ service: "server" }) +const e2eQuestionRoutesEnabled = () => + Env.get("OPENCODE_E2E_ENABLED") === "true" && !!Env.get("OPENCODE_E2E_LLM_URL") export const QuestionRoutes = lazy(() => new Hono() @@ -23,11 +25,11 @@ export const QuestionRoutes = lazy(() => "json", z.object({ sessionID: SessionID.zod, - questions: z.array(Question.Info.zod), + questions: z.array(Question.Info.zod).min(1).max(4), }), ), async (c) => { - if (!Env.get("OPENCODE_E2E_LLM_URL")) return c.notFound() + if (!e2eQuestionRoutesEnabled()) return c.notFound() const json = c.req.valid("json") void AppRuntime.runPromise( @@ -53,7 +55,7 @@ export const QuestionRoutes = lazy(() => }), ), async (c) => { - if (!Env.get("OPENCODE_E2E_LLM_URL")) return c.notFound() + if (!e2eQuestionRoutesEnabled()) return c.notFound() const json = c.req.valid("json") await AppRuntime.runPromise(Bus.Service.use((bus) => bus.publish(Question.Event.Asked, json.request)))