From b642f9c7bc4bbeb23efc555bddc1c81bc89588a8 Mon Sep 17 00:00:00 2001 From: James Long Date: Thu, 9 Jul 2026 13:13:34 -0400 Subject: [PATCH] feat(tui): log event stream connections (#35973) --- packages/tui/src/context/sdk.tsx | 35 ++++++++++++++++++-- packages/tui/src/routes/session/index.tsx | 11 +++++- packages/tui/test/cli/tui/use-event.test.tsx | 15 +++++++-- 3 files changed, 56 insertions(+), 5 deletions(-) diff --git a/packages/tui/src/context/sdk.tsx b/packages/tui/src/context/sdk.tsx index d481ab3f85..d25b9683fc 100644 --- a/packages/tui/src/context/sdk.tsx +++ b/packages/tui/src/context/sdk.tsx @@ -7,9 +7,19 @@ import { createSimpleContext } from "./helper" import { useLog } from "./log" export type SDKConnectionStatus = "connected" | "connecting" | "reconnecting" +export type SDKConnectionEvent = { + readonly type: "client.connection" + readonly created: number + readonly data: { + readonly status: "connecting" | "connected" | "disconnected" | "reconnecting" + readonly attempt: number + readonly error?: string + } +} type SDKEventMap = { [Type in V2Event["type"]]: Extract } const connectTimeout = 2_000 +const connectionHistoryLimit = 50 export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ name: "SDK", @@ -20,8 +30,9 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ // Stops and starts the managed service; present only in service mode. reload?: () => Promise }) => { - const log = useLog() + const log = useLog({ component: "sdk" }) const abort = new AbortController() + const history: SDKConnectionEvent[] = [] let client = props.client let api = props.api const events = createGlobalEmitter() @@ -35,6 +46,11 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ }) let stream: AbortController | undefined + function record(status: SDKConnectionEvent["data"]["status"], attempt: number, error?: string) { + history.push({ type: "client.connection", created: Date.now(), data: { status, attempt, error } }) + if (history.length > connectionHistoryLimit) history.shift() + } + function start() { stream?.abort() const controller = new AbortController() @@ -54,6 +70,8 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ ) controller.signal.addEventListener("abort", cancel, { once: true }) const error = await (async () => { + record(attempt === 0 ? "connecting" : "reconnecting", attempt) + log.info("event stream connecting", { attempt }) const response = await client.v2.event.subscribe({ signal: connection.signal, sseMaxRetryAttempts: 0, @@ -69,7 +87,9 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ if (first.value.type !== "server.connected") return new Error("Event stream did not start with server.connected") clearTimeout(timeout) + record("connected", attempt) attempt = 0 + log.info("event stream connected") events.emit(first.value.type, first.value) setConnection({ status: "connected", attempt: 0, error: undefined }) connected() @@ -93,6 +113,12 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ }) if (abort.signal.aborted || controller.signal.aborted) return attempt += 1 + const message = error instanceof Error ? error.message : String(error) + record("disconnected", attempt, message) + log.info("event stream disconnected", { + attempt, + error: message, + }) // Re-resolve the transport before retrying: the server may have // moved (service restarted on a new port) or need starting. Static // transports (--server, standalone) resolve to the same address. @@ -107,7 +133,7 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ setConnection({ status: "reconnecting", attempt, - error: error instanceof Error ? error.message : String(error), + error: message, }) await wait(250, controller.signal) } @@ -143,6 +169,11 @@ export const { use: useSDK, provider: SDKProvider } = createSimpleContext({ error() { return connection.error }, + internal: { + history() { + return history.slice() + }, + }, }, reload: props.reload, } diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index 5210097fa9..917e605605 100644 --- a/packages/tui/src/routes/session/index.tsx +++ b/packages/tui/src/routes/session/index.tsx @@ -781,10 +781,19 @@ export function Session() { ) : await (async () => { if (options.debug) { - const events: unknown[] = [] + const events: { readonly created: number }[] = [] for await (const event of sdk.api.session.log({ sessionID: sessionData.id, follow: false })) { if (event.type !== "log.synced") events.push(event) } + // Durable events stay in aggregate order even when their wall-clock timestamps differ. + sdk.connection.internal.history().forEach((event) => { + const index = events.findIndex((item) => item.created > event.created) + if (index === -1) { + events.push(event) + return + } + events.splice(index, 0, event) + }) return JSON.stringify({ info: sessionData, events }, null, 2) + EOL } diff --git a/packages/tui/test/cli/tui/use-event.test.tsx b/packages/tui/test/cli/tui/use-event.test.tsx index 6c9aa750b9..1d03da2279 100644 --- a/packages/tui/test/cli/tui/use-event.test.tsx +++ b/packages/tui/test/cli/tui/use-event.test.tsx @@ -108,7 +108,9 @@ function Probe(props: { describe("useEvent", () => { test("logs only durable events", async () => { const logs: Array<{ message: string; tags: Readonly> }> = [] - const { app, emit, seen } = await mount(undefined, (_level, message, tags) => logs.push({ message, tags })) + const { app, emit, seen } = await mount(undefined, (_level, message, tags) => { + if (message === "event") logs.push({ message, tags }) + }) const durable = event( { id: "evt_renamed", @@ -128,7 +130,7 @@ describe("useEvent", () => { expect(logs).toEqual([ { message: "event", - tags: { type: "session.renamed", aggregateID: "ses_test", seq: 1 }, + tags: { component: "sdk", type: "session.renamed", aggregateID: "ses_test", seq: 1 }, }, ]) } finally { @@ -202,6 +204,15 @@ describe("useEvent", () => { expect(sdk.client).toBe(replacement.client) expect(sdk.api).toBe(replacement.api) + const history = sdk.connection.internal.history() + expect(history.map((event) => [event.data.status, event.data.attempt])).toEqual([ + ["connecting", 0], + ["connected", 0], + ["disconnected", 1], + ["reconnecting", 1], + ["connected", 1], + ]) + expect(history.every((event) => Number.isFinite(event.created))).toBe(true) } finally { app.renderer.destroy() }