feat(opencode): add SSE replay for missed session events

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 }
 }
 

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)) {

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<void>
@@ -517,6 +518,9 @@ function makeProject(
       gotoSession,
       trackSession,
       trackDirectory,
+      get url() {
+        return backend.url
+      },
       get directory() {
         return need().directory
       },

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<typeof clearSessionDockSeed>[0]
 type PermissionRule = { permission: string; pattern: string; action: "allow" | "deny" | "ask" }
+type ProjectQuestionSeed = {
+  url: string
+  directory: string
+  sdk: Sdk
+}
 
 async function withDockSession<T>(
   sdk: Sdk,
@@ -80,6 +87,74 @@ async function withDockSeed<T>(sdk: Sdk, sessionID: string, fn: () => Promise<T>
   }
 }
 
+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(

packages/app/src/context/global-sdk.tsx

@@ -4,11 +4,13 @@ import { createGlobalEmitter } from "@solid-primitives/event-bus"
 import { makeEventListener } from "@solid-primitives/event-listener"
 import { batch, onCleanup, onMount } from "solid-js"
 import z from "zod"
+import type { E2EWindow } from "@/testing/terminal"
 import { createSdkForServer } from "@/utils/server"
 import { coalesceQueuedEvents, type QueuedGlobalEvent } from "./global-sdk-event-queue"
 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"),
@@ -91,6 +93,7 @@ export const { use: useGlobalSDK, provider: GlobalSDKProvider } = createSimpleCo
     const HEARTBEAT_TIMEOUT_MS = 15_000
     let lastEventAt = Date.now()
     let heartbeat: ReturnType<typeof setTimeout> | undefined
+    const replayCursor = createSseCursor()
     const resetHeartbeat = () => {
       lastEventAt = Date.now()
       if (heartbeat) clearTimeout(heartbeat)
@@ -119,6 +122,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
@@ -178,7 +185,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

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")
+  })
+})

packages/app/src/context/global-sdk/sse-cursor.ts

@@ -0,0 +1,20 @@
+export function createSseCursor() {
+  let value: string | undefined
+
+  return {
+    current() {
+      return value
+    },
+    update(id: string | undefined) {
+      if (!id) return
+      // SSE ids come from the server replay layer; keep this helper transport-only.
+      value = id
+    },
+    headers() {
+      if (!value) return undefined
+      const headers = new Headers()
+      headers.set("Last-Event-ID", value)
+      return headers
+    },
+  }
+}

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<string, Promise<void>>()
   const sessionLoads = new Map<string, Promise<void>>()
   const sessionMeta = new Map<string, { limit: number }>()
+  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)

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)
+  })
+})