fix(desktop): recover Runtime Host reads after reconnect

apps/desktop/src/main/__tests__/runtime-host-session-observer.test.ts

@@ -511,7 +511,8 @@ test('fences transcript range failures to the current registration and Host sour
   });
   const request = (consumerId: string, generation: string) => ({
     consumerId,
-    generation,
+    sessionId: 'session-1',
+    hostEpoch: `host-${generation}`,
     anchorSequence: null,
     maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES,
   });
@@ -620,9 +621,10 @@ test('fences transcript range failures across same-source replica recovery', asy
   const observations = new RuntimeHostSessionObservationRegistry();
   const batches: DesktopTranscriptBatch[] = [];
   const consumerId = 'consumer-replica-recovery';
-  const request = (generation: string) => ({
+  const request = (hostEpoch: string) => ({
     consumerId,
-    generation,
+    sessionId: 'session-1',
+    hostEpoch,
     anchorSequence: 1,
     maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES,
   });
@@ -644,7 +646,7 @@ test('fences transcript range failures across same-source replica recovery', asy
   };
   await observations.attach(observer);
   const opened = await observations.openTranscript('session-1', consumerId, target);
-  const staleLoad = observations.loadTranscriptBefore(request(opened.generation), target.id);
+  const staleLoad = observations.loadTranscriptBefore(request(opened.hostEpoch), target.id);
   await waitFor(() => staleRangeStarted);
 
   firstEvents.push({
@@ -658,9 +660,8 @@ test('fences transcript range failures across same-source replica recovery', asy
   staleRange.reject(new Error('stale replica rejected its range'));
   await assert.doesNotReject(staleLoad);
 
-  const generation = batches.at(-1)!.generation;
   await assert.rejects(
-    observations.loadTranscriptBefore(request(generation), target.id),
+    observations.loadTranscriptBefore(request(opened.hostEpoch), target.id),
     (error) => error === currentFailure,
   );
   await observations.close();
@@ -861,7 +862,7 @@ test('keeps a bounded transcript batch window in flight until the renderer ackno
   await observer.close();
 });
 
-test('finishes transcript open against a replacement that arrives while reset delivery waits', async () => {
+test('finishes transcript open and replays a stale range request after replacement', async () => {
   const firstEvents = new AsyncFrameQueue();
   const secondEvents = new AsyncFrameQueue();
   const message: StoredMessage = {
@@ -873,6 +874,8 @@ test('finishes transcript open against a replacement that arrives while reset de
     modelId: 'test-model',
   };
   let opens = 0;
+  let rangeLoads = 0;
+  const requestedAnchors: Array<number | null> = [];
   const observer = new RuntimeHostSessionObserver({
     client: {
       openSession: async () => {
@@ -882,6 +885,49 @@ test('finishes transcript open against a replacement that arrives while reset de
           snapshot: continuitySnapshot(),
           transcript: Promise.resolve([message]),
           events,
+          transcriptBootstrap: {
+            throughSequence: 0,
+            overlayMessageCount: 0,
+            durable: {
+              kind: 'page',
+              sessionId: 'session-1',
+              source: 'durable',
+              direction: 'older',
+              throughSequence: 0,
+              rawBytes: 1,
+              fragments: [],
+              nextCursor: 'older',
+            },
+            overlay: {
+              kind: 'page',
+              sessionId: 'session-1',
+              source: 'overlay',
+              direction: 'older',
+              throughSequence: null,
+              rawBytes: 0,
+              fragments: [],
+              nextCursor: null,
+            },
+          },
+          loadTranscriptOverlay: async () => [],
+          decodeTranscriptPage: async (page) => ({
+            messages: page.rawBytes === 1 ? [{ identity: 0, message }] : [],
+            nextCursor: page.nextCursor,
+          }),
+          loadTranscriptPage: async (input) => {
+            rangeLoads += 1;
+            requestedAnchors.push(input.anchorSequence);
+            return {
+              kind: 'page',
+              sessionId: 'session-1',
+              source: input.source,
+              direction: input.direction,
+              throughSequence: input.throughSequence,
+              rawBytes: 0,
+              fragments: [],
+              nextCursor: null,
+            };
+          },
           async close() {
             events.end();
           },
@@ -891,10 +937,21 @@ test('finishes transcript open against a replacement that arrives while reset de
     emitSessionsChanged() {},
   });
   const batches: DesktopTranscriptBatch[] = [];
+  let autoAcknowledge = false;
   const opening = observer.openTranscript('session-1', 'consumer-recovery', {
     id: 22,
     send(_channel, batch) {
       batches.push(batch);
+      if (autoAcknowledge) {
+        queueMicrotask(() =>
+          observer.acknowledgeTranscript(
+            'consumer-recovery',
+            batch.generation,
+            batch.deliverySequence,
+            22,
+          ),
+        );
+      }
     },
     once() {},
     off() {},
@@ -905,6 +962,7 @@ test('finishes transcript open against a replacement that arrives while reset de
   );
 
   await waitFor(() => batches.length === 4);
+  const staleGeneration = batches[0]!.generation;
   firstEvents.push({
     kind: 'subscription.closed',
     hostEpoch: 'host-1',
@@ -937,6 +995,58 @@ test('finishes transcript open against a replacement that arrives while reset de
   const opened = await result;
   assert.equal(opened.error, undefined);
   assert.equal(opened.value?.generation, batches.at(-1)?.generation);
+  assert.notEqual(opened.value?.generation, staleGeneration);
+  rangeLoads = 0;
+  requestedAnchors.length = 0;
+  autoAcknowledge = true;
+  // The renderer dispatched this range request before the replacement replica
+  // was installed; the same Session and Host epoch continue the read against
+  // the current replica, and the requested slice is what the replica loads.
+  await assert.doesNotReject(() =>
+    observer.loadTranscriptBefore(
+      {
+        consumerId: 'consumer-recovery',
+        sessionId: 'session-1',
+        hostEpoch: 'host-1',
+        anchorSequence: 0,
+        maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES,
+      },
+      22,
+    ),
+  );
+  assert.equal(rangeLoads, 1);
+  assert.deepEqual(requestedAnchors, [0]);
+  // Durable sequence identity is Session- and Host-epoch-scoped: a request
+  // for a different Session or Host epoch must reject instead of silently
+  // reading a different slice of the transcript.
+  await assert.rejects(
+    () =>
+      observer.loadTranscriptBefore(
+        {
+          consumerId: 'consumer-recovery',
+          sessionId: 'session-1',
+          hostEpoch: 'other-host',
+          anchorSequence: 0,
+          maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES,
+        },
+        22,
+      ),
+    /Desktop transcript host epoch changed/,
+  );
+  await assert.rejects(
+    () =>
+      observer.loadTranscriptBefore(
+        {
+          consumerId: 'consumer-recovery',
+          sessionId: 'other-session',
+          hostEpoch: 'host-1',
+          anchorSequence: 0,
+          maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES,
+        },
+        22,
+      ),
+    /Desktop transcript consumer belongs to another session/,
+  );
   await observer.close();
 });
 
@@ -1227,7 +1337,8 @@ test('keeps a transcript consumer available after a delivery fails', async () =>
     observer.loadTranscriptAround(
       {
         consumerId,
-        generation: opened.generation,
+        sessionId: opened.sessionId,
+        hostEpoch: opened.hostEpoch,
         anchorSequence: 0,
         maxBytes: DESKTOP_TRANSCRIPT_FRAGMENT_MAX_BYTES,
       },

apps/desktop/src/main/__tests__/transcript-identity.test.ts

@@ -0,0 +1,85 @@
+/*
+ * 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.
+ */
+
+import assert from 'node:assert/strict';
+import test from 'node:test';
+import type { DesktopTranscriptBatch } from '../../preload/transcript-contract.js';
+import {
+  adoptTranscriptIdentity,
+  type DesktopTranscriptIdentity,
+} from '../../preload/transcript-identity.js';
+
+function batch(overrides: Partial<DesktopTranscriptBatch> = {}): DesktopTranscriptBatch {
+  return {
+    sessionId: 'session-1',
+    generation: 'generation-1',
+    hostEpoch: 'host-1',
+    durableThrough: null,
+    fragments: [],
+    evictedDurableSequences: [],
+    completedOverlayMessageIds: [],
+    hasOlder: false,
+    hasNewer: false,
+    reset: false,
+    ready: true,
+    deliverySequence: 1,
+    ...overrides,
+  };
+}
+
+test('initializes the tracked identity from the first accepted batch', () => {
+  const first = batch({ generation: 'generation-1', hostEpoch: 'host-1' });
+  assert.deepEqual(adoptTranscriptIdentity(undefined, first), {
+    generation: 'generation-1',
+    hostEpoch: 'host-1',
+  });
+});
+
+test('keeps the current identity for a regular batch', () => {
+  const current: DesktopTranscriptIdentity = { generation: 'generation-1', hostEpoch: 'host-1' };
+  assert.equal(adoptTranscriptIdentity(current, batch()), current);
+});
+
+test('adopts a reset batch identity so the next range request targets the replacement Host', () => {
+  const current: DesktopTranscriptIdentity = { generation: 'generation-1', hostEpoch: 'host-1' };
+  const replacement = batch({
+    generation: 'generation-2',
+    hostEpoch: 'host-2',
+    reset: true,
+  });
+  const next = adoptTranscriptIdentity(current, replacement);
+  assert.deepEqual(next, { generation: 'generation-2', hostEpoch: 'host-2' });
+  // The adopted identity stays in effect for later regular batches.
+  assert.equal(
+    adoptTranscriptIdentity(next, batch({ generation: 'generation-2', hostEpoch: 'host-2' })),
+    next,
+  );
+});
+
+test('does not mutate an identity already captured by a dispatched request', () => {
+  const dispatched: DesktopTranscriptIdentity = { generation: 'generation-1', hostEpoch: 'host-1' };
+  const replacement = batch({
+    generation: 'generation-2',
+    hostEpoch: 'host-2',
+    reset: true,
+  });
+  adoptTranscriptIdentity(dispatched, replacement);
+  // The host-1 request keeps its epoch; the Main-process guard fails it closed.
+  assert.deepEqual(dispatched, { generation: 'generation-1', hostEpoch: 'host-1' });
+});

apps/desktop/src/main/app-ipc-main.ts

@@ -95,7 +95,7 @@ export function registerAppIpc(
   // observe the latest selection, not a snapshot taken at registration.
   const currentProjectRoot = (): Promise<string> => projectRoot.current();
 
-  targetIpc.handle('app:info', async () => {
+  handleReconnectableRead(targetIpc, 'app:info', async () => {
     const selection = await deps.projectManagement.current();
     const projectPath = allowLocalProjectPaths ? selection.path : '';
     return {

apps/desktop/src/main/runtime-host-desktop-candidate.ts

@@ -36,6 +36,7 @@ import {
   type RuntimeHostCandidateLaunchBarrier,
   type RuntimeHostSpawnedProcess,
   type RemoteRuntimeHostProfile,
+  type CandidateExitDetails,
 } from "@maka/runtime-host/client";
 import {
   INTERACTIVE_RUNTIME_HOST_COMPOSITION_ID,
@@ -167,6 +168,8 @@ export interface DesktopRuntimeHostCandidateStartInput
   readonly generation?: string;
   readonly takeoverHostEpoch?: string;
   readonly signal?: AbortSignal;
+  /** Candidate-exit sink forwarded to the launcher; the Desktop owns the sink. */
+  readonly onExit?: (details: CandidateExitDetails) => void;
   readonly candidateLaunchBarrier?: RuntimeHostCandidateLaunchBarrier;
   readonly remote?: {
     readonly profile: RemoteRuntimeHostProfile;
@@ -781,6 +784,7 @@ function connectInput(
       ? {}
       : { handshakeTimeoutMs: input.handshakeTimeoutMs }),
     ...(input.signal === undefined ? {} : { signal: input.signal }),
+    ...(input.onExit === undefined ? {} : { onExit: input.onExit }),
   };
 }
 

apps/desktop/src/main/runtime-host-desktop-manager.ts

@@ -27,6 +27,7 @@ import {
   LOCAL_RUNTIME_HOST_PROFILE,
   sameResolvedRuntimeHostProfileTarget,
   startRuntimeHostReconnectLifecycle,
+  type CandidateExitDetails,
   type ResolvedRuntimeHostProfile,
   type RuntimeHostReconnectBackoff,
   type RuntimeHostReconnectLifecycle,
@@ -665,6 +666,9 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
             signal,
             starting ? target.input.remote?.sshInteraction : 'batch',
           ),
+        onReconnectError: (error) => {
+          console.warn('[runtime-host] reconnect attempt failed:', error);
+        },
         onFatalError: (error) => {
           if (!starting && target.valid) {
             target.valid = false;
@@ -693,10 +697,12 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
     sshInteraction: RuntimeHostSshInteraction | undefined,
   ): Promise<DesktopRuntimeHostCandidate> {
     let takeoverHostEpoch: string | undefined;
+    const inheritedExit = target.input.onExit;
     while (true) {
       const result = await this.startCandidate(
         {
           ...target.input,
+          onExit: (details) => this.#reportCandidateExit(inheritedExit, details),
           ...(target.input.remote
             ? {
                 remote: {
@@ -776,6 +782,19 @@ class RuntimeHostDesktopManagerImpl implements RuntimeHostDesktopManager {
     return target.lifecycle;
   }
 
+  /** Desktop-owned candidate-exit diagnostics; honors an embedder-supplied sink. */
+  #reportCandidateExit(
+    inherited: ((details: CandidateExitDetails) => void) | undefined,
+    details: CandidateExitDetails,
+  ): void {
+    inherited?.(details);
+    if (details.code === 0 && details.signal === null) {
+      console.info('[runtime-host] candidate exited cleanly', details);
+      return;
+    }
+    console.error('[runtime-host] candidate exited unexpectedly', details);
+  }
+
   async #waitForReadyCandidate(
     lifecycle: RuntimeHostReconnectLifecycle<DesktopRuntimeHostCandidate>,
     previous?: DesktopRuntimeHostCandidate,

apps/desktop/src/main/runtime-host-session-execution-ipc-main.ts

@@ -780,7 +780,8 @@ function normalizeTranscriptRangeRequest(input: unknown): DesktopTranscriptRange
   }
   return {
     consumerId: requiredId(value.consumerId, 'Transcript consumer'),
-    generation: requiredId(value.generation, 'Transcript generation'),
+    sessionId: requiredId(value.sessionId, 'Session'),
+    hostEpoch: requiredId(value.hostEpoch, 'Host epoch'),
     anchorSequence: anchorSequence as number | null,
     maxBytes: maxBytes as number,
   };