From 319fd692d1c83bc05b3a38e4673f2e2fa5398db0 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Thu, 30 Jul 2026 06:29:31 -0700 Subject: [PATCH] fix(tui): preserve session state across switches and reconnects (#116399) * fix(tui): isolate session lifecycle state * fix(tui): preserve session incarnation identity * fix(tui): preserve gateway disconnect fallback --- src/tui/embedded-backend.test.ts | 129 ++++++++++++++++++++- src/tui/embedded-backend.ts | 39 ++++++- src/tui/gateway-chat.test.ts | 41 ++++++- src/tui/gateway-chat.ts | 27 ++++- src/tui/tui-backend.ts | 1 + src/tui/tui-command-handlers.test.ts | 164 +++++++++++++++++++++++++++ src/tui/tui-command-handlers.ts | 43 +++++-- src/tui/tui-event-handlers.test.ts | 50 ++++++++ src/tui/tui-event-handlers.ts | 1 + src/tui/tui-pty-harness.e2e.test.ts | 6 + src/tui/tui-pty-local.e2e.test.ts | 5 + src/tui/tui-run-lifecycle.ts | 14 ++- src/tui/tui-session-actions.ts | 1 + src/tui/tui-types.ts | 1 + src/tui/tui.test.ts | 18 ++- src/tui/tui.ts | 86 +++++++++++++- 16 files changed, 600 insertions(+), 26 deletions(-) diff --git a/src/tui/embedded-backend.test.ts b/src/tui/embedded-backend.test.ts index b81e2a609f51..cbe7e07bbaf1 100644 --- a/src/tui/embedded-backend.test.ts +++ b/src/tui/embedded-backend.test.ts @@ -137,7 +137,8 @@ vi.mock("../agents/agent-scope.js", () => ({ agents?: { list?: Array<{ id?: string; default?: boolean }> }; }) => cfg?.agents?.list?.find((agent) => agent.default)?.id ?? cfg?.agents?.list?.[0]?.id ?? "main", - resolveSessionAgentId: () => "main", + resolveSessionAgentId: (params: { sessionKey?: string; agentId?: string }) => + params.agentId ?? /^agent:([^:]+):/.exec(params.sessionKey ?? "")?.[1] ?? "main", })); vi.mock("../agents/runtime-plugins.js", () => ({ @@ -470,6 +471,8 @@ describe("EmbeddedTuiBackend", () => { event: "agent", payload: { runId: "run-local-1", + sessionKey: "agent:main:main", + agentId: "main", stream: "assistant", data: { delta: "hello" }, }, @@ -479,6 +482,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-local-1", sessionKey: "agent:main:main", + agentId: "main", state: "delta", deltaText: "hello", message: { @@ -492,6 +496,8 @@ describe("EmbeddedTuiBackend", () => { event: "agent", payload: { runId: "run-local-1", + sessionKey: "agent:main:main", + agentId: "main", stream: "lifecycle", data: { phase: "end", stopReason: "stop" }, }, @@ -501,6 +507,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-local-1", sessionKey: "agent:main:main", + agentId: "main", state: "final", stopReason: "stop", message: { @@ -942,6 +949,58 @@ describe("EmbeddedTuiBackend", () => { }); }); + it("reports the newest matching non-BTW local run in embedded history", async () => { + loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ + cfg: {}, + canonicalKey: sessionKey, + storePath: "/tmp/openclaw-work-sessions.json", + store: {}, + entry: { sessionId: "session-work-global" }, + })); + const first = deferred<{ payloads: Array<{ text: string }>; meta: Record }>(); + const second = deferred<{ payloads: Array<{ text: string }>; meta: Record }>(); + const side = deferred<{ text: string }>(); + agentCommandFromIngressMock + .mockReturnValueOnce(first.promise) + .mockReturnValueOnce(second.promise); + runBtwSideQuestionMock.mockReturnValueOnce(side.promise); + + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const backend = new EmbeddedTuiBackend(); + backend.start(); + await backend.sendChat({ + sessionKey: "global", + agentId: "work", + message: "first", + runId: "run-work-first", + }); + await backend.sendChat({ + sessionKey: "global", + agentId: "work", + message: "second", + runId: "run-work-newest", + }); + await backend.sendChat({ + sessionKey: "global", + agentId: "work", + message: "/btw detached", + runId: "run-work-btw", + }); + + await expect( + backend.loadHistory({ sessionKey: "global", agentId: "work" }), + ).resolves.toMatchObject({ + inFlightRun: { runId: "run-work-newest", text: "" }, + }); + + side.resolve({ text: "side done" }); + first.resolve({ payloads: [{ text: "first done" }], meta: {} }); + await vi.waitFor(() => expect(agentCommandFromIngressMock).toHaveBeenCalledTimes(2)); + second.resolve({ payloads: [{ text: "second done" }], meta: {} }); + await flushMicrotasks(); + await backend.stop(); + }); + it("uses reset-archive fallback for embedded TUI history reads", async () => { loadSessionEntryMock.mockReturnValue({ cfg: {}, @@ -1047,6 +1106,59 @@ describe("EmbeddedTuiBackend", () => { } }); + it("stamps the selected global agent on chat, agent, and BTW envelopes", async () => { + loadSessionEntryMock.mockImplementation((sessionKey: string) => ({ + cfg: {}, + canonicalKey: sessionKey, + storePath: "/tmp/openclaw-work-sessions.json", + store: {}, + entry: { sessionId: "session-work-global" }, + })); + const pending = deferred<{ + payloads: Array<{ text: string }>; + meta: Record; + }>(); + agentCommandFromIngressMock.mockReturnValueOnce(pending.promise); + runBtwSideQuestionMock.mockResolvedValueOnce({ text: "side done" }); + const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); + const backend = new EmbeddedTuiBackend(); + const events: Array<{ event: string; payload: unknown }> = []; + backend.onEvent = (event) => events.push({ event: event.event, payload: event.payload }); + backend.start(); + + await backend.sendChat({ + sessionKey: "global", + agentId: "work", + message: "hello", + runId: "run-global-work", + }); + registeredListener?.({ + runId: "run-global-work", + stream: "assistant", + data: { delta: "hello" }, + }); + pending.resolve({ payloads: [{ text: "hello" }], meta: {} }); + await flushMicrotasks(); + await backend.sendChat({ + sessionKey: "global", + agentId: "work", + message: "/btw detached", + runId: "run-global-work-btw", + }); + await flushMicrotasks(); + + expect( + events.filter((event) => ["chat", "agent", "chat.side_result"].includes(event.event)), + ).not.toHaveLength(0); + for (const event of events) { + if (!["chat", "agent", "chat.side_result"].includes(event.event)) { + continue; + } + expect(event.payload).toMatchObject({ sessionKey: "global", agentId: "work" }); + } + await backend.stop(); + }); + it("waits for local post-turn maintenance before emitting chat final", async () => { const { EmbeddedTuiBackend } = await import("./embedded-backend.js"); const pending = deferred<{ @@ -1726,6 +1838,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-local-first-terminal", sessionKey: "agent:main:main", + agentId: "main", state: "aborted", }, }); @@ -1774,6 +1887,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-validation-loop", sessionKey: "agent:main:main", + agentId: "main", state: "aborted", errorMessage: "edit tool validation failed: edits: must have required properties edits", }, @@ -1830,6 +1944,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-recovered-validation", sessionKey: "agent:main:main", + agentId: "main", state: "aborted", }, }); @@ -1872,6 +1987,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-unsafe-abort", sessionKey: "agent:main:main", + agentId: "main", state: "aborted", }, }); @@ -1929,6 +2045,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-local-idle-stop", sessionKey: "agent:main:main", + agentId: "main", state: "final", message: { role: "assistant", @@ -2365,6 +2482,7 @@ describe("EmbeddedTuiBackend", () => { expect(chatPayloads.at(-1)).toStrictEqual({ runId: "run-local-authoritative-final", sessionKey: "agent:main:main", + agentId: "main", state: "final", stopReason: "stop", message: { @@ -2429,6 +2547,7 @@ describe("EmbeddedTuiBackend", () => { expect(chatPayloads.at(-1)).toStrictEqual({ runId: "run-local-no", sessionKey: "agent:main:main", + agentId: "main", state: "final", stopReason: "stop", message: { @@ -2676,6 +2795,7 @@ describe("EmbeddedTuiBackend", () => { kind: "btw", runId: "run-btw-1", sessionKey: "agent:main:main", + agentId: "main", question: "what changed?", text: "nothing important", }, @@ -2685,6 +2805,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-btw-1", sessionKey: "agent:main:main", + agentId: "main", state: "final", }, }, @@ -2741,6 +2862,7 @@ describe("EmbeddedTuiBackend", () => { kind: "btw", runId: "run-side-1", sessionKey: "agent:main:main", + agentId: "main", question: "what changed?", text: "alias answer", }, @@ -2750,6 +2872,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-side-1", sessionKey: "agent:main:main", + agentId: "main", state: "final", }, }, @@ -2791,6 +2914,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-tool-first", sessionKey: "agent:main:main", + agentId: "main", state: "delta", deltaText: "", message: { @@ -2804,6 +2928,8 @@ describe("EmbeddedTuiBackend", () => { event: "agent", payload: { runId: "run-tool-first", + sessionKey: "agent:main:main", + agentId: "main", stream: "tool", data: { phase: "start", toolCallId: "tc-tool-first", name: "exec" }, }, @@ -2813,6 +2939,7 @@ describe("EmbeddedTuiBackend", () => { payload: { runId: "run-tool-first", sessionKey: "agent:main:main", + agentId: "main", state: "final", message: { role: "assistant", diff --git a/src/tui/embedded-backend.ts b/src/tui/embedded-backend.ts index 252719b0a6aa..4b3abeba0126 100644 --- a/src/tui/embedded-backend.ts +++ b/src/tui/embedded-backend.ts @@ -91,7 +91,7 @@ import { setEmbeddedPluginApprovalBroker, } from "../infra/embedded-plugin-approval-broker.js"; import { logInfo, logWarn } from "../logger.js"; -import { normalizeAgentId } from "../routing/session-key.js"; +import { agentSessionKeysMatchByRequestKey, normalizeAgentId } from "../routing/session-key.js"; import { defaultRuntime } from "../runtime.js"; import { INTERNAL_MESSAGE_CHANNEL } from "../utils/message-channel.js"; import { @@ -116,7 +116,7 @@ import { formatTuiErrorMessage } from "./tui-formatters.js"; type LocalRunState = { sessionKey: string; - agentId?: string; + agentId: string; controller: AbortController; buffer: string; lastBroadcastText?: string; @@ -453,9 +453,14 @@ export class EmbeddedTuiBackend implements TuiBackend { const runId = opts.runId ?? randomUUID(); const question = resolveBtwQuestion(opts.message); const isQueueCommand = resolveTextCommand(opts.message)?.command.key === "queue"; + const agentId = resolveSessionAgentId({ + sessionKey: opts.sessionKey, + config: getRuntimeConfig(), + agentId: opts.agentId, + }); const runScope = { sessionKey: opts.sessionKey, - agentId: opts.agentId, + agentId, }; const abortableSessionRun = this.hasAbortableSessionRun(runScope); const stopCommand = abortableSessionRun && isChatStopCommandText(opts.message); @@ -512,7 +517,7 @@ export class EmbeddedTuiBackend implements TuiBackend { const queuedRunReadiness = createQueuedRunReadiness(); this.runs.set(runId, { sessionKey: opts.sessionKey, - agentId: opts.agentId, + agentId, controller, buffer: "", isBtw: Boolean(question), @@ -657,6 +662,22 @@ export class EmbeddedTuiBackend implements TuiBackend { const capped = capArrayByJsonBytes(replaced.messages, maxHistoryBytes).items; const bounded = enforceChatHistoryFinalBudget({ messages: capped, maxBytes: maxHistoryBytes }); const messages = bounded.messages; + const newestInFlightRun = [...this.runs.entries()].findLast( + ([, run]) => + !run.isBtw && + !run.finalSent && + agentSessionKeysMatchByRequestKey(run.sessionKey, opts.sessionKey) && + normalizeAgentId(run.agentId) === normalizeAgentId(sessionAgentId), + ); + const inFlightRun = newestInFlightRun + ? { + runId: newestInFlightRun[0], + text: projectLiveAssistantBufferedText( + normalizeLiveAssistantBufferedText(newestInFlightRun[1].buffer).trim(), + { suppressLeadFragments: true }, + ).text.trim(), + } + : undefined; let thinkingLevel = entry?.thinkingLevel; if (!thinkingLevel) { @@ -691,6 +712,7 @@ export class EmbeddedTuiBackend implements TuiBackend { fastMode: entry?.fastMode, verboseLevel: sessionInfo.verboseLevel, runtimePluginsPrewarm, + ...(inFlightRun ? { inFlightRun } : {}), }; } @@ -1216,6 +1238,7 @@ export class EmbeddedTuiBackend implements TuiBackend { this.emit("chat", { runId, sessionKey: run.sessionKey, + agentId: run.agentId, state: "delta", ...deltaPayload, message: { @@ -1247,6 +1270,7 @@ export class EmbeddedTuiBackend implements TuiBackend { this.emit("chat", { runId, sessionKey: run.sessionKey, + agentId: run.agentId, state: "final", ...(stopReason ? { stopReason } : {}), ...(shouldIncludeMessage @@ -1277,6 +1301,7 @@ export class EmbeddedTuiBackend implements TuiBackend { this.emit("chat", { runId, sessionKey: run.sessionKey, + agentId: run.agentId, state: "aborted", ...(diagnostic ? { errorMessage: diagnostic } : {}), }); @@ -1297,6 +1322,7 @@ export class EmbeddedTuiBackend implements TuiBackend { this.emit("chat", { runId, sessionKey: run.sessionKey, + agentId: run.agentId, state: "error", ...(errorMessage ? { errorMessage } : {}), }); @@ -1311,6 +1337,7 @@ export class EmbeddedTuiBackend implements TuiBackend { this.emit("chat", { runId, sessionKey: run.sessionKey, + agentId: run.agentId, state: "delta", deltaText: "", message: { @@ -1339,6 +1366,8 @@ export class EmbeddedTuiBackend implements TuiBackend { this.emit("agent", { runId: evt.runId, + sessionKey: run.sessionKey, + agentId: run.agentId, stream: evt.stream, data: evt.data, }); @@ -1471,6 +1500,7 @@ export class EmbeddedTuiBackend implements TuiBackend { kind: "btw", runId: params.runId, sessionKey: result.sessionKey, + agentId: run.agentId, question: run.question, text: result.text, ...(result.isError ? { isError: true } : {}), @@ -1520,6 +1550,7 @@ export class EmbeddedTuiBackend implements TuiBackend { kind: "btw", runId: params.runId, sessionKey: run.sessionKey, + agentId: run.agentId, question: run.question, text, }); diff --git a/src/tui/gateway-chat.test.ts b/src/tui/gateway-chat.test.ts index cd43fec7aec4..2ce50aa2645f 100644 --- a/src/tui/gateway-chat.test.ts +++ b/src/tui/gateway-chat.test.ts @@ -665,7 +665,7 @@ describe("GatewayChatClient", () => { expect(stopped).toBe(true); }); - it("identifies the TUI as a tui client and skips device identity on insecure local ui paths", async () => { + it("identifies the TUI and forwards one structured connect failure per failed socket", async () => { const constructedOptions: Array> = []; vi.resetModules(); @@ -705,6 +705,45 @@ describe("GatewayChatClient", () => { tlsFingerprint: "sha256:11:22:33:44", deviceIdentity: null, }); + const onConnectError = vi.fn(); + const onDisconnected = vi.fn(); + client.onConnectError = onConnectError; + client.onDisconnected = onDisconnected; + const connectError = new GatewayClientRequestError({ + code: "INVALID_REQUEST", + message: "pairing required", + details: { code: "PAIRING_REQUIRED", requestId: "pair-1" }, + }); + const options = constructedOptions[0] as { + onConnectError?: (error: Error) => void; + onHelloOk?: (hello: unknown) => void; + onClose?: (code: number, reason: string) => void; + }; + + options.onConnectError?.(connectError); + options.onConnectError?.(new Error("duplicate failure for the same socket")); + options.onClose?.(1008, "pairing required"); + + expect(onConnectError).toHaveBeenCalledExactlyOnceWith(connectError); + expect(onDisconnected).not.toHaveBeenCalled(); + + const retryError = new Error("retry failed"); + options.onConnectError?.(retryError); + expect(onConnectError).toHaveBeenCalledOnce(); + options.onHelloOk?.({}); + options.onConnectError?.(retryError); + expect(onConnectError).toHaveBeenNthCalledWith(2, retryError); + + options.onHelloOk?.({}); + onDisconnected.mockClear(); + client.onConnectError = (error) => { + onConnectError(error); + client.onConnectError = undefined; + }; + ( + client as unknown as { notifyUnclosedConnectError: (error: Error) => void } + ).notifyUnclosedConnectError(new Error("one-shot structured failure")); + expect(onDisconnected).not.toHaveBeenCalled(); } finally { vi.doUnmock("../gateway/client.js"); vi.resetModules(); diff --git a/src/tui/gateway-chat.ts b/src/tui/gateway-chat.ts index f7a619d77b80..c5788aff9e16 100644 --- a/src/tui/gateway-chat.ts +++ b/src/tui/gateway-chat.ts @@ -132,11 +132,13 @@ export class GatewayChatClient implements TuiBackend { private client: GatewayClient; private readyPromise: Promise; private resolveReady?: () => void; + private pendingConnectError?: Error; readonly connection: ResolvedGatewayConnection; hello?: HelloOk; onEvent?: (evt: GatewayEvent) => void; onConnected?: () => void; + onConnectError?: (error: Error) => void; onDisconnected?: (reason: string) => void; onGap?: (info: { expected: number; received: number }) => void; @@ -170,6 +172,7 @@ export class GatewayChatClient implements TuiBackend { minProtocol: MIN_CLIENT_PROTOCOL_VERSION, maxProtocol: PROTOCOL_VERSION, onHelloOk: (hello) => { + this.pendingConnectError = undefined; this.hello = hello; this.resolveReady?.(); this.onConnected?.(); @@ -186,8 +189,12 @@ export class GatewayChatClient implements TuiBackend { this.readyPromise = new Promise((resolve) => { this.resolveReady = resolve; }); + if (this.pendingConnectError && this.onConnectError) { + return; + } this.onDisconnected?.(reason); }, + onConnectError: (error) => this.notifyConnectError(error), onGap: (info) => { this.onGap?.(info); }, @@ -212,14 +219,30 @@ export class GatewayChatClient implements TuiBackend { }) .then((readiness) => { if (!readiness.ready && !readiness.aborted) { - this.onDisconnected?.("gateway event loop readiness timeout"); + this.notifyUnclosedConnectError(new Error("gateway event loop readiness timeout")); } }) .catch((err: unknown) => { - this.onDisconnected?.(err instanceof Error ? err.message : String(err)); + this.notifyUnclosedConnectError(err instanceof Error ? err : new Error(String(err))); }); } + private notifyConnectError(error: Error) { + if (this.pendingConnectError) { + return; + } + this.pendingConnectError = error; + this.onConnectError?.(error); + } + + private notifyUnclosedConnectError(error: Error) { + const hasStructuredHandler = Boolean(this.onConnectError); + this.notifyConnectError(error); + if (!hasStructuredHandler) { + this.onDisconnected?.(error.message); + } + } + stop() { // Keep TUI teardown ordered after the transport closes. Otherwise the // late close callback can re-arm UI timers after shutdown cleared them. diff --git a/src/tui/tui-backend.ts b/src/tui/tui-backend.ts index d68138895485..6aeae5adaae6 100644 --- a/src/tui/tui-backend.ts +++ b/src/tui/tui-backend.ts @@ -181,6 +181,7 @@ export type TuiBackend = { }; onEvent?: (evt: TuiEvent) => void; onConnected?: () => void; + onConnectError?: (error: Error) => void; onDisconnected?: (reason: string) => void; onGap?: (info: { expected: number; received: number }) => void; start: () => void; diff --git a/src/tui/tui-command-handlers.test.ts b/src/tui/tui-command-handlers.test.ts index 1dd385668db7..e09bb52f7ccb 100644 --- a/src/tui/tui-command-handlers.test.ts +++ b/src/tui/tui-command-handlers.test.ts @@ -122,6 +122,7 @@ function createHarness(params?: { activityStatus?: string; opts?: { local?: boolean }; currentSessionId?: string | null; + sessionGeneration?: number; currentAgentId?: string; currentSessionKey?: string; sessionProjection?: SessionProjectionState; @@ -180,6 +181,7 @@ function createHarness(params?: { currentAgentId: params?.currentAgentId ?? "main", currentSessionKey: params?.currentSessionKey ?? "agent:main:main", currentSessionId: params?.currentSessionId ?? null, + sessionGeneration: params?.sessionGeneration ?? 0, sessionProjection: params?.sessionProjection, activeChatRunId: params?.activeChatRunId ?? null, pendingSubmit: params?.pendingSubmit ?? null, @@ -814,6 +816,168 @@ describe("tui command handlers", () => { expect(noteLocalRunId).toHaveBeenCalledWith("run-accepted"); }); + it("cleans a delayed ACK without mutating the newly selected viewport", async () => { + const deferred = createDeferred<{ runId: string; status: string }>(); + const harness = createHarness({ sendChat: vi.fn(() => deferred.promise) }); + const sending = harness.handleCommand("old session prompt"); + const provisionalRunId = (firstMockArg(harness.sendChat, "sendChat") as { runId: string }) + .runId; + const nextProjection = createSessionProjection( + { sessionKey: "agent:main:second", agentId: "main" }, + [ + { + role: "user", + content: [{ type: "text", text: "new session prompt" }], + __openclaw: { id: "new-user", seq: 1 }, + }, + ], + ); + harness.state.currentSessionKey = "agent:main:second"; + harness.state.sessionProjection = nextProjection; + harness.state.activeChatRunId = "new-active"; + harness.state.pendingSubmit = { + phase: "accepted", + runId: "new-pending", + draftText: "new draft", + }; + harness.state.activityStatus = "streaming"; + + deferred.resolve({ runId: "old-accepted", status: "error" }); + await sending; + + expect(harness.state.sessionProjection).toBe(nextProjection); + expect(harness.state.activeChatRunId).toBe("new-active"); + expect(harness.state.pendingSubmit).toEqual({ + phase: "accepted", + runId: "new-pending", + draftText: "new draft", + }); + expect(harness.loadHistory).not.toHaveBeenCalled(); + expect(harness.addSystem).not.toHaveBeenCalled(); + expect(harness.setActivityStatus).toHaveBeenCalledExactlyOnceWith("sending"); + expect(harness.forgetLocalRunId).toHaveBeenCalledWith(provisionalRunId); + expect(harness.forgetLocalRunId).toHaveBeenCalledWith("old-accepted"); + }); + + it("ignores a delayed ACK after the selected session is replaced in place", async () => { + const deferred = createDeferred<{ runId: string; status: string }>(); + const harness = createHarness({ + currentSessionId: "session-old", + sendChat: vi.fn(() => deferred.promise), + }); + const sending = harness.handleCommand("old incarnation prompt"); + harness.state.currentSessionId = "session-new"; + harness.state.activeChatRunId = "new-active"; + harness.state.pendingSubmit = { + phase: "accepted", + runId: "new-pending", + draftText: null, + }; + + deferred.resolve({ runId: "old-accepted", status: "error" }); + await sending; + + expect(harness.state.currentSessionId).toBe("session-new"); + expect(harness.state.activeChatRunId).toBe("new-active"); + expect(harness.state.pendingSubmit?.runId).toBe("new-pending"); + expect(harness.loadHistory).not.toHaveBeenCalled(); + expect(harness.addSystem).not.toHaveBeenCalled(); + }); + + it("allows a first send to bind a previously unknown session incarnation", async () => { + const deferred = createDeferred<{ runId: string }>(); + const harness = createHarness({ + currentSessionId: null, + sendChat: vi.fn(() => deferred.promise), + }); + const sending = harness.handleCommand("first session prompt"); + const provisionalRunId = (firstMockArg(harness.sendChat, "sendChat") as { runId: string }) + .runId; + harness.state.currentSessionId = "session-created"; + deferred.resolve({ runId: provisionalRunId }); + + await sending; + + expect(harness.state.currentSessionId).toBe("session-created"); + expect(getPendingSubmitAcceptedRunId(harness.state)).toEqual(expect.any(String)); + }); + + it("rejects a delayed ACK when an unknown session is replaced before binding", async () => { + const deferred = createDeferred<{ runId: string }>(); + const harness = createHarness({ + currentSessionId: null, + sessionGeneration: 0, + sendChat: vi.fn(() => deferred.promise), + }); + const sending = harness.handleCommand("old unbound prompt"); + harness.state.currentSessionId = "replacement-session"; + harness.state.sessionGeneration = 1; + harness.state.pendingSubmit = { + phase: "accepted", + runId: "replacement-pending", + draftText: null, + }; + deferred.resolve({ runId: "old-accepted" }); + + await sending; + + expect(harness.state.pendingSubmit?.runId).toBe("replacement-pending"); + expect(harness.loadHistory).not.toHaveBeenCalled(); + expect(harness.addSystem).not.toHaveBeenCalled(); + }); + + it("accepts a delayed ACK after returning to the exact original session incarnation", async () => { + const deferred = createDeferred<{ runId: string }>(); + const harness = createHarness({ + currentSessionKey: "agent:main:a", + currentSessionId: "session-a", + sessionGeneration: 3, + sendChat: vi.fn(() => deferred.promise), + }); + const sending = harness.handleCommand("original prompt"); + const provisionalRunId = (firstMockArg(harness.sendChat, "sendChat") as { runId: string }) + .runId; + harness.state.currentSessionKey = "agent:main:b"; + harness.state.currentSessionId = "session-b"; + harness.state.currentSessionKey = "agent:main:a"; + harness.state.currentSessionId = "session-a"; + deferred.resolve({ runId: provisionalRunId }); + + await sending; + + expect(getPendingSubmitAcceptedRunId(harness.state)).toBe(provisionalRunId); + expect(harness.setActivityStatus).toHaveBeenLastCalledWith("waiting"); + }); + + it("cleans a delayed send rejection without reporting it in a new session", async () => { + const deferred = createDeferred(); + const harness = createHarness({ sendChat: vi.fn(() => deferred.promise) }); + const sending = harness.handleCommand("old session prompt"); + const provisionalRunId = (firstMockArg(harness.sendChat, "sendChat") as { runId: string }) + .runId; + const nextProjection = createSessionProjection({ + sessionKey: "agent:work:second", + agentId: "work", + }); + harness.state.currentAgentId = "work"; + harness.state.currentSessionKey = "agent:work:second"; + harness.state.sessionProjection = nextProjection; + harness.state.pendingSubmit = { + phase: "accepted", + runId: "new-pending", + draftText: null, + }; + + deferred.reject(new Error("old gateway failure")); + await sending; + + expect(harness.state.sessionProjection).toBe(nextProjection); + expect(harness.state.pendingSubmit?.runId).toBe("new-pending"); + expect(harness.addSystem).not.toHaveBeenCalled(); + expect(harness.dropPendingUser).not.toHaveBeenCalled(); + expect(harness.forgetLocalRunId).toHaveBeenCalledWith(provisionalRunId); + }); + it("clears optimistic state when chat send returns a terminal timeout ack", async () => { const sendChat = vi.fn().mockImplementation(async (opts: { runId: string }) => ({ runId: opts.runId, diff --git a/src/tui/tui-command-handlers.ts b/src/tui/tui-command-handlers.ts index 25c9d97e7e0a..323d5f86ec49 100644 --- a/src/tui/tui-command-handlers.ts +++ b/src/tui/tui-command-handlers.ts @@ -920,6 +920,13 @@ export function createCommandHandlers(context: CommandHandlerContext) { return; } const runId = randomUUID(); + const sendSelection = captureSessionSelection(); + const sendSessionId = state.currentSessionId; + const sendSessionGeneration = state.sessionGeneration ?? 0; + const isCurrentSendViewport = () => + isCurrentSessionSelection(sendSelection) && + (state.sessionGeneration ?? 0) === sendSessionGeneration && + (sendSessionId === null || state.currentSessionId === sendSessionId); const sendScope = readTuiSessionProjectionScope(state); try { if (!isBtw) { @@ -945,9 +952,9 @@ export function createCommandHandlers(context: CommandHandlerContext) { } tui.requestRender(); const sendResult = await client.sendChat({ - sessionKey: state.currentSessionKey, - ...(state.currentSessionKey === "global" ? { agentId: state.currentAgentId } : {}), - sessionId: state.currentSessionId, + sessionKey: sendSelection.sessionKey, + ...(sendSelection.sessionKey === "global" ? { agentId: sendSelection.agentId } : {}), + sessionId: sendSessionId, message: text, thinking: opts.thinking, deliver: deliverDefault, @@ -958,6 +965,23 @@ export function createCommandHandlers(context: CommandHandlerContext) { const terminalAckFailure = isTerminalChatSendAckFailure(sendResult.status); const terminalAckSuccess = isTerminalChatSendAckSuccess(sendResult.status); const terminalAck = terminalAckFailure || terminalAckSuccess; + if (!isCurrentSendViewport()) { + if (isBtw) { + forgetLocalBtwRunId?.(runId); + if (acceptedRunId !== runId) { + forgetLocalBtwRunId?.(acceptedRunId); + } + } else { + forgetLocalRunId?.(runId); + if (acceptedRunId !== runId) { + forgetLocalRunId?.(acceptedRunId); + } + clearPendingSubmit(state, runId); + clearPendingSubmit(state, acceptedRunId); + consumeCompletedRunForPendingSend?.(acceptedRunId); + } + return; + } if (isBtw && terminalAck) { forgetLocalBtwRunId?.(runId); if (acceptedRunId !== runId) { @@ -1049,13 +1073,16 @@ export function createCommandHandlers(context: CommandHandlerContext) { } catch (err) { if (isBtw) { forgetLocalBtwRunId?.(runId); - } - if (!isBtw && state.activeChatRunId && state.activeChatRunId === runId) { - forgetLocalRunId?.(state.activeChatRunId); - } - if (!isBtw) { + } else { forgetLocalRunId?.(runId); } + if (!isCurrentSendViewport()) { + clearPendingSubmit(state, runId); + return; + } + if (!isBtw && state.activeChatRunId === runId) { + forgetLocalRunId?.(state.activeChatRunId); + } if (!isBtw) { // Only clear the failed send's ownership. A queued run may have // terminalized or handed ownership off while the RPC was pending. diff --git a/src/tui/tui-event-handlers.test.ts b/src/tui/tui-event-handlers.test.ts index f64f9ee5a084..cbc788d3daec 100644 --- a/src/tui/tui-event-handlers.test.ts +++ b/src/tui/tui-event-handlers.test.ts @@ -226,6 +226,56 @@ describe("tui-event-handlers: handleAgentEvent", () => { expect(state.activeChatRunId).toBeNull(); }); + it("invalidates old global-agent run ownership before accepting new-agent events", () => { + const { state, chatLog, handleAgentEvent, dispose } = createHandlersHarness({ + state: { + currentSessionKey: "global", + currentAgentId: "work", + activeChatRunId: "run-work", + }, + }); + handleAgentEvent({ + runId: "run-work", + sessionKey: "global", + agentId: "work", + stream: "lifecycle", + data: { phase: "start" }, + }); + + state.currentAgentId = "main"; + dispose(); + state.activeChatRunId = null; + handleAgentEvent({ + runId: "run-work", + sessionKey: "global", + agentId: "work", + stream: "tool", + data: { phase: "start", toolCallId: "stale-tool", name: "exec", args: {} }, + }); + + expect(state.activeChatRunId).toBeNull(); + expect(chatLog.startTool).not.toHaveBeenCalled(); + }); + + it("retires a reconnect run immediately when history proves it is absent", () => { + const { state, reconnectStreamingWatchdog, handleChatEvent, chatLog, setActivityStatus } = + createHandlersHarness({ + state: { activeChatRunId: "run-stale", activityStatus: "streaming" }, + }); + + reconnectStreamingWatchdog(null); + handleChatEvent({ + runId: "run-stale", + sessionKey: "agent:main:main", + state: "delta", + message: { role: "assistant", content: "late stale output" }, + }); + + expect(state.activeChatRunId).toBeNull(); + expect(setActivityStatus).toHaveBeenCalledWith("idle"); + expect(chatLog.updateAssistant).not.toHaveBeenCalled(); + }); + it("processes tool events when runId matches activeChatRunId (even if sessionId differs)", () => { const { chatLog, tui, handleAgentEvent } = createHandlersHarness({ state: { currentSessionId: "session-xyz", activeChatRunId: "run-123" }, diff --git a/src/tui/tui-event-handlers.ts b/src/tui/tui-event-handlers.ts index 938621489140..c9e1413982ec 100644 --- a/src/tui/tui-event-handlers.ts +++ b/src/tui/tui-event-handlers.ts @@ -471,6 +471,7 @@ export function createEventHandlers(context: EventHandlerContext) { finalizedRunIds, displayedRunIds, } = collectTrackedSessionRunIds(); + state.sessionGeneration = (state.sessionGeneration ?? 0) + 1; // Reduce the old epoch before adopting its replacement ID; otherwise the // canonical reducer correctly rejects the reset as a foreign session. reduceTuiSessionProjection(state, { diff --git a/src/tui/tui-pty-harness.e2e.test.ts b/src/tui/tui-pty-harness.e2e.test.ts index c77b6f95aa1b..49505e46b2de 100644 --- a/src/tui/tui-pty-harness.e2e.test.ts +++ b/src/tui/tui-pty-harness.e2e.test.ts @@ -427,6 +427,12 @@ describe.sequential("TUI PTY harness", () => { await gapFixture.run.write("history gap proof\r"); await gapFixture.waitForLogEntry((entry) => entry.method === "gapHistoryRecovered"); await gapFixture.run.waitForOutput("PTY_GAP_RECOVERED"); + const gapNotice = "gateway event gap: expected 4, got 5"; + await gapFixture.run.waitForOutput(gapNotice); + const recoveredOutput = gapFixture.run.visibleOutput(); + expect(recoveredOutput.lastIndexOf(gapNotice)).toBeGreaterThan( + recoveredOutput.lastIndexOf("PTY_GAP_RECOVERED"), + ); await gapFixture.run.write("after gap recovery proof\r"); await gapFixture.waitForLogEntry( diff --git a/src/tui/tui-pty-local.e2e.test.ts b/src/tui/tui-pty-local.e2e.test.ts index 1ebf38d1cf7c..89d2f7159951 100644 --- a/src/tui/tui-pty-local.e2e.test.ts +++ b/src/tui/tui-pty-local.e2e.test.ts @@ -1177,6 +1177,11 @@ describe("TUI PTY real backends", () => { await fixture.gateway.startGateway(); gatewayStopped = false; await waitForOutputAfter(fixture.run, "gateway reconnected", reconnectOffset); + await waitForOutputAfter( + fixture.run, + "gateway reconnected after transport loss", + reconnectOffset, + ); await fixture.run.write("\r", { delay: false }); await waitFor({ timeoutMs: LOCAL_OUTPUT_TIMEOUT_MS, diff --git a/src/tui/tui-run-lifecycle.ts b/src/tui/tui-run-lifecycle.ts index b8930419eb3b..71bef9833a2d 100644 --- a/src/tui/tui-run-lifecycle.ts +++ b/src/tui/tui-run-lifecycle.ts @@ -250,7 +250,7 @@ export function createTuiRunLifecycle(context: TuiRunLifecycleContext) { flushPendingHistoryRefreshIfIdle(); }; - const reconnectStreamingWatchdog = () => { + const reconnectStreamingWatchdog = (historyInFlightRunId?: string | null) => { clearStreamingWatchdog(); const activeRunId = state.activeChatRunId; if (!activeRunId) { @@ -258,6 +258,14 @@ export function createTuiRunLifecycle(context: TuiRunLifecycleContext) { clearStaleStreamingIfNoTrackedRunRemains(); return; } + if (historyInFlightRunId === null) { + runCoordinator.noteFinalizedRun(activeRunId, { displayedFinal: true }); + state.activeChatRunId = null; + clearPendingTerminalLifecycleError(activeRunId); + setActivityStatus("idle"); + flushPendingHistoryRefreshIfIdle(); + return; + } if (!sessionRuns.has(activeRunId)) { reconnectPendingRunId = null; state.activeChatRunId = null; @@ -392,9 +400,7 @@ export function createTuiRunLifecycle(context: TuiRunLifecycleContext) { }; const dispose = () => { - runCoordinator.clear(); - clearStreamingWatchdog(); - clearPendingTerminalLifecycleErrors(); + clearTrackedRunState(); }; return { diff --git a/src/tui/tui-session-actions.ts b/src/tui/tui-session-actions.ts index 2c3afa6f6586..c3536aa1dff7 100644 --- a/src/tui/tui-session-actions.ts +++ b/src/tui/tui-session-actions.ts @@ -432,6 +432,7 @@ export function createSessionActions(context: SessionActionContext) { if (!result?.entry || !isCurrentSessionSelection(requestSelection)) { return false; } + state.sessionGeneration = (state.sessionGeneration ?? 0) + 1; reduceTuiSessionProjection(state, { type: "sessionReset", scope: readTuiSessionProjectionScope(state), diff --git a/src/tui/tui-types.ts b/src/tui/tui-types.ts index 9e90de533be9..3561a119a0d2 100644 --- a/src/tui/tui-types.ts +++ b/src/tui/tui-types.ts @@ -183,6 +183,7 @@ export type TuiStateAccess = { currentAgentId: string; currentSessionKey: string; currentSessionId: string | null; + sessionGeneration?: number; sessionProjection?: SessionProjectionState; activeChatRunId: string | null; pendingSubmit: TuiPendingSubmit | null; diff --git a/src/tui/tui.test.ts b/src/tui/tui.test.ts index 83d4412c0c93..500eea82dbae 100644 --- a/src/tui/tui.test.ts +++ b/src/tui/tui.test.ts @@ -410,7 +410,7 @@ describe("resolveTuiCtrlCAction", () => { it("exits immediately after a gateway disconnect", () => { expect( resolveTuiCtrlCAction({ - hasInput: true, + hasInput: false, now: 2000, lastCtrlCAt: 0, wasDisconnected: true, @@ -421,10 +421,24 @@ describe("resolveTuiCtrlCAction", () => { }); }); + it("clears a nonempty draft before exiting after a gateway disconnect", () => { + expect( + resolveTuiCtrlCAction({ + hasInput: true, + now: 2000, + lastCtrlCAt: 0, + wasDisconnected: true, + }), + ).toEqual({ + action: "clear", + nextLastCtrlCAt: 2000, + }); + }); + it("forces exit when shutdown is already in progress", () => { expect( resolveTuiCtrlCAction({ - hasInput: false, + hasInput: true, now: 2000, lastCtrlCAt: 1000, exitRequested: true, diff --git a/src/tui/tui.ts b/src/tui/tui.ts index 9569b835f423..72a476214110 100644 --- a/src/tui/tui.ts +++ b/src/tui/tui.ts @@ -572,6 +572,9 @@ export function resolveTuiCtrlCAction(params: { if (params.exitRequested === true) { return { action: "force-exit", nextLastCtrlCAt: params.lastCtrlCAt }; } + if (params.hasInput) { + return resolveCtrlCAction(params); + } if (params.wasDisconnected === true) { return { action: "exit", nextLastCtrlCAt: params.lastCtrlCAt }; } @@ -605,6 +608,8 @@ export async function runTui(opts: RunTuiOptions): Promise { let initialSessionApplied = false; let rememberedSessionApplied = false; let currentSessionId: string | null = null; + const sessionGenerations = new Map(); + const sessionIds = new Map(); let activeChatRunId: string | null = null; let pendingSubmit: TuiPendingSubmit | null = null; let historyLoaded = false; @@ -637,6 +642,16 @@ export async function runTui(opts: RunTuiOptions): Promise { let statusTimer: NodeJS.Timeout | null = null; let statusStartedAt: number | null = null; let lastActivityStatus = activityStatus; + let invalidateSessionRunOwnership: () => void = () => undefined; + let retireHistoryAbsentRun: (_runId: string) => void = () => undefined; + + const currentSessionGenerationKey = (): string => + JSON.stringify([currentAgentId, currentSessionKey]); + const readCurrentSessionGeneration = () => + sessionGenerations.get(currentSessionGenerationKey()) ?? 0; + const writeCurrentSessionGeneration = (value: number) => { + sessionGenerations.set(currentSessionGenerationKey(), value); + }; const state: TuiStateAccess = { get agentDefaultId() { @@ -667,7 +682,11 @@ export async function runTui(opts: RunTuiOptions): Promise { return currentAgentId; }, set currentAgentId(value) { + if (currentAgentId === value) { + return; + } currentAgentId = value; + invalidateSessionRunOwnership(); pluginApprovals?.sessionChanged(); taskSuggestions?.sessionChanged(); }, @@ -683,8 +702,23 @@ export async function runTui(opts: RunTuiOptions): Promise { return currentSessionId; }, set currentSessionId(value) { + if (value) { + const generationKey = currentSessionGenerationKey(); + const previousSessionId = sessionIds.get(generationKey); + // The first ID binds an unresolved selection; reset/replacement owners bump explicitly. + if (previousSessionId && previousSessionId !== value) { + writeCurrentSessionGeneration(readCurrentSessionGeneration() + 1); + } + sessionIds.set(generationKey, value); + } currentSessionId = value; }, + get sessionGeneration() { + return readCurrentSessionGeneration(); + }, + set sessionGeneration(value) { + writeCurrentSessionGeneration(Math.max(readCurrentSessionGeneration(), value)); + }, get activeChatRunId() { return activeChatRunId; }, @@ -850,6 +884,19 @@ export async function runTui(opts: RunTuiOptions): Promise { const statusContainer = new Container(); const footer = new Text("", 1, 0); const chatLog = new ChatLog(); + const connectionNotices: string[] = []; + const addConnectionNotice = (text: string) => { + connectionNotices.push(text); + if (connectionNotices.length > 12) { + connectionNotices.shift(); + } + chatLog.addSystem(text, { coalesceConsecutive: true }); + }; + const restoreConnectionNotices = () => { + for (const notice of connectionNotices) { + chatLog.addSystem(notice, { coalesceConsecutive: true }); + } + }; const editor = new CustomEditor(tui, editorTheme); const root = new Container(); root.addChild(header); @@ -1381,10 +1428,27 @@ export async function runTui(opts: RunTuiOptions): Promise { refreshSessionInfo, applySessionInfoFromPatch, applySessionMutationResult, - loadHistory, + loadHistory: loadHistorySnapshot, setSession, abortActive, } = sessionActions; + const loadHistory = async (options?: { retireMissingReconnectRun?: boolean }) => { + const activeRunAtStart = state.activeChatRunId; + const result = await loadHistorySnapshot(); + if (result.loaded) { + if ( + options?.retireMissingReconnectRun === true && + activeRunAtStart && + !result.inFlightRunId && + activeRunAtStart === state.activeChatRunId + ) { + retireHistoryAbsentRun(activeRunAtStart); + } + restoreConnectionNotices(); + tui.requestRender(); + } + return result; + }; const taskSuggestions = createTuiTaskSuggestionController({ client, chatLog, @@ -1426,6 +1490,12 @@ export async function runTui(opts: RunTuiOptions): Promise { forgetLocalBtwRunId, clearLocalBtwRunIds, }); + retireHistoryAbsentRun = () => reconnectStreamingWatchdog(null); + invalidateSessionRunOwnership = () => { + disposeEventHandlers(); + state.activeChatRunId = null; + setActivityStatus("idle"); + }; const deferredFinish = createDeferredTuiFinish(); const forceExit = () => { @@ -1560,7 +1630,7 @@ export async function runTui(opts: RunTuiOptions): Promise { const handleCtrlC = () => { const now = Date.now(); const decision = resolveTuiCtrlCAction({ - hasInput: editor.getText().trim().length > 0, + hasInput: editor.getText().length > 0, now, lastCtrlCAt, exitRequested, @@ -1737,13 +1807,16 @@ export async function runTui(opts: RunTuiOptions): Promise { if (!ownsConnection()) { return; } - await loadHistory(); + await loadHistory({ retireMissingReconnectRun: reconnected }); if (!ownsConnection()) { return; } if (activityStatus === "starting up") { setActivityStatus("idle"); } + if (reconnected) { + addConnectionNotice("gateway reconnected after transport loss"); + } setConnectionStatus( isLocalMode ? "local ready" : reconnected ? "gateway reconnected" : "gateway connected", 4000, @@ -1773,7 +1846,7 @@ export async function runTui(opts: RunTuiOptions): Promise { }); }; - client.onDisconnected = (reason) => { + const handleBackendDisconnected = (reason: string) => { if (exitRequested) { return; } @@ -1805,12 +1878,17 @@ export async function runTui(opts: RunTuiOptions): Promise { updateFooter(); tui.requestRender(); }; + client.onConnectError = (error) => { + handleBackendDisconnected(formatTuiErrorMessage(error)); + }; + client.onDisconnected = handleBackendDisconnected; client.onGap = (info) => { if (exitRequested || !isConnected) { return; } setConnectionStatus(`event gap: expected ${info.expected}, got ${info.received}`, 5000); + addConnectionNotice(`gateway event gap: expected ${info.expected}, got ${info.received}`); reconcileHistoryAfterGap(); void (async () => { try {