packages/runtime-host/src/__tests__/session-subscription-client.test.ts

/*
 * 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 { deferred } from '@maka/core/test-only/async-primitives';
import assert from 'node:assert/strict';
import { createHash, randomUUID } from 'node:crypto';
import { mkdtemp, rm } from 'node:fs/promises';
import { createServer, type Server } from 'node:net';
import { tmpdir } from 'node:os';
import { join } from 'node:path';
import { test } from 'node:test';
import { setImmediate as delayImmediate, setTimeout as delay } from 'node:timers/promises';
import {
  prepareStorageRootControlDirectory,
  resolveStorageRoot,
} from '@maka/storage/root-authority';
import {
  decodeStoredMessage as decodePersistedStoredMessage,
  type StoredMessage,
} from '@maka/core/session';
import { markPersisted } from '@maka/core/persisted-value';
import {
  connectRuntimeHost,
  RuntimeHostSubscriptionError,
  SessionRemovedSubscriptionError,
  type RuntimeHostConnection,
} from '../client/index.js';
import { clientSubscription } from './fixtures/client-session-subscription.js';
import { prepareRuntimeHostEndpoint } from '../control/endpoint.js';
import { removeHostRegistration, writeHostRegistration } from '../control/registration.js';
import {
  decodeClientFrame,
  encodeProtocolMessage,
  RUNTIME_HOST_COMPATIBILITY_EPOCH,
  RUNTIME_HOST_PROTOCOL_VERSION,
  RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION,
  SESSION_CONTINUITY_SCHEMA_VERSION,
  SESSION_TRANSCRIPT_PAGE_MAX_BYTES,
  type SessionTranscriptBootstrap,
  type SessionTranscriptFragment,
  type SessionTranscriptPage,
  type HostFrame,
  type HostStatusResult,
  type RequestFrame,
  type SubscriptionFrame,
} from '../protocol/index.js';
import { FramedTransport } from '../transport/framed-transport.js';
import { frameLocalIpcProtocolMessage } from '../transport/local-ipc-framing.js';

const decodeStoredMessage = (value: unknown): StoredMessage =>
  decodePersistedStoredMessage(markPersisted<StoredMessage>(value));
const PROTOCOL = {
  min: RUNTIME_HOST_PROTOCOL_VERSION,
  max: RUNTIME_HOST_PROTOCOL_VERSION,
} as const;

test('registers a subscription before receiving a coalesced first frame', async () => {
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      const request = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
      const opened = openResult(hostEpoch, 'subscription-ordered');
      await writeRawLocalIpc(
        transport,
        Buffer.concat([
          encodeLocalIpcTestFrame({
            requestId: request.requestId,
            operation: 'subscription.open',
            ok: true,
            result: opened,
          }),
          encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, 1)),
        ]),
      );
      await answerClose(transport, opened.subscriptionId);
    },
    async (connection) => {
      const subscription = await connection.openSessionSubscription({
        sessionId: 'session-1',
        transcript: { kind: 'none' },
      });
      assert.deepEqual(await subscription[Symbol.asyncIterator]().next(), {
        done: false,
        value: deltaFrame(connection.hostEpoch, subscription.subscriptionId, 1),
      });
      await subscription.close();
    },
  );
});

test('unobserved PTY bytes do not consume the Session iterator or sequence', async () => {
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      const request = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
      const opened = openResult(hostEpoch, 'subscription-pty');
      const frame = {
        kind: 'subscription.runtime_resource_pty_data' as const,
        hostEpoch,
        subscriptionId: opened.subscriptionId,
        sessionId: 'session-1',
        ref: 'maka://runtime/background-tasks/shell-1',
        ptySequence: 7,
        data: 'ready',
      };
      await writeRawLocalIpc(
        transport,
        Buffer.concat([
          encodeLocalIpcTestFrame({
            requestId: request.requestId,
            operation: 'subscription.open',
            ok: true,
            result: opened,
          }),
          encodeLocalIpcTestFrame(frame),
          encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, 1)),
        ]),
      );
      await answerClose(transport, opened.subscriptionId);
    },
    async (connection) => {
      const subscription = await connection.openSessionSubscription({
        sessionId: 'session-1',
        transcript: { kind: 'none' },
      });
      assert.deepEqual(await subscription[Symbol.asyncIterator]().next(), {
        done: false,
        value: deltaFrame(connection.hostEpoch, subscription.subscriptionId, 1),
      });
      await subscription.close();
    },
  );
});

test('PTY callbacks bypass a stalled Session iterator and isolate consumer failures', async () => {
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-1'),
    async () => undefined,
    async () => {
      throw new Error('unexpected read');
    },
  );
  let delivered = 0;
  subscription.subscribePtyData(() => {
    throw new Error('broken display');
  });
  const unsubscribe = subscription.subscribePtyData(() => {
    delivered += 1;
  });
  for (let ptySequence = 1; ptySequence <= 1000; ptySequence += 1) {
    subscription.accept({
      kind: 'subscription.runtime_resource_pty_data',
      hostEpoch: 'host-1',
      subscriptionId: 'subscription-1',
      sessionId: 'session-1',
      ref: 'maka://runtime/background-tasks/shell-1',
      ptySequence,
      data: 'bytes',
    });
  }
  unsubscribe();
  assert.equal(delivered, 1000);
  subscription.accept(deltaFrame('host-1', 'subscription-1', 1));
  assert.deepEqual(await subscription.next(), {
    done: false,
    value: deltaFrame('host-1', 'subscription-1', 1),
  });
  await subscription.close();
});

test('domain callbacks validate identity, support unsubscribe, and stop on close', async () => {
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-domain'),
    async () => undefined,
    async () => {
      throw new Error('unexpected read');
    },
  );
  const domains: string[] = [];
  const unsubscribe = subscription.subscribeSessionDomainChanges((frame) => {
    domains.push(frame.domain);
  });
  subscription.accept({
    kind: 'subscription.session_domain_changed',
    hostEpoch: 'host-1',
    subscriptionId: 'subscription-domain',
    sequence: 1,
    sessionId: 'session-1',
    domain: 'todo',
  });
  assert.deepEqual(domains, ['todo']);

  unsubscribe();
  subscription.accept({
    kind: 'subscription.session_domain_changed',
    hostEpoch: 'host-1',
    subscriptionId: 'subscription-domain',
    sequence: 2,
    sessionId: 'session-1',
    domain: 'usage',
  });
  assert.deepEqual(domains, ['todo']);

  await subscription.close();
  assert.throws(
    () =>
      subscription.accept({
        kind: 'subscription.session_domain_changed',
        hostEpoch: 'host-1',
        subscriptionId: 'subscription-domain',
        sequence: 3,
        sessionId: 'other-session',
        domain: 'todo',
      }),
    /Session subscription is closed|Session subscription frame identity changed/,
  );
});

test('isolates a sequence gap and continues requests on the same connection', async () => {
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      const request = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
      const opened = openResult(hostEpoch, 'subscription-gap');
      await writeRawLocalIpc(
        transport,
        Buffer.concat([
          encodeLocalIpcTestFrame({
            requestId: request.requestId,
            operation: 'subscription.open',
            ok: true,
            result: opened,
          }),
          encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, 2)),
        ]),
      );
      await answerClose(transport, opened.subscriptionId);
      await answerStatus(transport, hostEpoch);
    },
    async (connection) => {
      const subscription = await connection.openSessionSubscription({
        sessionId: 'session-1',
        transcript: { kind: 'none' },
      });
      await assert.rejects(
        () => subscription[Symbol.asyncIterator]().next(),
        hasSubscriptionReason('sequence_gap'),
      );
      await assert.rejects(
        // @ts-expect-error host.status must use the validated status() API.
        () => connection.request('host.status', {}),
        /status requires the validated status\(\) API/,
      );
      assert.equal((await connection.status()).hostEpoch, connection.hostEpoch);
    },
  );
});

test('fails the connection when status reports a different Host identity', async () => {
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      const request = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
      const opened = openResult(hostEpoch, 'subscription-status-identity');
      await writeProtocolFrame(transport, {
        requestId: request.requestId,
        operation: 'subscription.open',
        ok: true,
        result: opened,
      });
      await answerClose(transport, opened.subscriptionId);
      await answerStatus(transport, 'different-host-epoch');
    },
    async (connection) => {
      const subscription = await connection.openSessionSubscription({
        sessionId: 'session-1',
        transcript: { kind: 'none' },
      });
      await subscription.close();
      await assert.rejects(() => connection.status(), /status for a different Host identity/);
      await connection.closed;
    },
  );
});

test('rejects epoch and Session correlation changes per subscription', async () => {
  for (const changed of ['epoch', 'session', 'graph'] as const) {
    await withProtocolPeer(
      async (transport, hostEpoch, rootId) => {
        const request = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
        const opened = openResult(hostEpoch, `subscription-${changed}`);
        await writeProtocolFrame(transport, {
          requestId: request.requestId,
          operation: 'subscription.open',
          ok: true,
          result: opened,
        });
        await writeProtocolFrame(
          transport,
          changed === 'graph'
            ? {
                kind: 'subscription.agent_graph_changed',
                hostEpoch,
                subscriptionId: opened.subscriptionId,
                sequence: 1,
                rootSessionId: 'session-2',
                graphId: 'agent_graph_1',
                reason: 'observation',
              }
            : {
                ...deltaFrame(
                  changed === 'epoch' ? 'different-epoch' : hostEpoch,
                  opened.subscriptionId,
                  1,
                ),
                ...(changed === 'session' ? { sessionId: 'session-2' } : {}),
              },
        );
        await answerClose(transport, opened.subscriptionId);
        await answerStatus(transport, hostEpoch);
      },
      async (connection) => {
        const subscription = await connection.openSessionSubscription({
          sessionId: 'session-1',
          transcript: { kind: 'none' },
        });
        await assert.rejects(
          () => subscription[Symbol.asyncIterator]().next(),
          hasSubscriptionReason(changed === 'epoch' ? 'host_epoch_changed' : 'correlation_changed'),
        );
        assert.equal((await connection.status()).hostEpoch, connection.hostEpoch);
      },
    );
  }
});

test('evicts a locally slow iterator and keeps the connection usable', async () => {
  const closeObserved = deferred<void>();
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      const request = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
      const opened = openResult(hostEpoch, 'subscription-slow');
      const frames = [
        encodeLocalIpcTestFrame({
          requestId: request.requestId,
          operation: 'subscription.open',
          ok: true,
          result: opened,
        }),
      ];
      for (let sequence = 1; sequence <= 33; sequence += 1) {
        frames.push(
          encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, sequence)),
        );
      }
      await writeRawLocalIpc(transport, Buffer.concat(frames));
      await answerClose(transport, opened.subscriptionId, closeObserved.resolve);
      await answerStatus(transport, hostEpoch);
    },
    async (connection) => {
      const subscription = await connection.openSessionSubscription({
        sessionId: 'session-1',
        transcript: { kind: 'none' },
      });
      await closeObserved.promise;
      await assert.rejects(
        () => subscription[Symbol.asyncIterator]().next(),
        hasSubscriptionReason('slow_consumer'),
      );
      assert.equal((await connection.status()).hostEpoch, connection.hostEpoch);
    },
  );
});

test('ends every active subscription with connection_closed on EOF', async () => {
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      const request = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
      await writeProtocolFrame(transport, {
        requestId: request.requestId,
        operation: 'subscription.open',
        ok: true,
        result: openResult(hostEpoch, 'subscription-eof'),
      });
      transport.closeAfterFlush();
    },
    async (connection) => {
      const subscription = await connection.openSessionSubscription({
        sessionId: 'session-1',
        transcript: { kind: 'none' },
      });
      await assert.rejects(
        () => subscription[Symbol.asyncIterator]().next(),
        hasSubscriptionReason('connection_closed'),
      );
    },
  );
});

test('records the close reason before a full queue can reject the frame', () => {
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-1'),
    async () => undefined,
    async () => {
      throw new Error('unexpected read');
    },
  );
  // Fill the client queue so the closed frame itself overflows it.
  for (let sequence = 1; sequence <= 32; sequence += 1) {
    subscription.accept(deltaFrame('host-1', 'subscription-1', sequence));
  }
  assert.throws(
    () =>
      subscription.accept({
        kind: 'subscription.closed',
        hostEpoch: 'host-1',
        subscriptionId: 'subscription-1',
        sequence: 33,
        reason: 'session_removed',
      }),
    hasSubscriptionReason('slow_consumer'),
  );
  assert.ok(subscription.deathCause instanceof SessionRemovedSubscriptionError);
});

test('a transcript read surfaces the terminal error, not the dead-state mask', () => {
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-1'),
    async () => undefined,
    async () => {
      throw new Error('unexpected read');
    },
  );
  const failure = new RuntimeHostSubscriptionError('sequence_gap', 'test gap');
  subscription.fail(failure);
  assert.equal(subscription.deathCause, failure);
  assert.throws(
    () =>
      subscription.loadTranscriptPage({
        direction: 'older',
        throughSequence: null,
        cursor: null,
        anchorSequence: null,
        maxBytes: 1024,
      }),
    (error: unknown) => error === failure,
  );
});

test('loads a canonical transcript while live frames continue on the same connection', async () => {
  const message = {
    type: 'assistant' as const,
    id: 'message-1',
    turnId: 'turn-1',
    ts: 1,
    text: 'snapshot text',
    modelId: 'test-model',
  };
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      const openRequest = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
      const opened = openResult(
        hostEpoch,
        'subscription-transcript',
        transcriptBootstrap(Buffer.from(JSON.stringify(message), 'utf8')),
      );
      await writeRawLocalIpc(
        transport,
        Buffer.concat([
          encodeLocalIpcTestFrame({
            requestId: openRequest.requestId,
            operation: 'subscription.open',
            ok: true,
            result: opened,
          }),
          encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, 1)),
        ]),
      );
      await answerClose(transport, opened.subscriptionId);
    },
    async (connection) => {
      const subscription = await connection.openSessionSubscription({
        sessionId: 'session-1',
        transcript: { kind: 'tail', maxBytes: 16 * 1024 },
      });
      assert.deepEqual(await subscription.loadTranscript(decodeStoredMessage), [message]);
      assert.deepEqual(await subscription[Symbol.asyncIterator]().next(), {
        done: false,
        value: deltaFrame(connection.hostEpoch, subscription.subscriptionId, 1),
      });
      await subscription.close();
    },
  );
});

test('resumes bounded index preparation before publishing the canonical transcript', async () => {
  const message = {
    type: 'assistant' as const,
    id: 'message-1',
    turnId: 'turn-1',
    ts: 1,
    text: 'snapshot text',
    modelId: 'test-model',
  };
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      let openRequest = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
      for (let batch = 0; batch < 3; batch++) {
        await writeProtocolFrame(transport, {
          requestId: openRequest.requestId,
          operation: 'subscription.open',
          ok: false,
          error: { code: 'transcript_preparing', message: `indexed through ${batch * 64}` },
        });
        const next = decodeClientFrame(await transport.read(1_000));
        assert.ok(!('kind' in next) && next.operation === 'subscription.open');
        assert.deepEqual(next.input, openRequest.input);
        openRequest = next;
      }
      const opened = openResult(
        hostEpoch,
        'subscription-transcript',
        transcriptBootstrap(Buffer.from(JSON.stringify(message), 'utf8')),
      );
      await writeRawLocalIpc(
        transport,
        Buffer.concat([
          encodeLocalIpcTestFrame({
            requestId: openRequest.requestId,
            operation: 'subscription.open',
            ok: true,
            result: opened,
          }),
          encodeLocalIpcTestFrame(deltaFrame(hostEpoch, opened.subscriptionId, 1)),
        ]),
      );
      await answerClose(transport, opened.subscriptionId);
    },
    async (connection) => {
      const subscription = await connection.openSessionSubscription({
        sessionId: 'session-1',
        transcript: { kind: 'tail', maxBytes: 16 * 1024 },
      });
      assert.deepEqual(await subscription.loadTranscript(decodeStoredMessage), [message]);
      assert.deepEqual(await subscription[Symbol.asyncIterator]().next(), {
        done: false,
        value: deltaFrame(connection.hostEpoch, subscription.subscriptionId, 1),
      });
      await subscription.close();
    },
  );
});

test('reassembles bounded backward pages with a timeout independent of index preparation', async () => {
  const message = {
    type: 'user' as const,
    id: 'user-1',
    turnId: 'turn-1',
    ts: 1,
    text: 'hello',
  };
  const encoded = Buffer.from(JSON.stringify(message), 'utf8');
  const splitAt = Math.floor(encoded.byteLength / 2);
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      let openRequest = await acceptConnectionAndReadOpen(transport, hostEpoch, rootId);
      await new Promise<void>((resolve) => setTimeout(resolve, 700));
      await writeProtocolFrame(transport, {
        requestId: openRequest.requestId,
        operation: 'subscription.open',
        ok: false,
        error: { code: 'transcript_preparing', message: 'Preparing history' },
      });
      const next = decodeClientFrame(await transport.read(1_000));
      assert.ok(!('kind' in next) && next.operation === 'subscription.open');
      openRequest = next;
      const opened = openResult(hostEpoch, 'subscription-fragmented', {
        durable: transcriptPage({
          rawBytes: encoded.byteLength - splitAt,
          fragments: [
            {
              sequence: 0,
              byteOffset: splitAt,
              totalBytes: encoded.byteLength,
              payloadDigest: null,
              data: encoded.subarray(splitAt).toString('base64'),
            },
          ],
          nextCursor: 'cursor-1',
        }),
      });
      await writeProtocolFrame(transport, {
        requestId: openRequest.requestId,
        operation: 'subscription.open',
        ok: true,
        result: opened,
      });
      const continuationRequest = decodeClientFrame(await transport.read(1_000));
      assert.ok(!('kind' in continuationRequest));
      assert.equal(continuationRequest.operation, 'session.transcript.page');
      assert.deepEqual(continuationRequest.input, {
        subscriptionId: opened.subscriptionId,
        direction: 'older',
        throughSequence: 0,
        cursor: 'cursor-1',
        anchorSequence: null,
        maxBytes: SESSION_TRANSCRIPT_PAGE_MAX_BYTES,
      });
      await new Promise<void>((resolve) => setTimeout(resolve, 500));
      await writeProtocolFrame(transport, {
        requestId: continuationRequest.requestId,
        operation: 'session.transcript.page',
        ok: true,
        result: transcriptPage({
          rawBytes: splitAt,
          fragments: [
            {
              sequence: 0,
              byteOffset: 0,
              totalBytes: encoded.byteLength,
              payloadDigest: null,
              data: encoded.subarray(0, splitAt).toString('base64'),
            },
          ],
        }),
      });
      await answerClose(transport, opened.subscriptionId);
    },
    async (connection) => {
      const subscription = await connection.openSessionSubscription(
        {
          sessionId: 'session-1',
          transcript: { kind: 'tail', maxBytes: 16 * 1024 },
        },
        1_000,
      );
      assert.deepEqual(await subscription.loadTranscript(decodeStoredMessage), [message]);
      await subscription.close();
    },
  );
});

test('decodes one bounded page without walking the remaining transcript', async () => {
  const message = {
    type: 'user' as const,
    id: 'user-1',
    turnId: 'turn-1',
    ts: 1,
    text: 'hello',
  };
  const encoded = Buffer.from(JSON.stringify(message), 'utf8');
  const splitAt = Math.floor(encoded.byteLength / 2);
  const requests: Array<{ cursor: string | null; maxBytes: number }> = [];
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-bounded-page', {
      durable: {
        ...transcriptPage({
          rawBytes: encoded.byteLength - splitAt,
          fragments: [
            {
              sequence: 4,
              byteOffset: splitAt,
              totalBytes: encoded.byteLength,
              payloadDigest: null,
              data: encoded.subarray(splitAt).toString('base64'),
            },
          ],
          nextCursor: 'complete-message',
        }),
        throughSequence: 4,
      },
    }),
    async () => undefined,
    async (input) => {
      requests.push({ cursor: input.cursor, maxBytes: input.maxBytes });
      return {
        ...transcriptPage({
          rawBytes: splitAt,
          fragments: [
            {
              sequence: 4,
              byteOffset: 0,
              totalBytes: encoded.byteLength,
              payloadDigest: null,
              data: encoded.subarray(0, splitAt).toString('base64'),
            },
          ],
          nextCursor: 'older-records',
        }),
        throughSequence: 4,
      };
    },
  );

  const assemblyDeltas: number[] = [];
  const decoded = await subscription.decodeTranscriptPage(
    subscription.transcriptBootstrap!.durable,
    decodeStoredMessage,
    undefined,
    (deltaBytes) => assemblyDeltas.push(deltaBytes),
  );

  assert.deepEqual(decoded, {
    messages: [{ identity: 4, message }],
    nextCursor: 'older-records',
  });
  assert.deepEqual(requests, [{ cursor: 'complete-message', maxBytes: splitAt }]);
  assert.deepEqual(assemblyDeltas, [encoded.byteLength, -encoded.byteLength]);

  requests.length = 0;
  await assert.rejects(
    subscription.decodeTranscriptPage(
      subscription.transcriptBootstrap!.durable,
      decodeStoredMessage,
      encoded.byteLength - 1,
    ),
    RangeError,
  );
  assert.deepEqual(requests, []);
});

test('returns a page of complete messages without reading past its cursor', async () => {
  const prompt = {
    type: 'user' as const,
    id: 'user-1',
    turnId: 'turn-1',
    ts: 1,
    text: 'prompt',
  };
  const promptBytes = Buffer.from(JSON.stringify(prompt), 'utf8');
  const requests: string[] = [];
  const initial: SessionTranscriptPage = {
    ...transcriptPage({
      rawBytes: promptBytes.byteLength,
      fragments: [
        {
          sequence: 0,
          byteOffset: 0,
          totalBytes: promptBytes.byteLength,
          payloadDigest: null,
          data: promptBytes.toString('base64'),
        },
      ],
      nextCursor: 'answer',
    }),
    direction: 'newer',
    throughSequence: 1,
  };
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-newer-turn', {
      durable: initial,
    }),
    async () => undefined,
    async (input) => {
      requests.push(input.cursor!);
      throw new Error('a complete page must not read its continuation');
    },
  );

  const decoded = await subscription.decodeTranscriptPage(initial, decodeStoredMessage);

  assert.deepEqual(
    decoded.messages.map(({ identity, message }) => [identity, message.id]),
    [[0, 'user-1']],
  );
  assert.equal(decoded.nextCursor, 'answer');
  assert.deepEqual(requests, []);
});

test('loads a durable transcript whose sequences are sparse', async () => {
  const messages = [0, 2].map((sequence) =>
    Buffer.from(
      JSON.stringify({
        type: 'user',
        id: `user-${sequence}`,
        turnId: 'turn-1',
        ts: sequence + 1,
        text: `visible-${sequence}`,
      }),
      'utf8',
    ),
  );
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-projected', {
      durable: {
        ...transcriptPage({
          rawBytes: messages.reduce((total, message) => total + message.byteLength, 0),
          fragments: messages
            .map((message, index) => ({
              sequence: index * 2,
              byteOffset: 0,
              totalBytes: message.byteLength,
              payloadDigest: null,
              data: message.toString('base64'),
            }))
            .reverse(),
        }),
        throughSequence: 2,
      },
    }),
    async () => undefined,
    async () => {
      throw new Error('unexpected page request');
    },
  );

  assert.deepEqual(
    (await subscription.loadTranscript(decodeStoredMessage)).map((message) => message.id),
    ['user-0', 'user-2'],
  );
});

test('rejects a durable message that does not match its payload digest', async () => {
  const message = Buffer.from(
    JSON.stringify({
      type: 'user',
      id: 'user-1',
      turnId: 'turn-1',
      ts: 1,
      text: 'hello',
    }),
    'utf8',
  );
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-digest-mismatch', {
      durable: transcriptPage({
        rawBytes: message.byteLength,
        fragments: [
          {
            sequence: 0,
            byteOffset: 0,
            totalBytes: message.byteLength,
            payloadDigest: `sha256:${createHash('sha256').update('different').digest('hex')}`,
            data: message.toString('base64'),
          },
        ],
      }),
    }),
    async () => undefined,
    async () => {
      throw new Error('unexpected page request');
    },
  );

  const assemblyDeltas: number[] = [];
  await assert.rejects(
    () =>
      subscription.decodeTranscriptPage(
        subscription.transcriptBootstrap!.durable,
        decodeStoredMessage,
        undefined,
        (deltaBytes) => assemblyDeltas.push(deltaBytes),
      ),
    hasSubscriptionReason('correlation_changed'),
  );
  assert.deepEqual(assemblyDeltas, [message.byteLength, -message.byteLength]);
});

test('rejects a transcript cursor that does not advance', async () => {
  const message = Buffer.from(
    JSON.stringify({
      type: 'user',
      id: 'user-1',
      turnId: 'turn-1',
      ts: 1,
      text: 'hello',
    }),
    'utf8',
  );
  const repeated = transcriptPage({
    rawBytes: message.byteLength,
    nextCursor: 'stuck-cursor',
    fragments: [
      {
        sequence: 0,
        byteOffset: 0,
        totalBytes: message.byteLength,
        payloadDigest: null,
        data: message.toString('base64'),
      },
    ],
  });
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-stuck-cursor', {
      durable: repeated,
    }),
    async () => undefined,
    async () => repeated,
  );

  await assert.rejects(subscription.loadTranscript(decodeStoredMessage), {
    name: 'RuntimeHostSubscriptionError',
    reason: 'correlation_changed',
    message: 'Session transcript cursor did not advance',
  });
});

test('close stops transcript pagination after the in-flight page', async () => {
  const message = Buffer.from(
    JSON.stringify({
      type: 'user',
      id: 'user-1',
      turnId: 'turn-1',
      ts: 1,
      text: 'hello',
    }),
    'utf8',
  );
  const page = deferred<ReturnType<typeof transcriptPage>>();
  let pageRequests = 0;
  const subscription = clientSubscription(
    openResult('host-1', 'subscription-closing', {
      durable: transcriptPage({
        rawBytes: Math.floor(message.byteLength / 2),
        fragments: [
          {
            sequence: 0,
            byteOffset: Math.ceil(message.byteLength / 2),
            totalBytes: message.byteLength,
            payloadDigest: null,
            data: message.subarray(Math.ceil(message.byteLength / 2)).toString('base64'),
          },
        ],
        nextCursor: 'cursor-1',
      }),
    }),
    async () => undefined,
    async () => {
      pageRequests += 1;
      return page.promise;
    },
  );
  const loading = subscription.loadTranscript(decodeStoredMessage);
  await delayImmediate();
  await subscription.close();
  page.resolve(transcriptPage());
  await assert.rejects(() => loading, hasSubscriptionReason('connection_closed'));
  assert.equal(pageRequests, 1);
});

test('probes an otherwise idle accepted Runtime Host connection', { timeout: 2_000 }, async () => {
  const probed = deferred<void>();
  const observed = deferred<HostStatusResult>();
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      const hello = decodeClientFrame(await transport.read(1_000));
      assert.ok('kind' in hello && hello.kind === 'hello');
      await writeProtocolFrame(transport, {
        kind: 'accepted',
        rootId,
        hostEpoch,
        connectionId: 'connection-idle-liveness',
        selectedProtocol: RUNTIME_HOST_PROTOCOL_VERSION,
        compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH,
        compositionId: 'maka.interactive',
        compositionRevision: '1',
        state: 'ready',
      });
      await answerStatus(transport, hostEpoch);
    },
    async () => {
      const status = await observed.promise;
      assert.equal(status.state, 'ready');
      assert.equal(status.compositionId, 'maka.interactive');
      await probed.promise;
    },
    {
      livenessIntervalMs: 20,
      onLivenessProbe: probed.resolve,
      onHostStatus: observed.resolve,
    },
  );
});

test('tolerates a short Host stall without abandoning the connection', {
  timeout: 5_000,
}, async () => {
  const observed = deferred<HostStatusResult>();
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      const hello = decodeClientFrame(await transport.read(1_000));
      assert.ok('kind' in hello && hello.kind === 'hello');
      await writeProtocolFrame(transport, {
        kind: 'accepted',
        rootId,
        hostEpoch,
        connectionId: 'connection-active-liveness',
        selectedProtocol: RUNTIME_HOST_PROTOCOL_VERSION,
        compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH,
        compositionId: 'maka.interactive',
        compositionRevision: '1',
        state: 'ready',
      });
      const probe = decodeClientFrame(await transport.read(1_000));
      assert.ok(!('kind' in probe));
      assert.equal(probe.operation, 'host.status');

      // A transient pause beyond the old two-second deadline is recoverable.
      // Only the eventual matching response completes the probe.
      await delay(1_100);
      await writeProtocolFrame(transport, {
        kind: 'session.catalog.changed',
        revision: 1,
        sessionId: 'shared-session',
      });
      await delay(1_100);
      await writeProtocolFrame(transport, {
        requestId: probe.requestId,
        operation: 'host.status',
        ok: true,
        result: hostStatus(hostEpoch),
      });
      await answerStatus(transport, hostEpoch);
    },
    async (connection) => {
      await observed.promise;
      assert.equal((await connection.status()).hostEpoch, connection.hostEpoch);
    },
    {
      livenessIntervalMs: 20,
      onHostStatus: observed.resolve,
    },
  );
});

test('closes an unresponsive request path even while Host notifications continue', {
  timeout: 12_000,
}, async (t) => {
  let received = 0;
  let probes = 0;
  const probeReceived = deferred<void>();
  const notificationsReceived = deferred<void>();
  const finalNotificationReceived = deferred<void>();
  let sendFinalNotification!: () => Promise<void>;
  await withProtocolPeer(
    async (transport, hostEpoch, rootId) => {
      await transport.read(1_000);
      await writeProtocolFrame(transport, {
        kind: 'accepted',
        rootId,
        hostEpoch,
        connectionId: 'one-way-host',
        selectedProtocol: RUNTIME_HOST_PROTOCOL_VERSION,
        compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH,
        compositionId: 'maka.interactive',
        compositionRevision: '1',
        state: 'ready',
      });
      let revision = 0;
      sendFinalNotification = () =>
        writeProtocolFrame(transport, {
          kind: 'session.catalog.changed',
          revision: ++revision,
          sessionId: 'final-notification',
        });
      const notifications = setInterval(() => {
        void writeProtocolFrame(transport, {
          kind: 'session.catalog.changed',
          revision: ++revision,
          sessionId: 'shared-session',
        }).catch(() => undefined);
      }, 10);
      try {
        const probe = decodeClientFrame(await transport.read(1_000));
        assert.ok(!('kind' in probe));
        assert.equal(probe.operation, 'host.status');
        probeReceived.resolve();
        await transport.closed;
      } finally {
        clearInterval(notifications);
      }
    },
    async (connection) => {
      let closed = false;
      void connection.closed.then(() => {
        closed = true;
      });
      connection.subscribeSessionCatalogChanges((event) => {
        received += 1;
        if (received > 10) notificationsReceived.resolve();
        if (event.sessionId === 'final-notification') finalNotificationReceived.resolve();
      });
      t.mock.timers.tick(20);
      await probeReceived.promise;
      await notificationsReceived.promise;
      t.mock.timers.tick(7_999);
      await sendFinalNotification().catch(() => undefined);
      await Promise.race([finalNotificationReceived.promise, connection.closed]);
      assert.equal(closed, false, 'inbound events must not end the pending probe early');
      t.mock.timers.tick(1);
      await connection.closed;
      assert.ok(received > 10, 'inbound events must remain active during the failed probe');
      assert.equal(probes, 0, 'one-way events cannot acknowledge a probe');
    },
    {
      livenessIntervalMs: 20,
      onLivenessProbe: () => {
        probes += 1;
      },
    },
    () => t.mock.timers.enable({ apis: ['setTimeout'] }),
  );
});

async function withProtocolPeer(
  serve: (transport: FramedTransport, hostEpoch: string, rootId: string) => Promise<void>,
  run: (connection: RuntimeHostConnection) => Promise<void>,
  connectionOptions: {
    readonly livenessIntervalMs?: number;
    readonly onLivenessProbe?: () => void;
    readonly onHostStatus?: (status: HostStatusResult) => void;
  } = {},
  beforeConnect?: () => void,
): Promise<void> {
  const base = await mkdtemp(join(tmpdir(), 'maka-runtime-host-subscription-'));
  const capability = await resolveStorageRoot({
    path: join(base, 'root'),
    kind: 'interactive',
  });
  const { controlDirectory } = await prepareStorageRootControlDirectory(capability);
  const hostEpoch = randomUUID();
  const endpoint = await prepareRuntimeHostEndpoint({
    rootId: capability.rootId,
    hostEpoch,
  });
  const serverTask = deferred<void>();
  const server = createServer((socket) => {
    void serve(new FramedTransport(socket), hostEpoch, capability.rootId).then(
      serverTask.resolve,
      serverTask.reject,
    );
  });
  try {
    await listen(server, endpoint.path);
    await endpoint.prepareAfterListen();
    await writeHostRegistration(controlDirectory, {
      kind: 'maka-runtime-host',
      schemaVersion: RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION,
      rootId: capability.rootId,
      hostEpoch,
      endpoint: endpoint.path,
      protocolMin: RUNTIME_HOST_PROTOCOL_VERSION,
      protocolMax: RUNTIME_HOST_PROTOCOL_VERSION,
      compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH,
      compositionId: 'maka.interactive',
      compositionRevision: '1',
      state: 'ready',
      pid: process.pid,
      createdAt: new Date().toISOString(),
    });
    beforeConnect?.();
    const connected = await connectRuntimeHost({
      rootPath: join(base, 'root'),
      protocol: PROTOCOL,
      ...connectionOptions,
    });
    assert.equal(connected.kind, 'connected');
    if (connected.kind !== 'connected') return;
    try {
      await run(connected.connection);
    } finally {
      await connected.connection.close();
    }
    await serverTask.promise;
  } finally {
    await closeServer(server);
    await removeHostRegistration(controlDirectory, hostEpoch).catch(() => undefined);
    await endpoint.cleanup().catch(() => undefined);
    await rm(base, { recursive: true, force: true });
  }
}

async function acceptConnectionAndReadOpen(
  transport: FramedTransport,
  hostEpoch: string,
  rootId: string,
): Promise<Extract<RequestFrame, { operation: 'subscription.open' }>> {
  const hello = decodeClientFrame(await transport.read(1_000));
  assert.ok('kind' in hello && hello.kind === 'hello');
  await writeProtocolFrame(transport, {
    kind: 'accepted',
    rootId,
    hostEpoch,
    connectionId: 'connection-1',
    selectedProtocol: RUNTIME_HOST_PROTOCOL_VERSION,
    compatibilityEpoch: RUNTIME_HOST_COMPATIBILITY_EPOCH,
    compositionId: 'maka.interactive',
    compositionRevision: '1',
    state: 'ready',
  });
  const request = decodeClientFrame(await transport.read(1_000));
  assert.ok(!('kind' in request));
  assert.equal(request.operation, 'subscription.open');
  return request as Extract<RequestFrame, { operation: 'subscription.open' }>;
}

async function answerClose(
  transport: FramedTransport,
  subscriptionId: string,
  onObserved?: () => void,
): Promise<void> {
  const request = decodeClientFrame(await transport.read(1_000));
  assert.ok(!('kind' in request));
  assert.equal(request.operation, 'subscription.close');
  assert.deepEqual(request.input, { subscriptionId });
  onObserved?.();
  await writeProtocolFrame(transport, {
    requestId: request.requestId,
    operation: 'subscription.close',
    ok: true,
    result: { subscriptionId },
  });
}

async function answerStatus(transport: FramedTransport, hostEpoch: string): Promise<void> {
  const request = decodeClientFrame(await transport.read(1_000));
  assert.ok(!('kind' in request));
  assert.equal(request.operation, 'host.status');
  await writeProtocolFrame(transport, {
    requestId: request.requestId,
    operation: 'host.status',
    ok: true,
    result: hostStatus(hostEpoch),
  });
}

function hostStatus(hostEpoch: string): HostStatusResult {
  return {
    hostEpoch,
    compositionId: 'maka.interactive',
    compositionRevision: '1',
    state: 'ready',
    connections: 1,
    activeOperations: 1,
    activeResidencies: 0,
  };
}

function openResult(
  hostEpoch: string,
  subscriptionId: string,
  transcript: SessionTranscriptBootstrap | null = null,
) {
  return {
    hostEpoch,
    subscriptionId,
    nextSequence: 1,
    activeAssistantStreams: [],
    transcript,
    snapshot: {
      schemaVersion: SESSION_CONTINUITY_SCHEMA_VERSION,
      session: {
        sessionId: 'session-1',
        metadataRevision: 1,
        status: 'running' as const,
        createdAt: 1,
        isArchived: false,
      },
      projectionRevision: 1,
      rootTurn: {
        sessionId: 'session-1',
        turnId: 'turn-1',
        runId: 'run-1',
        status: 'running' as const,
      },
      goal: null,
      queue: { hostEpoch, queueRevision: 1, steering: [], followup: [] },
      interactions: { pending: [] },
    },
  };
}

function transcriptBootstrap(message: Buffer): SessionTranscriptBootstrap {
  return {
    durable: transcriptPage({
      rawBytes: message.byteLength,
      fragments: [
        {
          sequence: 0,
          byteOffset: 0,
          totalBytes: message.byteLength,
          payloadDigest: null,
          data: message.toString('base64'),
        },
      ],
    }),
  };
}

function transcriptPage(
  options: {
    rawBytes?: number;
    fragments?: readonly SessionTranscriptFragment[];
    nextCursor?: string | null;
    endsAtTurnBoundary?: boolean;
  } = {},
): SessionTranscriptPage {
  return {
    kind: 'page',
    sessionId: 'session-1',
    direction: 'older',
    throughSequence: 0,
    rawBytes: options.rawBytes ?? 0,
    fragments: options.fragments ?? [],
    nextCursor: options.nextCursor ?? null,
    endsAtTurnBoundary: options.endsAtTurnBoundary ?? options.nextCursor == null,
  };
}

function deltaFrame(
  hostEpoch: string,
  subscriptionId: string,
  sequence: number,
): SubscriptionFrame {
  return {
    kind: 'subscription.session_delta',
    hostEpoch,
    subscriptionId,
    sequence,
    sessionId: 'session-1',
    delta: {
      kind: 'text',
      turnId: 'turn-1',
      runId: 'run-1',
      messageId: 'message-1',
      startOffset: 0,
      text: `chunk-${sequence}`,
    },
  };
}

function hasSubscriptionReason(reason: RuntimeHostSubscriptionError['reason']) {
  return (error: unknown) =>
    error instanceof RuntimeHostSubscriptionError && error.reason === reason;
}

function writeProtocolFrame(transport: FramedTransport, frame: HostFrame): Promise<void> {
  return transport.write(encodeProtocolMessage(frame));
}

function encodeLocalIpcTestFrame(frame: HostFrame): Buffer {
  return frameLocalIpcProtocolMessage(encodeProtocolMessage(frame));
}

function writeRawLocalIpc(transport: FramedTransport, frame: Uint8Array): Promise<void> {
  return new Promise((resolve, reject) => {
    transport.socket.write(frame, (error) => (error ? reject(error) : resolve()));
  });
}

function listen(server: Server, path: string): Promise<void> {
  return new Promise((resolve, reject) => {
    server.once('error', reject);
    server.listen(path, resolve);
  });
}

function closeServer(server: Server): Promise<void> {
  if (!server.listening) return Promise.resolve();
  return new Promise((resolve, reject) => {
    server.close((error) => (error ? reject(error) : resolve()));
  });
}