Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import type {
ConnectOrSpawnRuntimeHostInput,
RuntimeHostConnection,
} from '@maka/runtime-host/client';
import { RuntimeHostOperationError } from '@maka/runtime-host/client';
import {
SESSION_CONTINUITY_SCHEMA_VERSION,
type ClientCapabilityCallFrame,
Expand Down Expand Up @@ -895,6 +896,69 @@ test('drops a stale shared Session observation when Guest access is gone', async
await observations.close();
});

test('forgets an observed Session the Host no longer serves instead of blocking every reconnect', async () => {
const observations = new RuntimeHostSessionObservationRegistry();
const firstIpc = ipcHarness();
const firstHost = connectionHarness('missing-session-source', {
sessionId: 'session-1',
subscriptionSnapshot: continuitySnapshot(),
});
const firstCandidate = await createDesktopRuntimeHostCandidate(
firstHost.connection,
deps(firstIpc),
observations,
);
await firstIpc.invoke('sessions:observe', 'session-1', 'observer-1');
await firstCandidate.close();

// The replacement Host no longer serves session-1: subscription.open
// deterministically answers not_found.
const changes: Array<{ reason: string; sessionId?: string }> = [];
const missingHost = connectionHarness('missing-session-host', {
sessionId: 'session-1',
subscriptionError: new RuntimeHostOperationError(
'subscription.open',
'not_found',
'Runtime Host Session was not found',
),
});
const secondCandidate = await createDesktopRuntimeHostCandidate(
missingHost.connection,
{
...deps(ipcHarness()),
emitSessionsChanged: (_scope, reason, sessionId) => {
changes.push({ reason, ...(sessionId === undefined ? {} : { sessionId }) });
},
},
observations,
);

// The stale active registration is forgotten instead of failing the
// candidate start, and the renderer is told to drop the Session view.
assert.deepEqual(observations.observedSessionIds(), []);
assert.ok(
changes.some(
({ reason, sessionId }) => reason === 'deleted' && sessionId === 'session-1',
),
);
await secondCandidate.close();

// A later reconnect observes new Sessions on the same registry.
const thirdIpc = ipcHarness();
const thirdHost = connectionHarness('missing-session-recovered', {
sessionId: 'session-2',
});
const thirdCandidate = await createDesktopRuntimeHostCandidate(
thirdHost.connection,
deps(thirdIpc),
observations,
);
await thirdIpc.invoke('sessions:observe', 'session-2', 'observer-2');
assert.deepEqual(observations.observedSessionIds(), ['session-2']);
await thirdCandidate.close();
await observations.close();
});

type IpcHandler = Parameters<Pick<IpcMain, 'handle'>['handle']>[1];

function ipcHarness(onSend?: (channel: string, payload: unknown) => void) {
Expand Down
9 changes: 6 additions & 3 deletions apps/desktop/src/main/runtime-host-desktop-candidate.ts
Original file line number Diff line number Diff line change
Expand Up @@ -703,11 +703,14 @@ export async function createDesktopRuntimeHostCandidate(
once: target.once.bind(target),
off: target.off.bind(target),
}),
(missingSessionId) => emitSessionsChanged("deleted", missingSessionId),
);
const restoredSessionIdSet = new Set(restoredSessionIds);
const failedSessionIds = observedSessionIds.filter(
(sessionId) => !restoredSessionIdSet.has(sessionId),
);
// Attach forgets Sessions the Host no longer serves, so only Sessions
// that are still registered but failed to restore count as failures.
const failedSessionIds = sessionObservations
.observedSessionIds()
.filter((sessionId) => !restoredSessionIdSet.has(sessionId));
if (failedSessionIds.length > 0) {
throw new Error(
`Failed to restore Session observations: ${failedSessionIds.join(', ')}`,
Expand Down
25 changes: 25 additions & 0 deletions apps/desktop/src/main/runtime-host-session-observation-registry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
* under the License.
*/

import { RuntimeHostOperationError } from "@maka/runtime-host/client";
import type {
RuntimeHostSessionObserver,
RuntimeHostRendererTarget,
Expand Down Expand Up @@ -74,6 +75,20 @@ function requireTranscriptSource(
return source as SessionObservationSource & TranscriptSource;
}

/**
* A `subscription.open`/`not_found` answer is deterministic: the Host no
* longer serves this Session (Host restart with ephemeral state, Session GC,
* or deletion by another client). Unlike `session.transcript.page`/`not_found`
* (see `isRecoverableSubscriptionFailure` in the subscription owner), there is
* nothing to retry — the registration must be forgotten instead of blocking
* every reconnect.
*/
function isMissingRuntimeHostSessionError(error: unknown): boolean {
if (!(error instanceof RuntimeHostOperationError)) return false;
if (error.operation !== "subscription.open") return false;
return error.code === "not_found";
}

interface SessionObservationRegistration {
readonly sessionId: string;
readonly messageAdmissions: boolean;
Expand Down Expand Up @@ -139,6 +154,7 @@ export class RuntimeHostSessionObservationRegistry {
async attach(
source: SessionObservationSource,
bindTarget: ObservationTargetBinding = (target) => target,
onSessionMissing?: (sessionId: string) => void,
): Promise<string[]> {
this.#assertOpen();
if (this.#source && this.#source !== source) {
Expand Down Expand Up @@ -176,6 +192,15 @@ export class RuntimeHostSessionObservationRegistry {
this.#source === source &&
this.#registrations.get(observerId) === registration
) {
if (isMissingRuntimeHostSessionError(error)) {
// The Host no longer serves this Session. Forget the
// registration regardless of lifecycle so the stale entry
// cannot fail every future reconnect, and let the upper layer
// drop the Session view.
onSessionMissing?.(registration.sessionId);
this.#deleteRegistration(observerId, registration);
return undefined;
}
if (registration.lifecycle === "pending") {
this.#deleteRegistration(observerId, registration);
}
Expand Down