diff --git a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts index c5b865ed70..6f6aa9b1fa 100644 --- a/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts +++ b/apps/desktop/src/main/__tests__/runtime-host-session-execution-ipc-main.test.ts @@ -58,6 +58,102 @@ test('registers Session observation as one reconnectable operation', () => { assert.equal(ipc.reconnectableChannels.has('sessions:observe'), true); }); +test('treats pending Session observation teardown as IPC cancellation', async () => { + const observations = new RuntimeHostSessionObservationRegistry(); + const ipc = observationIpcHarness(observations); + + const observing = ipc.invoke('sessions:observe', 'session-1', 'observer-1'); + await Promise.resolve(); + await observations.unobserve('observer-1'); + assert.deepEqual(await observing, { kind: 'cancelled' }); + assert.deepEqual(observations.trackedSessionIds(), []); + await observations.close(); +}); + +test('treats pending transcript teardown as IPC cancellation', async () => { + const observations = new RuntimeHostSessionObservationRegistry(); + const ipc = observationIpcHarness(observations); + + const opening = ipc.invoke('sessions:transcript:open', 'session-1', 'consumer-1'); + await Promise.resolve(); + await observations.closeTranscript('consumer-1', 9); + assert.deepEqual(await opening, { kind: 'cancelled' }); + assert.deepEqual(observations.trackedSessionIds(), []); + await observations.close(); +}); + +test('preserves genuine Session observation initialization failures', async () => { + const observations = new RuntimeHostSessionObservationRegistry(); + const sessionFailure = new Error('seed failed'); + const transcriptFailure = new Error('transcript open failed'); + await observations.attach({ + async observe() { + throw sessionFailure; + }, + async unobserve() {}, + async openTranscript() { + throw transcriptFailure; + }, + async loadTranscriptBefore() {}, + async loadTranscriptAround() {}, + async closeTranscript() {}, + }); + const ipc = observationIpcHarness(observations); + + await assert.rejects( + ipc.invoke('sessions:observe', 'session-1', 'observer-1'), + (error) => error === sessionFailure, + ); + await assert.rejects( + ipc.invoke('sessions:transcript:open', 'session-1', 'consumer-1'), + (error) => error === transcriptFailure, + ); + await observations.close(); +}); + +test('returns explicit ready results for Session observation IPC', async () => { + const observations = new RuntimeHostSessionObservationRegistry(); + const transcript = { + sessionId: 'session-1', + generation: 'generation-1', + hostEpoch: 'host-epoch-1', + readThroughMessageId: null, + }; + await observations.attach({ + async observe() {}, + async unobserve() {}, + async openTranscript() { + return transcript; + }, + async loadTranscriptBefore() {}, + async loadTranscriptAround() {}, + async closeTranscript() {}, + }); + const ipc = observationIpcHarness(observations); + + assert.deepEqual(await ipc.invoke('sessions:observe', 'session-1', 'observer-1'), { + kind: 'ready', + value: undefined, + }); + assert.deepEqual(await ipc.invoke('sessions:transcript:open', 'session-1', 'consumer-1'), { + kind: 'ready', + value: transcript, + }); + await observations.close(); +}); + +function observationIpcHarness(observations: RuntimeHostSessionObservationRegistry) { + const ipc = ipcHarness(); + registerRuntimeHostSessionObservationIpc( + { + observations, + resolveSideConversation: async () => false, + }, + ipc, + ); + return ipc; +} + test("keeps synthetic E2E interactions visible through Host hydration and retires their answer", async () => { const observer = observerWithSnapshot(); const ipc = ipcHarness(); diff --git a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts index 6e21583524..5f44f8d5d5 100644 --- a/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts +++ b/apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts @@ -53,7 +53,11 @@ import { } from "./ipc-reconnect-policy.js"; import type { DesktopRuntimeHostClient } from "./runtime-host-client.js"; import type { SessionCopyCleanupAuthority } from '@maka/storage/session-copy-cleanup'; -import type { RuntimeHostSessionObservationRegistry } from "./runtime-host-session-observation-registry.js"; +import type { RuntimeHostObservationIpcResult } from '../shared/runtime-host-observation-ipc.js'; +import { + RuntimeHostObservationCancelledError, + type RuntimeHostSessionObservationRegistry, +} from "./runtime-host-session-observation-registry.js"; import { RuntimeHostSessionObserver, type RuntimeHostSessionObserverTarget, @@ -193,21 +197,25 @@ export function registerRuntimeHostSessionObservationIpc( 'sessions:observe', async (event, sessionId: unknown, observerId: unknown) => { const normalizedSessionId = requiredId(sessionId, 'Session'); - await deps.observations.observe( - normalizedSessionId, - requiredId(observerId, 'Session observer'), - event.sender as RuntimeHostSessionObserverTarget, - await deps.resolveSideConversation(normalizedSessionId), + return observationIpcResult( + deps.observations.observe( + normalizedSessionId, + requiredId(observerId, 'Session observer'), + event.sender as RuntimeHostSessionObserverTarget, + await deps.resolveSideConversation(normalizedSessionId), + ), ); }, ); ipcMain.handle( 'sessions:transcript:open', async (event, sessionId: unknown, consumerId: unknown) => - deps.observations.openTranscript( - requiredId(sessionId, 'Session'), - requiredId(consumerId, 'Transcript consumer'), - event.sender as RuntimeHostTranscriptTarget, + observationIpcResult( + deps.observations.openTranscript( + requiredId(sessionId, 'Session'), + requiredId(consumerId, 'Transcript consumer'), + event.sender as RuntimeHostTranscriptTarget, + ), ), ); ipcMain.handle('sessions:transcript:load-before', async (event, input: unknown) => { @@ -229,6 +237,17 @@ export function registerRuntimeHostSessionObservationIpc( } } +async function observationIpcResult( + operation: Promise, +): Promise> { + try { + return { kind: 'ready', value: await operation }; + } catch (error) { + if (error instanceof RuntimeHostObservationCancelledError) return { kind: 'cancelled' }; + throw error; + } +} + /** * Project Host-owned Session execution onto the Desktop renderer IPC contract. * The adapter owns client validation and presentation events, never Runtime diff --git a/apps/desktop/src/main/runtime-host-session-observation-registry.ts b/apps/desktop/src/main/runtime-host-session-observation-registry.ts index 81e650240b..85c4cb3905 100644 --- a/apps/desktop/src/main/runtime-host-session-observation-registry.ts +++ b/apps/desktop/src/main/runtime-host-session-observation-registry.ts @@ -60,6 +60,10 @@ interface ObservationReadiness { reject(error: Error): void; } +export class RuntimeHostObservationCancelledError extends Error { + readonly name = 'RuntimeHostObservationCancelledError'; +} + function requireTranscriptSource( source: SessionObservationSource | undefined, ): SessionObservationSource & TranscriptSource { @@ -210,10 +214,10 @@ export class RuntimeHostSessionObservationRegistry { ([, registration]) => registration.sessionId === sessionId, ); for (const [observerId, registration] of observations) { - this.#deleteRegistration(observerId, registration); + this.#cancelRegistration(observerId, registration); } for (const [consumerId, registration] of transcripts) { - this.#deleteTranscript(consumerId, registration); + this.#cancelTranscript(consumerId, registration); } if (source) { await Promise.allSettled([ @@ -232,10 +236,10 @@ export class RuntimeHostSessionObservationRegistry { ([, registration]) => registration.target.id === targetId, ); for (const [observerId, registration] of observations) { - this.#deleteRegistration(observerId, registration); + this.#cancelRegistration(observerId, registration); } for (const [consumerId, registration] of transcripts) { - this.#deleteTranscript(consumerId, registration); + this.#cancelTranscript(consumerId, registration); } if (!source) return; @@ -427,7 +431,7 @@ export class RuntimeHostSessionObservationRegistry { if (targetId !== undefined && registration.target.id !== targetId) { throw new Error('Desktop transcript consumer belongs to another renderer'); } - this.#deleteTranscript(consumerId, registration); + this.#cancelTranscript(consumerId, registration); await this.#source?.closeTranscript?.(consumerId, targetId); } @@ -439,15 +443,11 @@ export class RuntimeHostSessionObservationRegistry { this.#bindTarget = (target) => target; const registrations = [...this.#registrations]; const transcripts = [...this.#transcripts]; - this.#registrations.clear(); - for (const [, registration] of registrations) { - registration.target.off("destroyed", registration.destroyedListener); - registration.ready.reject( - new Error("Session observation ended before it became ready"), - ); + for (const [observerId, registration] of registrations) { + this.#cancelRegistration(observerId, registration); } for (const [consumerId, registration] of transcripts) { - this.#deleteTranscript(consumerId, registration); + this.#cancelTranscript(consumerId, registration); } if (source) { await Promise.allSettled([ @@ -460,20 +460,32 @@ export class RuntimeHostSessionObservationRegistry { async #remove(observerId: string): Promise { const registration = this.#registrations.get(observerId); if (!registration) return; - this.#deleteRegistration(observerId, registration); + this.#cancelRegistration(observerId, registration); await this.#source?.unobserve(observerId); } + #cancelRegistration( + observerId: string, + registration: SessionObservationRegistration, + ): void { + this.#deleteRegistration( + observerId, + registration, + new RuntimeHostObservationCancelledError( + "Session observation ended before it became ready", + ), + ); + } + #deleteRegistration( observerId: string, registration: SessionObservationRegistration, + error = new Error("Session observation ended before it became ready"), ): void { if (this.#registrations.get(observerId) !== registration) return; this.#registrations.delete(observerId); registration.target.off("destroyed", registration.destroyedListener); - registration.ready.reject( - new Error("Session observation ended before it became ready"), - ); + registration.ready.reject(error); } async #runTranscriptOperation( @@ -562,12 +574,26 @@ export class RuntimeHostSessionObservationRegistry { } } - #deleteTranscript(consumerId: string, registration: TranscriptRegistration): void { + #cancelTranscript(consumerId: string, registration: TranscriptRegistration): void { + this.#deleteTranscript( + consumerId, + registration, + new RuntimeHostObservationCancelledError( + 'Transcript observation ended before it became ready', + ), + ); + } + + #deleteTranscript( + consumerId: string, + registration: TranscriptRegistration, + error = new Error('Transcript observation ended before it became ready'), + ): void { if (this.#transcripts.get(consumerId) !== registration) return; this.#transcripts.delete(consumerId); registration.restore?.resolve(); registration.target.off('destroyed', registration.destroyedListener); - registration.ready.reject(new Error('Transcript observation ended before it became ready')); + registration.ready.reject(error); } #bindTranscriptTarget(target: RuntimeHostTranscriptTarget): RuntimeHostTranscriptTarget { diff --git a/apps/desktop/src/preload/preload.ts b/apps/desktop/src/preload/preload.ts index 10949441a2..11826f6f86 100644 --- a/apps/desktop/src/preload/preload.ts +++ b/apps/desktop/src/preload/preload.ts @@ -73,6 +73,7 @@ import type { AppIconSelectResult, } from './bridge-contract.js'; import type { ExternalSessionImportIpcResult } from './external-session-import-result.js'; +import type { RuntimeHostObservationIpcResult } from '../shared/runtime-host-observation-ipc.js'; import { projectDesktopExternalSessionCatalogItem, type DesktopExternalSessionCatalogItem, @@ -2148,7 +2149,11 @@ const makaBridge = { let unsubscribeEvents = () => {}; let unsubscribeObservationSeed = () => {}; const observeDispatch = runtimeHostSessionRef(sessionId).then((session) => { - if (disposed) return { completion: Promise.resolve() }; + if (disposed) { + return { + completion: Promise.resolve({ kind: 'cancelled' } as const), + }; + } const profileId = runtimeHostMetadataFor(session.scope)?.profileId; if (!profileId) throw new Error('The Runtime Host profile for this task is unavailable'); // Keep the renderer listener across Host target epochs. The observer @@ -2178,13 +2183,19 @@ const makaBridge = { session.scope, session.sessionId, observerId, - ), + ) as Promise>, }; }); const observing = observeDispatch.then(({ completion }) => completion); void observing.then( - () => { - if (!disposed) onSeeded?.(); + (result) => { + if (result.kind === 'cancelled') { + disposed = true; + unsubscribeObservationSeed(); + unsubscribeEvents(); + } else if (!disposed) { + onSeeded?.(); + } }, (error: unknown) => { if (!disposed) onSeedError?.(error); @@ -2367,7 +2378,7 @@ const makaBridge = { session.scope, session.sessionId, consumerId, - ) as Promise, + ) as Promise>, }; }); let closeTask: Promise | undefined; @@ -2381,14 +2392,20 @@ const makaBridge = { void closeTask.catch(() => undefined); }; registerCancellation?.(requestClose); - let opened: DesktopTranscriptOpenResult; + let openResult: RuntimeHostObservationIpcResult; try { - opened = await openDispatch.then(({ completion }) => completion); + openResult = await openDispatch.then(({ completion }) => completion); } catch (error) { closed = true; ipcRenderer.off(channel, listener); throw error; } + if (openResult.kind === 'cancelled') { + closed = true; + ipcRenderer.off(channel, listener); + throw new Error('Desktop transcript open was cancelled'); + } + const opened = openResult.value; if (closed) throw new Error('Desktop transcript open was cancelled'); identity ??= { generation: opened.generation, hostEpoch: opened.hostEpoch }; const range = ( diff --git a/apps/desktop/src/shared/runtime-host-observation-ipc.ts b/apps/desktop/src/shared/runtime-host-observation-ipc.ts new file mode 100644 index 0000000000..fe45a3b07f --- /dev/null +++ b/apps/desktop/src/shared/runtime-host-observation-ipc.ts @@ -0,0 +1,22 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +export type RuntimeHostObservationIpcResult = + | { readonly kind: 'ready'; readonly value: T } + | { readonly kind: 'cancelled' };