From 97105e7d3cf36e8df90b7dcf0445577c4a618838 Mon Sep 17 00:00:00 2001 From: testikun Date: Tue, 25 Aug 2026 11:28:54 +0800 Subject: [PATCH] fix(desktop): recover stale transcript subscriptions --- .../runtime-host-session-observer.test.ts | 65 ++++++++++++++++++- ...runtime-host-session-subscription-owner.ts | 8 ++- 2 files changed, 71 insertions(+), 2 deletions(-) diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts index 18cabd68ee..29e4ebcde1 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts @@ -28,7 +28,10 @@ import { type SessionTranscriptPage, type SubscriptionFrame, } from "@maka/runtime-host/protocol"; -import { RuntimeHostSubscriptionError } from "@maka/runtime-host/client"; +import { + RuntimeHostOperationError, + RuntimeHostSubscriptionError, +} from "@maka/runtime-host/client"; import type { DesktopRuntimeHostSession } from "../runtime-host-client.js"; import { DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES, @@ -1622,6 +1625,66 @@ test("reopens an evicted active subscription without a renderer resubscribe", as await observer.close(); }); +test("recovers when transcript paging loses the active subscription", async () => { + const firstEvents = new AsyncFrameQueue(); + const secondEvents = new AsyncFrameQueue(); + const target = eventTarget(12); + let openCount = 0; + const observer = new RuntimeHostSessionObserver({ + client: { + openSession: async () => { + openCount += 1; + const first = openCount === 1; + const events = first ? firstEvents : secondEvents; + return runtimeHostSessionFixture({ + snapshot: continuitySnapshot(), + transcript: Promise.resolve([]), + events, + loadTranscriptPage: async () => { + if (first) { + throw new RuntimeHostOperationError( + "session.transcript.page", + "not_found", + "Session subscription was not found", + ); + } + return { + kind: "page", + sessionId: "session-1", + source: "durable", + direction: "newer", + throughSequence: 0, + rawBytes: 0, + fragments: [], + nextCursor: null, + }; + }, + async close() { + events.end(); + }, + }); + }, + }, + emitSessionsChanged() {}, + }); + + await observer.observe("session-1", "observer-1", target); + firstEvents.push({ + kind: "subscription.transcript_advanced", + hostEpoch: "host-1", + subscriptionId: "subscription-session-1", + sessionId: "session-1", + sequence: 1, + throughSequence: 0, + }); + await waitFor(() => openCount === 2); + + secondEvents.push(deltaFrame(1, 0, "recovered")); + await waitFor(() => target.events.some((event) => event.type === "text_delta")); + assert.equal(target.events.some((event) => event.type === "error"), false); + await observer.close(); +}); + test('does not activate a refresh candidate that fails during commit preparation', async () => { const firstEvents = new AsyncFrameQueue(); const secondEvents = new AsyncFrameQueue(); diff --git a/apps/desktop/src/main/runtime-host-session-subscription-owner.ts b/apps/desktop/src/main/runtime-host-session-subscription-owner.ts index 70908c3087..21d95402db 100644 --- a/apps/desktop/src/main/runtime-host-session-subscription-owner.ts +++ b/apps/desktop/src/main/runtime-host-session-subscription-owner.ts @@ -17,7 +17,10 @@ * under the License. */ -import { RuntimeHostSubscriptionError } from "@maka/runtime-host/client"; +import { + RuntimeHostOperationError, + RuntimeHostSubscriptionError, +} from "@maka/runtime-host/client"; import type { SessionAssistantStreamIdentity, SessionContinuitySnapshot, @@ -414,6 +417,9 @@ function subscriptionClosedError( } function isRecoverableSubscriptionFailure(error: unknown): boolean { + if (error instanceof RuntimeHostOperationError) { + return error.operation === "session.transcript.page" && error.code === "not_found"; + } if (!(error instanceof RuntimeHostSubscriptionError)) return false; return ( error.reason === "slow_consumer" ||