@@ -2163,6 +2173,7 @@ function HomeSessionRow(props: {
{props.record.projectName}
+
serverSync().session.data.question,
+ info: () => serverSync().session.data.info,
+ // The same settle the replied/rejected events drive: a row the reconcile
+ // read proved dead leaves the shared store (idempotent against events).
+ removeQuestion: (clears) => serverSync().session.set(produce((draft) => applyInboxClears(draft.question, clears))),
+ isLive: (sessionID, requestID) =>
+ (serverSync().session.data.question[sessionID] ?? []).some((row) => row.id === requestID),
+ // The engine's /question list — the shared pending rows seam, read the
+ // same dual-protocol way bootstrapDirectory reads it.
+ listQuestions: async (directory) => {
+ const sdk = serverSDK()
+ if ((await sdk.protocol) === "v1")
+ return (await sdk.createClient({ directory, throwOnError: true }).question.list()).data ?? []
+ return (await sdk.api.question.request.list({ location: { directory } })).data
+ },
+ // The probe-verified reply path, scoped to the REQUESTING session's
+ // directory exactly like the dock's reply (amicode#117/#249).
+ reply: (input) =>
+ serverSDK().api.question.reply({
+ sessionID: input.sessionID,
+ requestID: input.requestID,
+ location: { directory: input.directory },
+ answers: input.answers,
+ }),
+ telemetry: postClientLog,
+ connected: () => serverSDK().event.status() === "connected",
+ })
+ onCleanup(() => inbox.dispose())
+
const update: TitlebarUpdate = {
version: () => {
const state = platform.updater?.state()
@@ -60,6 +102,9 @@ export default function NewLayout(props: ParentProps) {
into layout.tsx's LegacyLayout branch, which newLayoutDesigns never
reaches (VaultPanel was invisible here). */}
+ {/* amicode#1786 (S2): the question inbox's global host — the same
+ one-host-per-window pattern as the vault drawer. */}
+
{/* opencode#78 follow-up: a solver switch restarts the opencode server
under the webview. Speaks only for switches the app requested — unlike
the removed ConnectionBanner, silence is still the default. */}
diff --git a/packages/app-bundle/overlay/packages/app/src/pages/layout/project-avatar-state.ts b/packages/app-bundle/overlay/packages/app/src/pages/layout/project-avatar-state.ts
index ef255d9f2..8a89332ae 100644
--- a/packages/app-bundle/overlay/packages/app/src/pages/layout/project-avatar-state.ts
+++ b/packages/app-bundle/overlay/packages/app/src/pages/layout/project-avatar-state.ts
@@ -4,6 +4,7 @@ import { useNotification } from "@/context/notification"
import { usePermission } from "@/context/permission"
import { sessionPermissionRequest, sessionQuestionRequest } from "@/pages/session/composer/session-request-tree"
import { ServerConnection } from "@/context/server"
+import { deriveSessionListChip, type SessionListChip } from "@/pages/layout/session-list-chip"
export function useSessionTabAvatarState(
server: Accessor,
@@ -54,3 +55,33 @@ export function useSessionTabAvatarState(
// they were both just `unread`.
return { unread, loading, needsAttention, hasError }
}
+
+// amicode#1787 (S1): the session list row's status chip. READ-ONLY derivation
+// over the per-server SHARED ServerSession store — the same `data` leaves the
+// S2 question inbox derives from (question), the #1617/#1637 floors already
+// guard (session_working / session_status), and the existing re-seed seams
+// (connected edge, sync-gap frame, visibility-gated self-heal) keep current.
+// No new state, no polling; a session with no evidence derives undefined and
+// the row renders no chip (#1768 honest absence).
+export function useSessionListChipState(
+ server: Accessor,
+ sessionId: Accessor,
+ lastActivityAt: Accessor,
+) {
+ const global = useGlobal()
+ const connection = createMemo(() => global.servers.list().find((item) => ServerConnection.key(item) === server()))
+ const sync = createMemo(() => {
+ const conn = connection()
+ if (conn) return global.ensureServerCtx(conn).sync
+ })
+ return createMemo(() => {
+ const serverSync = sync()
+ if (!serverSync) return undefined
+ return deriveSessionListChip({
+ source: serverSync.session.data,
+ sessionID: sessionId(),
+ lastActivityAt: lastActivityAt(),
+ now: Date.now(),
+ })
+ })
+}
diff --git a/packages/app-bundle/overlay/packages/app/src/pages/layout/session-list-chip.test.ts b/packages/app-bundle/overlay/packages/app/src/pages/layout/session-list-chip.test.ts
new file mode 100644
index 000000000..62ef28a3a
--- /dev/null
+++ b/packages/app-bundle/overlay/packages/app/src/pages/layout/session-list-chip.test.ts
@@ -0,0 +1,85 @@
+import { describe, expect, test } from "bun:test"
+import type { QuestionRequest } from "@opencode-ai/sdk/v2/client"
+import { deriveSessionListChip, sessionListChipLabel } from "./session-list-chip"
+
+const questionRow = (id: string): QuestionRequest => ({ id, sessionID: "s2", questions: [] })
+
+// amicode#1787 (S1) — status chips on the session list: the tri-state
+// derivation over the shared rows (#1768's busy/idle/omitted semantics on the
+// ServerSession store's leaves — the same question rows the S2 inbox reads, so
+// "waiting on question" means the same thing on both surfaces).
+describe("amicode#1787 (S1) — session list chip derivation", () => {
+ test("a busy shared row reads as a running chip", () => {
+ const chip = deriveSessionListChip({
+ source: {
+ session_status: { s1: { type: "busy" } },
+ question: {},
+ },
+ sessionID: "s1",
+ now: 1_000_000,
+ })
+ expect(chip).toEqual({ kind: "running", retry: false })
+ })
+
+ test("a pending question row reads as waiting on question — the same rows the S2 inbox lists", () => {
+ const chip = deriveSessionListChip({
+ source: { session_status: { s2: { type: "busy" } }, question: { s2: [questionRow("q1")] } },
+ sessionID: "s2",
+ now: 1_000_000,
+ })
+ // A busy turn blocked on a question is WAITING ON THE QUESTION — the
+ // actionable state wins over running.
+ expect(chip).toEqual({ kind: "waiting-on-question" })
+ })
+
+ test("an explicit idle row reads as idle with age (label formats the age)", () => {
+ const now = 1_000_000_000
+ const idleChip = (ageMs: number | undefined) =>
+ deriveSessionListChip({
+ source: { session_status: { s3: { type: "idle" } }, question: {} },
+ sessionID: "s3",
+ lastActivityAt: ageMs === undefined ? undefined : now - ageMs,
+ now,
+ })
+ expect(idleChip(40 * 60_000)).toEqual({ kind: "idle", ageMs: 40 * 60_000 })
+ expect(sessionListChipLabel(idleChip(40 * 60_000)!)).toBe("idle 40m")
+ expect(sessionListChipLabel(idleChip(2 * 3_600_000)!)).toBe("idle 2h")
+ expect(sessionListChipLabel(idleChip(3 * 24 * 3_600_000)!)).toBe("idle 3d")
+ expect(sessionListChipLabel(idleChip(30_000)!)).toBe("idle")
+ expect(sessionListChipLabel(idleChip(undefined)!)).toBe("idle")
+ })
+
+ test("a retry status reads as a running chip with the retrying hint", () => {
+ const chip = deriveSessionListChip({
+ source: {
+ session_status: { s4: { type: "retry", attempt: 1, message: "provider 429", next: 1_100_000 } },
+ question: {},
+ },
+ sessionID: "s4",
+ now: 1_000_000,
+ })
+ expect(chip).toEqual({ kind: "running", retry: true })
+ expect(sessionListChipLabel(chip!)).toBe("retrying")
+ })
+
+ test("a session with NO status evidence is OMITTED: no chip, never a lying idle", () => {
+ const chip = deriveSessionListChip({
+ source: { session_status: {}, question: {} },
+ sessionID: "s5",
+ lastActivityAt: 1_000_000 - 3_600_000,
+ now: 1_000_000,
+ })
+ // #1768 rhyme: absence keeps the current value — at the chip level, no
+ // evidence renders nothing (an idle chip would assert idle as fact).
+ expect(chip).toBeUndefined()
+ })
+
+ test("the working floor alone reads as running (the #1617/#1637 floors feed the same chip)", () => {
+ const chip = deriveSessionListChip({
+ source: { session_status: {}, question: {}, session_working: (id) => id === "s6" },
+ sessionID: "s6",
+ now: 1_000_000,
+ })
+ expect(chip).toEqual({ kind: "running", retry: false })
+ })
+})
diff --git a/packages/app-bundle/overlay/packages/app/src/pages/layout/session-list-chip.tsx b/packages/app-bundle/overlay/packages/app/src/pages/layout/session-list-chip.tsx
new file mode 100644
index 000000000..7a0e9b87a
--- /dev/null
+++ b/packages/app-bundle/overlay/packages/app/src/pages/layout/session-list-chip.tsx
@@ -0,0 +1,101 @@
+// amicode#1787 (S1) — status chips on the session list. Render-only: the chip
+// DERIVES from the ServerSession store's shared-row leaves (busy/heartbeat
+// status + the question rows), the same leaves the #1617/#1637 floors already
+// guard and the S2 question inbox already derives from — so "waiting on
+// question" means the same thing on the list and in the inbox.
+//
+// The #1768 rhyme (peer-fleet session-status, one data plane over): tri-state
+// busy/idle/OMITTED — a session with NO status evidence renders NO chip
+// (absence never reads as idle), an unreachable owner's omission keeps the
+// current chip (the store's recency-guarded reconcile owns that floor), and
+// the row self-corrects when an explicit frame lands (owner returns / the
+// shared row TTLs out and a reconcile read reports idle). No polling: the chip
+// re-derives reactively off the store, which the event flow and the existing
+// re-seed seams (connected edge, sync-gap frame, visibility-gated self-heal)
+// already keep current.
+import type { QuestionRequest, SessionStatus } from "@opencode-ai/sdk/v2/client"
+import { Show } from "solid-js"
+
+/** The chip a session list row shows. */
+export type SessionListChip =
+ | { kind: "running"; retry: boolean }
+ | { kind: "waiting-on-question" }
+ | { kind: "idle"; ageMs: number | undefined }
+
+/** The shared-row source: the ServerSession store's own leaves. `session_working`
+ * is optional so a plain record (a status map without the working floor) still
+ * derives — the store passes its method, tests pass the store itself. */
+export type SessionListChipSource = {
+ session_status: Record
+ question: Record
+ session_working?: (sessionID: string) => boolean
+}
+
+/** Derive one row's chip. NO status row + no question + no working floor is
+ * the OMITTED state: it returns undefined and the row renders no chip —
+ * absence never reads as idle. */
+export function deriveSessionListChip(input: {
+ source: SessionListChipSource
+ sessionID: string
+ /** The session record's last-activity time (session.time.updated). */
+ lastActivityAt?: number
+ now: number
+}): SessionListChip | undefined {
+ const { source, sessionID } = input
+ // The question rows are the same rows the S2 inbox lists — one definition
+ // of "waiting on question" across both surfaces.
+ if ((source.question[sessionID]?.length ?? 0) > 0) return { kind: "waiting-on-question" }
+ const status = source.session_status[sessionID]
+ const working = source.session_working?.(sessionID) ?? false
+ if (working || (status?.type ?? "idle") !== "idle") return { kind: "running", retry: status?.type === "retry" }
+ if (status === undefined) return undefined // omitted — honest absence
+ return {
+ kind: "idle",
+ ageMs: input.lastActivityAt === undefined ? undefined : Math.max(0, input.now - input.lastActivityAt),
+ }
+}
+
+/** The chip's user-facing text. English-only by design, matching the tab
+ * status dot's labels (session-tab-status.tsx). */
+export function sessionListChipLabel(chip: SessionListChip): string {
+ if (chip.kind === "waiting-on-question") return "waiting on question"
+ if (chip.kind === "running") return chip.retry ? "retrying" : "running"
+ if (chip.ageMs === undefined) return "idle"
+ if (chip.ageMs < 60_000) return "idle"
+ if (chip.ageMs < 3_600_000) return `idle ${Math.floor(chip.ageMs / 60_000)}m`
+ if (chip.ageMs < 24 * 3_600_000) return `idle ${Math.floor(chip.ageMs / 3_600_000)}h`
+ return `idle ${Math.floor(chip.ageMs / (24 * 3_600_000))}d`
+}
+
+// ── the chip mark ────────────────────────────────────────────────────────────
+// Reads like the neighboring session-control-chip / status dot: a small
+// muted chip with a colored dot — the dot's tone reuses the SAME status
+// palette as SessionTabStatusDot (running yellow, needs-you green, idle grey),
+// so one session reads identically on the tab, in the dock, and on the list.
+
+const CHIP_TONE: Record = {
+ running: "var(--status-running)",
+ "waiting-on-question": "var(--status-done)",
+ idle: "var(--status-idle)",
+}
+
+export function SessionListStatusChip(props: { chip: SessionListChip | undefined }) {
+ return (
+
+ {(chip) => {
+ const label = sessionListChipLabel(chip)
+ return (
+
+
+ {label}
+
+ )
+ }}
+
+ )
+}
diff --git a/packages/app-bundle/overlay/packages/app/src/pages/layout/session-list-chip.wiring.test.tsx b/packages/app-bundle/overlay/packages/app/src/pages/layout/session-list-chip.wiring.test.tsx
new file mode 100644
index 000000000..cff1168fc
--- /dev/null
+++ b/packages/app-bundle/overlay/packages/app/src/pages/layout/session-list-chip.wiring.test.tsx
@@ -0,0 +1,64 @@
+// amicode#1787 (S1) — the session list chip's WIRING, source-asserted per the
+// repo's component-source-assertion pattern (session-header.test.tsx,
+// home-archive.test.ts): the DECISION logic is pure + unit-tested in
+// session-list-chip.tsx and the store semantics in server-session.test.ts;
+// here we assert the SolidJS surface binds them — the chip renders on the home
+// session row, rides the shared ServerSession store (the same question rows
+// the S2 inbox reads), and is render-only (no timers, no fetches, no writes).
+import { describe, expect, test } from "bun:test"
+import { readFileSync } from "node:fs"
+import { resolve } from "node:path"
+
+const homeSource = readFileSync(resolve(__dirname, "../home.tsx"), "utf8")
+const chipSource = readFileSync(resolve(__dirname, "session-list-chip.tsx"), "utf8")
+const avatarSource = readFileSync(resolve(__dirname, "project-avatar-state.ts"), "utf8")
+
+function homeSessionRowBody(src: string): string {
+ const start = src.indexOf("function HomeSessionRow(props: {")
+ expect(start).toBeGreaterThan(-1)
+ const next = src.indexOf("function HomeSessionSkeleton", start)
+ expect(next).toBeGreaterThan(start)
+ return src.slice(start, next)
+}
+
+describe("amicode#1787 — the session list chip renders on the home session row", () => {
+ test("HomeSessionRow derives and renders the chip off the shared store", () => {
+ const body = homeSessionRowBody(homeSource)
+ expect(body).toContain("useSessionListChipState")
+ expect(body).toContain(" {
+ // The hook hands the derivation the per-server shared store's own `data`
+ // (session_working / session_status / question) — not a per-directory
+ // child-store view — so "waiting on question" is the SAME question map the
+ // inbox's collectQuestionInbox walks (layout-new.tsx wires the inbox to
+ // `serverSync().session.data.question`).
+ expect(avatarSource).toContain("source: serverSync.session.data")
+ })
+
+ test("the chip component renders the tri-state with an honest-absence fallback (no chip when omitted)", () => {
+ expect(chipSource).toContain('data-slot="session-list-chip"')
+ expect(chipSource).toContain("data-kind={chip.kind}")
+ // The omitted state renders NOTHING — absence never reads as idle.
+ expect(chipSource).toContain("props.chip === undefined")
+ })
+
+ test("render-only: the chip adds no polling, no routes, no state writes (AC3/AC4)", () => {
+ for (const forbidden of [
+ "setInterval",
+ "setTimeout",
+ "requestAnimationFrame",
+ "fetch(",
+ "amicodeGet",
+ "reconcileStatuses",
+ ".set(",
+ "produce(",
+ ])
+ expect(chipSource).not.toContain(forbidden)
+ // The hook reads only — no store mutation from the chip lane.
+ const hook = avatarSource.slice(avatarSource.indexOf("export function useSessionListChipState"))
+ for (const forbidden of ["set(", "produce(", "reconcileStatuses", "setInterval"])
+ expect(hook).not.toContain(forbidden)
+ })
+})
diff --git a/packages/extension/src/chat_panel.ts b/packages/extension/src/chat_panel.ts
index 6a7461823..7e43099d1 100644
--- a/packages/extension/src/chat_panel.ts
+++ b/packages/extension/src/chat_panel.ts
@@ -370,6 +370,36 @@ export class ChatPanel implements LaunchTarget {
return ChatPanel.current;
}
+ /** #1788 (S3): the host's follow set — the session ids every LIVE chat tab
+ * is showing (a live tab IS a session this client has opened or pinned; a
+ * draft or closed route follows nothing). The followed-session watcher
+ * reads this live at each stream arrival. */
+ static followedSessionIds(): ReadonlySet {
+ const ids = new Set();
+ for (const panel of ChatPanel.live) if (panel.sessionID) ids.add(panel.sessionID);
+ return ids;
+ }
+
+ /** #1788 (S3): the session the FOCUSED panel shows, when the focused panel
+ * is showing one — "you're already there"; arrivals for it are observed
+ * but never notified. Undefined when no chat tab is the active editor. */
+ static focusedSessionId(): string | undefined {
+ const panel = ChatPanel.lastFocused;
+ return panel?.panel.active && panel.sessionID ? panel.sessionID : undefined;
+ }
+
+ /** #1788 (S3): reveal the live chat tab showing this session (a
+ * notification's activation action). True when a tab was found and
+ * revealed; a session no tab shows cannot be revealed. */
+ static revealSession(sessionID: string): boolean {
+ for (const panel of ChatPanel.live) {
+ if (panel.sessionID !== sessionID) continue;
+ panel.panel.reveal();
+ return true;
+ }
+ return false;
+ }
+
/** Snapshot the last focused panel's currently settled session for one host operation. */
static lastFocusedMutationContext(): MutationTrackingContext | undefined {
const panel = ChatPanel.lastFocused;
diff --git a/packages/extension/src/extension.ts b/packages/extension/src/extension.ts
index 0a93c3a50..9c27d9172 100644
--- a/packages/extension/src/extension.ts
+++ b/packages/extension/src/extension.ts
@@ -36,6 +36,7 @@ import {
import { resolveLabTomlPath, checkLabToml } from "./lab_config";
import { computeSetupState, writeSetupStateFile } from "./setup_state";
import { OpencodeEventClient } from "./sse_client";
+import { FollowedSessionWatcher } from "./followed_session_watcher";
import { RunsManager } from "./runs_manager";
import { stageDemoRun } from "./demo_replay";
import { writeStopFile, stopPlan, forceStop, runLogMtime } from "./run_controls";
@@ -151,6 +152,11 @@ let serverManager: ServerManager | undefined;
let amicodeService: { url: string; authHeader: string } | undefined;
let statusBar: StatusBarManager | undefined;
let sseClient: OpencodeEventClient | undefined;
+/** #1788 (S3): the passive-notification watcher — observes the sseClient's
+ * parsed events (no new connection) and raises VS Code notifications on
+ * followed sessions. Created once per activation, before any posture branch
+ * can construct the sseClient it rides on. */
+let followedSessionWatcher: FollowedSessionWatcher | undefined;
let runsManager: RunsManager | undefined;
let opencodeReadyUrl: URL | undefined;
/** Set once the binary + vault are known; the watcher's onRunFinished closure
@@ -1143,6 +1149,20 @@ export async function activate(ctx: vscode.ExtensionContext): Promise {
});
ctx.subscriptions.push({ dispose: () => unregisterBugReport() });
+ // #1788 (S3): passive notifications. The watcher rides the ONE sseClient
+ // stream every posture below opens (its observe-only onEvent sink) — no new
+ // connection. Follow = a session this client has opened or pinned (a live
+ // chat tab), derived live from ChatPanel; the follow management surface is
+ // future work, not this slice.
+ followedSessionWatcher = new FollowedSessionWatcher({
+ followedSessions: () => ChatPanel.followedSessionIds(),
+ isSessionFocused: (sessionID) => ChatPanel.focusedSessionId() === sessionID,
+ showInformationMessage: (m, ...items) => vscode.window.showInformationMessage(m, ...items) as Promise,
+ onFocusSession: (sessionID) => ChatPanel.revealSession(sessionID),
+ log: (line) => opencodeChannel.appendLine(line),
+ });
+ ctx.subscriptions.push(followedSessionWatcher);
+
// #1760 P4 — the hub-host client posture: amicode.hubServerUrl armed →
// spawn NOTHING and ride the hub (ALL platforms; the darwin-only fleet
// guard does NOT gate this — erlich is a Linux remote-SSH host and is
@@ -1199,6 +1219,8 @@ export async function activate(ctx: vscode.ExtensionContext): Promise {
channel: opencodeChannel,
statusBar,
authorization: serverAuthHeaders.Authorization,
+ // #1788 (S3): the watcher observes this stream — never a new connection.
+ onEvent: (e) => followedSessionWatcher?.observe(e),
});
ctx.subscriptions.push(sseClient);
// The hub session: probe the hub, attach through hysteresis. ONE server,
@@ -1269,6 +1291,8 @@ export async function activate(ctx: vscode.ExtensionContext): Promise {
channel: opencodeChannel,
statusBar,
authorization: serverAuthHeaders.Authorization,
+ // #1788 (S3): the watcher observes this stream — never a new connection.
+ onEvent: (e) => followedSessionWatcher?.observe(e),
});
ctx.subscriptions.push(sseClient);
// Poll the tunnel — when the canonical server is reachable the forward answers 200.
@@ -2198,6 +2222,8 @@ export async function activate(ctx: vscode.ExtensionContext): Promise {
channel: opencodeChannel,
statusBar,
authorization: serverAuthHeaders.Authorization,
+ // #1788 (S3): the watcher observes this stream — never a new connection.
+ onEvent: (e) => followedSessionWatcher?.observe(e),
});
ctx.subscriptions.push(sseClient);
@@ -2237,6 +2263,8 @@ export async function activate(ctx: vscode.ExtensionContext): Promise {
channel: opencodeChannel,
statusBar,
authorization: serverAuthHeaders.Authorization,
+ // #1788 (S3): the watcher observes this stream — never a new connection.
+ onEvent: (e) => followedSessionWatcher?.observe(e),
});
ctx.subscriptions.push(sseClient);
diff --git a/packages/extension/src/followed_session_watcher.ts b/packages/extension/src/followed_session_watcher.ts
new file mode 100644
index 000000000..46718afee
--- /dev/null
+++ b/packages/extension/src/followed_session_watcher.ts
@@ -0,0 +1,164 @@
+// ============================================================================
+// FOLLOWED SESSION WATCHER (#1788, S3 of #1785) — passive notifications on
+// FOLLOWED sessions: a session this client has OPENED or PINNED (a live chat
+// tab showing it — no new follow-management surface in this slice). The
+// extension host's existing SSE connection (sseClient) already sees every
+// hub event; the watcher is the sink that finally tells a human. The panel
+// taps you; you stop watching the panel.
+//
+// Observe-only by contract: the watcher consumes PARSED events handed to it
+// and never touches the stream, its cursor handling, or the connection
+// lifecycle. Dependency-injected (no direct vscode import — the neighbor
+// pattern from amicode_service/remote_activity_notifier.ts) so the POLICY —
+// who, when, how often — is testable under vitest without the VS Code host.
+// ============================================================================
+import type { OpencodeStreamEvent } from "./sse_client";
+
+export interface FollowedSessionWatcherDeps {
+ /** The follow set — session ids this client has opened or pinned, derived
+ * live at each arrival (a chat tab showing the session IS the follow). */
+ followedSessions: () => ReadonlySet;
+ /** True when a panel showing this session is the FOCUSED panel — "you're
+ * already there"; the arrival is observed but never notified. */
+ isSessionFocused: (sessionID: string) => boolean;
+ /** vscode.window.showInformationMessage — injected for testability. */
+ showInformationMessage: (message: string, ...items: string[]) => Thenable;
+ /** Fired when the user activates a notification's action — the wiring
+ * focuses the panel on that session. */
+ onFocusSession: (sessionID: string) => void;
+ /** Injectable clock for the turn-bracket and flash window. Default: Date.now. */
+ now?: () => number;
+ /** A turn that ends within this window of its busy frame reads as a flash —
+ * not a completion worth a ping. Default: FINISHED_MIN_TURN_MS. */
+ minTurnMs?: number;
+ /** Diagnostic sink (the extension's output channel). */
+ log?: (line: string) => void;
+}
+
+/** The flash window: turns shorter than this do not raise a "finished"
+ * notification (no flash spam from aborted or instantly-failing turns —
+ * and a reconnect replay's compressed busy→idle bracket reads the same
+ * way, so a replayed turn never re-notifies). */
+export const FINISHED_MIN_TURN_MS = 30_000;
+
+/** A plain, non-array object (the wire payloads the watcher validates). */
+function isRecord(value: unknown): value is Record {
+ return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+/** The session's display name for a notification — the stream-taught title,
+ * else the raw session id (honest, never invented). */
+function sessionName(titles: ReadonlyMap, sessionID: string): string {
+ return titles.get(sessionID) ?? sessionID;
+}
+
+export class FollowedSessionWatcher {
+ /** Question request ids ever OBSERVED (dedupe keys on the request id, never
+ * the arrival: the stream replays its gap on reconnect (#1264), and the
+ * same question.asked must not re-notify — only the FIRST arrival of a
+ * request id is ever eligible to notify). */
+ private readonly seenQuestions = new Set();
+ /** sessionID → when its busy status frame arrived (the turn bracket). */
+ private readonly busySince = new Map();
+ /** sessionID → max status seq seen (#1636): stale/replayed status frames
+ * are dropped exactly as the app's #1637 guard drops them. */
+ private readonly statusSeq = new Map();
+ /** sessionID → title, learned from the stream's session.created/updated. */
+ private readonly titles = new Map();
+ private disposed = false;
+
+ constructor(private readonly deps: FollowedSessionWatcherDeps) {}
+
+ /** Observe one parsed stream event. Never throws, never mutates anything
+ * upstream — a passive sink on the sseClient's dispatch path. */
+ observe(event: OpencodeStreamEvent): void {
+ if (this.disposed) return;
+ const props = event.properties ?? {};
+ switch (event.type) {
+ case "question.asked":
+ this.onQuestionAsked(props);
+ return;
+ case "session.status":
+ this.onStatus(props);
+ return;
+ case "session.created":
+ case "session.updated": {
+ // Title learning only — the notification needs the session's name.
+ if (isRecord(props.info) && typeof props.info.id === "string" && typeof props.info.title === "string") {
+ this.titles.set(props.info.id, props.info.title);
+ }
+ return;
+ }
+ default:
+ return;
+ }
+ }
+
+ dispose(): void {
+ this.disposed = true;
+ this.seenQuestions.clear();
+ this.busySince.clear();
+ this.statusSeq.clear();
+ this.titles.clear();
+ }
+
+ /** The turn bracket: a busy frame starts it, an idle frame ends it. The
+ * #1636 seq is honored locally exactly as the app's #1637 guard honors it,
+ * so a replayed/reordered status is a no-op. A turn that ends inside the
+ * flash window of its busy frame never notifies (no flash spam — and a
+ * reconnect replay's compressed bracket reads the same way); an idle with
+ * no observed busy cannot be measured and stays silent. */
+ private onStatus(props: Record): void {
+ const sessionID = typeof props.sessionID === "string" ? props.sessionID : "";
+ const status = isRecord(props.status) ? props.status : undefined;
+ if (sessionID === "" || !status) return;
+ const seq = typeof props.seq === "number" ? props.seq : undefined;
+ if (seq !== undefined) {
+ const seen = this.statusSeq.get(sessionID);
+ if (seen !== undefined && seq <= seen) return;
+ this.statusSeq.set(sessionID, seq);
+ }
+ const type = typeof status.type === "string" ? status.type : "idle";
+ if (type === "busy") {
+ this.busySince.set(sessionID, (this.deps.now ?? Date.now)());
+ return;
+ }
+ if (type !== "idle") return; // retry — the turn is still in flight
+ const startedAt = this.busySince.get(sessionID);
+ this.busySince.delete(sessionID);
+ if (startedAt === undefined) return;
+ const turnMs = (this.deps.now ?? Date.now)() - startedAt;
+ if (turnMs < (this.deps.minTurnMs ?? FINISHED_MIN_TURN_MS)) return;
+ if (!this.deps.followedSessions().has(sessionID)) return;
+ if (this.deps.isSessionFocused(sessionID)) return;
+ this.deps.log?.(`[notify] turn finished on followed session ${sessionID}`);
+ this.raise(sessionID, "finished its turn");
+ }
+
+ private onQuestionAsked(props: Record): void {
+ const requestID = typeof props.id === "string" ? props.id : "";
+ const sessionID = typeof props.sessionID === "string" ? props.sessionID : "";
+ if (requestID === "" || sessionID === "") return;
+ if (this.seenQuestions.has(requestID)) return; // replayed — never re-notify
+ this.seenQuestions.add(requestID);
+ if (!this.deps.followedSessions().has(sessionID)) return;
+ if (this.deps.isSessionFocused(sessionID)) return; // you're already there
+ const header =
+ Array.isArray(props.questions) && isRecord(props.questions[0]) && typeof props.questions[0].header === "string"
+ ? props.questions[0].header
+ : "a question";
+ this.deps.log?.(`[notify] question ${requestID} on followed session ${sessionID}`);
+ this.raise(sessionID, `is waiting on a question — ${header}`);
+ }
+
+ /** Raise one native notification for the session; its action focuses the
+ * panel on that session. */
+ private raise(sessionID: string, detail: string): void {
+ const name = sessionName(this.titles, sessionID);
+ void Promise.resolve(this.deps.showInformationMessage(`Amico: session "${name}" ${detail}`, "View")).then(
+ (action) => {
+ if (action === "View") this.deps.onFocusSession(sessionID);
+ },
+ );
+ }
+}
diff --git a/packages/extension/src/sse_client.ts b/packages/extension/src/sse_client.ts
index 40255a3bf..768dc885d 100644
--- a/packages/extension/src/sse_client.ts
+++ b/packages/extension/src/sse_client.ts
@@ -19,6 +19,9 @@ import { SseLivenessTracker, type SseState } from "./sse_liveness";
// dead SSE doesn't break the demo.
// ============================================================================
+/** One parsed stream event (the `data:` payload of an SSE frame). */
+export type OpencodeStreamEvent = { type?: string; properties?: Record };
+
export interface SseClientOptions {
channel: vscode.OutputChannel;
statusBar?: StatusBarManager;
@@ -29,6 +32,12 @@ export interface SseClientOptions {
/** Liveness states (L1, #638): fired on every transition. The status bar
* is driven automatically from this; this hook is for any other sink. */
onSseState?: (from: SseState, to: SseState) => void;
+ /** #1788 (S3): observe-only event sink — every PARSED event reaches it
+ * immediately, before the built-in routing. A passive consumer (the
+ * followed-session watcher): it must never mutate the stream, its cursor
+ * handling, or this connection. A throwing observer is CONTAINED here —
+ * logged, never propagated into the stream loop. */
+ onEvent?: (event: OpencodeStreamEvent) => void;
}
export class OpencodeEventClient implements vscode.Disposable {
@@ -197,7 +206,16 @@ export class OpencodeEventClient implements vscode.Disposable {
this.dispatch(event);
}
- private dispatch(event: { type?: string; properties?: Record }): void {
+ private dispatch(event: OpencodeStreamEvent): void {
+ // #1788 (S3): the observe-only sink first. Contained: a broken observer
+ // must never take the shared stream's own routing down with it.
+ if (this.opts.onEvent) {
+ try {
+ this.opts.onEvent(event);
+ } catch (err) {
+ this.opts.channel.appendLine(`[sse] event observer error: ${err instanceof Error ? err.message : String(err)}`);
+ }
+ }
const type = event.type ?? "";
const props = event.properties ?? {};
// Light routing: only the events we currently care about. We don't
diff --git a/packages/extension/test/chat_panel_sessions.test.ts b/packages/extension/test/chat_panel_sessions.test.ts
index dd2503a8c..352f6502d 100644
--- a/packages/extension/test/chat_panel_sessions.test.ts
+++ b/packages/extension/test/chat_panel_sessions.test.ts
@@ -177,4 +177,70 @@ describe("ChatPanel — side-by-side sessions (multi-instance registry)", () =>
expect(source.match(/d\.kind === "session-context"/g)?.length).toBeGreaterThanOrEqual(2);
expect(source.match(/d\.kind === "assessed-diff-invalidate"/g)?.length).toBeGreaterThanOrEqual(2);
});
+
+ // ── #1788 (S3): the host's follow set, derived from live panel state ──────
+ // Follow = a session this client has opened or pinned — a live chat tab
+ // showing it. These accessors are what the followed-session watcher reads.
+
+ function openPanels(n: number) {
+ const cap = capturePanels();
+ restore = cap.restore;
+ created = cap.created;
+ args = cap.args;
+ ChatPanel.openOrReveal(fakeCtx(), BASE);
+ for (let i = 1; i < n; i++) ChatPanel.openNew(fakeCtx(), DRAFT);
+ return cap;
+ }
+
+ it("#1788 — followedSessionIds is every LIVE tab's settled session (drafts and closed routes follow nothing)", () => {
+ openPanels(3);
+ const [p0, p1, p2] = created as unknown as Array<{
+ webview: { _simulateMessage(msg: unknown): void };
+ }>;
+ p0.webview._simulateMessage({ source: "amicode", kind: "session-context", sessionID: "ses_a" });
+ p1.webview._simulateMessage({ source: "amicode", kind: "session-context", sessionID: "ses_b" });
+ // A draft route (the third tab) holds no session and follows nothing.
+ p2.webview._simulateMessage({ source: "amicode", kind: "session-context", draft: true });
+
+ expect(new Set(ChatPanel.followedSessionIds())).toEqual(new Set(["ses_a", "ses_b"]));
+
+ // Closing a tab drops its session from the follow set (derived live).
+ p1.dispose();
+ expect(new Set(ChatPanel.followedSessionIds())).toEqual(new Set(["ses_a"]));
+ });
+
+ it("#1788 — focusedSessionId is the FOCUSED panel's session; an unfocused window reports none", () => {
+ openPanels(2);
+ const [p0, p1] = created as unknown as Array<{
+ webview: { _simulateMessage(msg: unknown): void };
+ _simulateViewState(active: boolean, visible: boolean): void;
+ }>;
+ p0.webview._simulateMessage({ source: "amicode", kind: "session-context", sessionID: "ses_a" });
+ p1.webview._simulateMessage({ source: "amicode", kind: "session-context", sessionID: "ses_b" });
+ p0._simulateViewState(true, true);
+
+ expect(ChatPanel.focusedSessionId()).toBe("ses_a");
+
+ // The user focuses a text editor: no chat panel is the active editor, so
+ // the window reports no focused session ("you're already there" lifts).
+ p0._simulateViewState(false, true);
+ expect(ChatPanel.focusedSessionId()).toBeUndefined();
+ });
+
+ it("#1788 — revealSession reveals the live tab showing that session (the notification click action)", () => {
+ openPanels(2);
+ const [p0, p1] = created as unknown as Array<{
+ webview: { _simulateMessage(msg: unknown): void };
+ revealCount: number;
+ }>;
+ p0.webview._simulateMessage({ source: "amicode", kind: "session-context", sessionID: "ses_a" });
+ p1.webview._simulateMessage({ source: "amicode", kind: "session-context", sessionID: "ses_b" });
+
+ expect(ChatPanel.revealSession("ses_b")).toBe(true);
+ expect(p1.revealCount).toBe(1);
+ expect(p0.revealCount).toBe(0);
+
+ // A session no tab shows (unfollowed by construction) cannot be revealed.
+ expect(ChatPanel.revealSession("ses_nowhere")).toBe(false);
+ });
});
diff --git a/packages/extension/test/followed_session_watcher.test.ts b/packages/extension/test/followed_session_watcher.test.ts
new file mode 100644
index 000000000..4e2b04c3f
--- /dev/null
+++ b/packages/extension/test/followed_session_watcher.test.ts
@@ -0,0 +1,207 @@
+// followed_session_watcher.test.ts — #1788 (S3 of #1785): passive
+// notifications on FOLLOWED sessions (a session this client has opened or
+// pinned). The watcher is a pure observer over the sseClient's parsed stream
+// events with every effect injected (no vscode import — the neighbor pattern
+// from remote_activity_notifier.ts), so the POLICY — who, when, how often —
+// is the tested surface, never the OS toast.
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import {
+ FollowedSessionWatcher,
+ type FollowedSessionWatcherDeps,
+} from "../src/followed_session_watcher";
+
+// The stream's parsed event shape (sse_client.ts): `data: {"type":..., "properties":{...}}`.
+
+function makeDeps(overrides?: Partial): FollowedSessionWatcherDeps {
+ return {
+ followedSessions: () => new Set(["ses_followed"]),
+ isSessionFocused: () => false,
+ showInformationMessage: vi.fn().mockResolvedValue(undefined),
+ onFocusSession: vi.fn(),
+ now: () => 0,
+ minTurnMs: 30_000,
+ ...overrides,
+ };
+}
+
+/** A question.asked frame, wire-shaped (properties = the QuestionRequest). */
+function questionAsked(requestID: string, sessionID: string, header = "Which solver?") {
+ return {
+ type: "question.asked",
+ properties: { id: requestID, sessionID, questions: [{ question: "Pick one.", header, options: [{ label: "A" }, { label: "B" }] }] },
+ };
+}
+
+/** The session-info frame that teaches the watcher the session's name. */
+function sessionUpdated(sessionID: string, title: string) {
+ return { type: "session.updated", properties: { info: { id: sessionID, title } } };
+}
+
+/** A #1636-wire-shaped status frame. */
+function sessionStatus(sessionID: string, type: "busy" | "idle" | "retry", seq: number) {
+ return { type: "session.status", properties: { sessionID, status: { type }, seq } };
+}
+
+describe("#1788 — followed session watcher (notification policy)", () => {
+ let deps: FollowedSessionWatcherDeps;
+
+ beforeEach(() => {
+ deps = makeDeps();
+ });
+
+ it("AC1 — question.asked on a followed, unfocused session notifies with the session name and question header; activating it focuses the panel", async () => {
+ (deps.showInformationMessage as ReturnType).mockResolvedValue("View");
+ const watcher = new FollowedSessionWatcher(deps);
+ watcher.observe(sessionUpdated("ses_followed", "Fix the flaky e2e"));
+
+ watcher.observe(questionAsked("req_1", "ses_followed", "Which solver?"));
+
+ expect(deps.showInformationMessage).toHaveBeenCalledOnce();
+ const [msg, action] = (deps.showInformationMessage as ReturnType).mock.calls[0];
+ // The session NAME and the question HEADER ride the notification (AC1).
+ expect(msg).toContain("Fix the flaky e2e");
+ expect(msg).toContain("Which solver?");
+ expect(action).toBe("View");
+
+ // Activating the notification focuses the panel on that session.
+ await vi.waitFor(() => expect(deps.onFocusSession).toHaveBeenCalledWith("ses_followed"));
+ });
+
+ it("AC2 — a turn-idle on a followed session that was busy raises a finished notification; a turn ending inside the flash window does not", async () => {
+ let clock = 10_000;
+ deps = makeDeps({ now: () => clock });
+ (deps.showInformationMessage as ReturnType).mockResolvedValue("View");
+ const watcher = new FollowedSessionWatcher(deps);
+ watcher.observe(sessionUpdated("ses_followed", "Long turn"));
+
+ // A LONG turn: busy at t=10s, idle at t=10s + 45s — past the flash window.
+ watcher.observe(sessionStatus("ses_followed", "busy", 1));
+ clock += 45_000;
+ watcher.observe(sessionStatus("ses_followed", "idle", 2));
+
+ expect(deps.showInformationMessage).toHaveBeenCalledOnce();
+ const [msg, action] = (deps.showInformationMessage as ReturnType).mock.calls[0];
+ expect(msg).toContain("Long turn");
+ expect(msg).toContain("finished");
+ expect(action).toBe("View");
+ await vi.waitFor(() => expect(deps.onFocusSession).toHaveBeenCalledWith("ses_followed"));
+
+ // A SHORT turn (a flash): busy at t, idle 5s later — inside the 30s
+ // window — must not notify.
+ watcher.observe(sessionStatus("ses_followed", "busy", 3));
+ clock += 5_000;
+ watcher.observe(sessionStatus("ses_followed", "idle", 4));
+ expect(deps.showInformationMessage).toHaveBeenCalledOnce(); // still just the long turn
+
+ // An idle with NO observed busy (the watcher attached mid-turn) cannot
+ // measure the turn and stays silent — never a guess.
+ watcher.observe(sessionStatus("ses_followed", "idle", 5));
+ expect(deps.showInformationMessage).toHaveBeenCalledOnce();
+ });
+
+ it("AC3 — the same question never notifies twice: a replayed question.asked (reconnect gap replay) is deduped by request id, and a genuinely new question still notifies", () => {
+ const watcher = new FollowedSessionWatcher(deps);
+
+ watcher.observe(questionAsked("req_1", "ses_followed"));
+ expect(deps.showInformationMessage).toHaveBeenCalledTimes(1);
+
+ // The stream replays its cursor gap on reconnect (#1264) — the SAME
+ // question.asked arrives again. The dedupe keys on the REQUEST id, never
+ // the arrival, so the replay passes silently.
+ watcher.observe(questionAsked("req_1", "ses_followed"));
+ watcher.observe(questionAsked("req_1", "ses_followed"));
+ expect(deps.showInformationMessage).toHaveBeenCalledTimes(1);
+
+ // A different question (a new request id) is a new arrival — it notifies.
+ watcher.observe(questionAsked("req_2", "ses_followed"));
+ expect(deps.showInformationMessage).toHaveBeenCalledTimes(2);
+ });
+
+ it("AC4 — notifications only fire for followed sessions; an unfollowed session's events pass silently", () => {
+ let clock = 10_000;
+ deps = makeDeps({ now: () => clock });
+ const watcher = new FollowedSessionWatcher(deps);
+
+ // A question, a long busy turn, and an idle — all on a session this
+ // client has NOT opened or pinned. All pass silently.
+ watcher.observe(questionAsked("req_other", "ses_unfollowed"));
+ watcher.observe(sessionStatus("ses_unfollowed", "busy", 1));
+ clock += 60_000;
+ watcher.observe(sessionStatus("ses_unfollowed", "idle", 2));
+ expect(deps.showInformationMessage).not.toHaveBeenCalled();
+
+ // The follow set is read LIVE at each arrival: the moment the client
+ // opens (or pins) the session, its next events notify.
+ deps.followedSessions = () => new Set(["ses_unfollowed"]);
+ watcher.observe(questionAsked("req_now_followed", "ses_unfollowed"));
+ expect(deps.showInformationMessage).toHaveBeenCalledTimes(1);
+ });
+
+ it("AC5 — no notification when the panel has that session open and focused (you're already there) — and the arrival is still deduped", () => {
+ deps = makeDeps({ isSessionFocused: (id) => id === "ses_followed" });
+ const watcher = new FollowedSessionWatcher(deps);
+
+ // You're watching the session when the question lands — no ping.
+ watcher.observe(questionAsked("req_watched", "ses_followed"));
+ expect(deps.showInformationMessage).not.toHaveBeenCalled();
+
+ // The arrival was still OBSERVED and deduped by request id: the replay
+ // after you navigate away (reconnect gap replay) never re-notifies for
+ // a question you were there to see.
+ deps.isSessionFocused = () => false;
+ watcher.observe(questionAsked("req_watched", "ses_followed"));
+ expect(deps.showInformationMessage).not.toHaveBeenCalled();
+
+ // A finished LONG turn while focused is equally silent.
+ deps.isSessionFocused = () => true;
+ let clock = 10_000;
+ deps.now = () => clock;
+ watcher.observe(sessionStatus("ses_followed", "busy", 1));
+ clock += 60_000;
+ watcher.observe(sessionStatus("ses_followed", "idle", 2));
+ expect(deps.showInformationMessage).not.toHaveBeenCalled();
+ });
+
+ it("replayed statuses are order-safe (#1636/#1637 lineage): a stale replayed idle must not kill a live turn bracket", () => {
+ let clock = 10_000;
+ deps = makeDeps({ now: () => clock });
+ const watcher = new FollowedSessionWatcher(deps);
+
+ // A real busy frame opens the turn bracket…
+ watcher.observe(sessionStatus("ses_followed", "busy", 1));
+ // …and a REPLAYED idle carrying the SAME (stale) seq arrives mid-turn
+ // (reconnect gap replay). The local seq guard drops it, so the bracket
+ // survives and the REAL idle still notifies.
+ clock += 5_000;
+ watcher.observe(sessionStatus("ses_followed", "idle", 1));
+ clock += 60_000;
+ watcher.observe(sessionStatus("ses_followed", "idle", 2));
+ expect(deps.showInformationMessage).toHaveBeenCalledTimes(1);
+ });
+
+ it("a session whose title was never observed falls back to its raw id in the notification", () => {
+ const watcher = new FollowedSessionWatcher(deps);
+ watcher.observe(questionAsked("req_untitled", "ses_followed"));
+ const [msg] = (deps.showInformationMessage as ReturnType).mock.calls[0];
+ expect(msg).toContain("ses_followed");
+ });
+
+ it("activation wiring — EVERY sseClient construction rides the watcher on the existing stream, and the watcher reads ChatPanel's live follow set", () => {
+ // Source-level wiring assertion (the chat_panel relay-test precedent):
+ // the extension holds ONE sseClient per posture branch (hub, fleet,
+ // local, adopted) — each must pass the watcher as its observe-only
+ // onEvent sink, so no posture can silently lose notifications.
+ const source = require("node:fs").readFileSync(
+ require("node:path").resolve(__dirname, "..", "src", "extension.ts"),
+ "utf8",
+ );
+ const sites = source.match(/new OpencodeEventClient\(\{[^}]*\}\)/gs) ?? [];
+ expect(sites.length).toBeGreaterThanOrEqual(4);
+ for (const site of sites) expect(site).toContain("onEvent:");
+ // The watcher is built once from ChatPanel's live follow state and its
+ // activation focuses the panel on the session.
+ expect(source).toContain("ChatPanel.followedSessionIds()");
+ expect(source).toContain("ChatPanel.focusedSessionId()");
+ expect(source).toContain("ChatPanel.revealSession");
+ });
+});
diff --git a/packages/extension/test/sse_client.test.ts b/packages/extension/test/sse_client.test.ts
index f003a4660..ce3c348f4 100644
--- a/packages/extension/test/sse_client.test.ts
+++ b/packages/extension/test/sse_client.test.ts
@@ -3,6 +3,7 @@ import * as http from "node:http";
import type { AddressInfo } from "node:net";
import { OpencodeEventClient } from "../src/sse_client";
import { mintServerPassword, serverAuthHeader, serverAuthToken } from "../src/server_auth";
+import { FollowedSessionWatcher } from "../src/followed_session_watcher";
// ============================================================================
// #163: with the per-boot server password armed, the fork 401s /event without
@@ -71,3 +72,117 @@ describe("OpencodeEventClient — authenticated /event subscription (#163)", ()
}
});
});
+
+// ============================================================================
+// #1788 (S3) — the observe-only event hook: every posture's extension host
+// holds ONE SSE connection; the followed-session watcher rides it. These
+// tests pin the seam against a REAL local http server: the observer sees the
+// parsed events on the SAME /event connection (no new connection is opened),
+// a throwing observer cannot break the stream loop, and the live
+// question.asked → notification path works end-to-end through the wire.
+// ============================================================================
+
+function eventServer() {
+ const requests: string[] = [];
+ const live: http.ServerResponse[] = [];
+ const srv = http.createServer((req, res) => {
+ requests.push(req.url ?? "");
+ live.push(res);
+ res.writeHead(200, { "content-type": "text/event-stream" });
+ res.write('data: {"type":"session.idle"}\n\n'); // held open — the client keeps streaming
+ });
+ return {
+ srv,
+ requests,
+ /** Push one raw SSE frame over every held-open connection (the test
+ * drives the wire exactly as the engine would mid-stream). */
+ push(frame: string) {
+ for (const res of live) res.write(frame);
+ },
+ };
+}
+
+describe("OpencodeEventClient — the #1788 observe-only event hook", () => {
+ it("delivers every parsed event to the observer on the SAME connection (AC6: no new connection)", async () => {
+ const { srv, requests } = eventServer();
+ await new Promise((r) => srv.listen(0, "127.0.0.1", r));
+ const { channel } = captureChannel();
+ const seen: string[] = [];
+ const client = new OpencodeEventClient({
+ channel,
+ onEvent: (e) => seen.push(e.type ?? ""),
+ });
+ try {
+ client.connect(new URL(`http://127.0.0.1:${(srv.address() as AddressInfo).port}`));
+ await vi.waitFor(() => expect(seen).toEqual(["session.idle"]));
+ // The observer is a SINK on the one /event request — never a second
+ // connection, and never a mutation of the stream.
+ expect(requests).toEqual(["/event"]);
+ } finally {
+ client.dispose();
+ srv.close();
+ }
+ });
+
+ it("a throwing observer is contained — the stream loop is never broken (shared-infrastructure guard)", async () => {
+ const { srv } = eventServer();
+ await new Promise((r) => srv.listen(0, "127.0.0.1", r));
+ const { lines, channel } = captureChannel();
+ const client = new OpencodeEventClient({
+ channel,
+ onEvent: () => {
+ throw new Error("observer bug");
+ },
+ });
+ try {
+ client.connect(new URL(`http://127.0.0.1:${(srv.address() as AddressInfo).port}`));
+ // The stream still connected (its own routing survived the observer).
+ await vi.waitFor(() => expect(lines.join("\n")).toContain("[sse] connected"));
+ expect(lines.join("\n")).toContain("observer error");
+ } finally {
+ client.dispose();
+ srv.close();
+ }
+ });
+
+ it("live path — a question.asked frame on the wire raises the followed-session notification through the existing stream (headless host assertion)", async () => {
+ const { srv, requests, push } = eventServer();
+ await new Promise((r) => srv.listen(0, "127.0.0.1", r));
+ const { channel } = captureChannel();
+ const showInformationMessage = vi.fn().mockResolvedValue(undefined);
+ const watcher = new FollowedSessionWatcher({
+ followedSessions: () => new Set(["ses_hub_1"]),
+ isSessionFocused: () => false,
+ showInformationMessage,
+ onFocusSession: vi.fn(),
+ now: () => 0,
+ });
+ const client = new OpencodeEventClient({
+ channel,
+ onEvent: (e) => watcher.observe(e),
+ });
+ try {
+ client.connect(new URL(`http://127.0.0.1:${(srv.address() as AddressInfo).port}`));
+ await vi.waitFor(() => expect(requests).toEqual(["/event"]));
+
+ // The hub streams the question while the panel is unfocused elsewhere —
+ // over the SAME /event connection the client already holds (AC6).
+ push(
+ 'data: {"type":"session.updated","properties":{"info":{"id":"ses_hub_1","title":"Attach e2e"}}}\n\n' +
+ 'data: {"type":"question.asked","properties":{"id":"req_live","sessionID":"ses_hub_1","questions":[{"header":"Which build?"}]}}\n\n',
+ );
+
+ await vi.waitFor(() => expect(showInformationMessage).toHaveBeenCalledOnce());
+ const [msg, action] = showInformationMessage.mock.calls[0];
+ expect(msg).toContain("Attach e2e");
+ expect(msg).toContain("Which build?");
+ expect(action).toBe("View");
+ // One connection, still — the watcher never opened anything.
+ expect(requests).toEqual(["/event"]);
+ } finally {
+ client.dispose();
+ watcher.dispose();
+ srv.close();
+ }
+ });
+});