packages/runtime-host/src/__tests__/execution-host.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 { withTimeout } from '@maka/core/test-only/async-primitives';
import assert from 'node:assert/strict';
import { fork, type ChildProcess } from 'node:child_process';
import { randomUUID } from 'node:crypto';
import {
  appendFile,
  chmod,
  mkdir,
  mkdtemp,
  readFile,
  readdir,
  rm,
  writeFile,
} from 'node:fs/promises';
import { createServer, type Server } from 'node:http';
import { connect, type Socket } from 'node:net';
import { tmpdir } from 'node:os';
import { dirname, join } from 'node:path';
import { DatabaseSync } from 'node:sqlite';
import { test } from 'node:test';
import { TOOL_BOUNDARY_PROTOCOL_V1 } from '@maka/core/runtime-event';
import { canonicalToolArgsHash } from '@maka/core/tool-args-identity';
import type { MessageContent } from '@maka/core/events';
import type { ConnectionCatalogEntry } from '@maka/core/runtime-policy';
import {
  decodeStoredMessage as decodePersistedStoredMessage,
  type StoredMessage,
} from '@maka/core/session';
import { markPersisted } from '@maka/core/persisted-value';
import type { SessionTodoItem } from '@maka/core/session-todo';
import type { ScheduledTask } from '@maka/core/scheduled-task';
import { isTerminalRuntimeEvent } from '@maka/core/runtime-event';
import type { RuntimeEvent } from '@maka/core/runtime-event';
import { buildSessionTodoTools } from '@maka/runtime/session-todo-tools';
import {
  buildRecoveredTerminalRuntimeEvent,
  classifyTerminalRuntimeLedger,
  commitTerminalRunWithRuntimeFact,
} from '@maka/runtime/terminal-run-commit';
import {
  FAKE_ASK_SANDBOX_BOUNDARY_PROMPT,
  FAKE_ASK_USER_QUESTION_PROMPT,
  FAKE_WAIT_FOR_STEERING_PROMPT,
} from '@maka/runtime/test-only/fake-backend';
import { type MakaTool, type MakaToolContext } from '@maka/runtime/tool-runtime';
import {
  openInteractiveExecutionStoresForRead,
  openInteractiveExecutionStoresForWrite,
} from '@maka/storage/execution-stores';
import { OPERATIONAL_STATE_DATABASE_NAME } from '@maka/storage/operational-state-store';
import { openInteractiveRuntimePolicyStoresForWrite } from '@maka/storage/runtime-policy-stores';
import {
  resolveRootControlNamespace,
  resolveStorageRoot,
  tryAcquireInteractiveRootOwner,
  tryAcquireInteractiveRootReader,
  type StorageRootCapability,
} from '@maka/storage/root-authority';
import { openInteractiveSessionTodoStoreForWrite } from '@maka/storage/session-todo-authority';
import {
  connectRuntimeHost,
  RuntimeHostOperationError,
  RuntimeHostSubscriptionError,
  type RuntimeHostConnection,
  type RuntimeHostSessionSubscription,
} from '../client/index.js';
import {
  decodeHostFrame,
  RUNTIME_HOST_PROTOCOL_VERSION,
  type ConnectionCatalogQueryResult,
  type InteractionPendingSnapshot,
  type SubscriptionFrame,
  type TurnMessageSubmitInput,
  type TurnSnapshot,
} from '../protocol/index.js';
import { SessionAdmissionGate } from '../server/session-admission-gate.js';
import { HostSessionTodoCoordinator } from '../server/session-todo-coordinator.js';
import { FramedTransport } from '../transport/framed-transport.js';

import {
  CONNECTION_EFFECT_MODEL_IDS,
  type ExecutionFixture,
  PROCESS_TIMEOUT_MS,
  SubscriptionProbe,
  assertJsonLines,
  attachment,
  connectClient,
  requireStartedTurn,
  operationError,
  quotedContent,
  sendStartWithoutReadingResponse,
  startConnectionEffectProvider,
  userRuntimeContent,
  waitForDurableMessageConflict,
  waitForPendingInteraction,
  waitForRunningTurn,
  waitForTerminalTurn,
  waitForTurn,
  withExecutionRoot,
} from './fixtures/execution-host-suite.js';

const decodeStoredMessage = (value: unknown): StoredMessage =>
  decodePersistedStoredMessage(markPersisted<StoredMessage>(value));

test('production Host resumes a Session through the ScheduledTask authority', {
  timeout: 30_000,
}, async () => {
  await withExecutionRoot(async (fixture) => {
    const host = await fixture.startHost();
    const desktop = await connectClient(fixture.root);
    const tui = await connectClient(fixture.root);
    try {
      const heartbeat = await desktop.request('scheduled-task.mutate', {
        kind: 'create',
        input: {
          title: 'session resume execution proof',
          intentBody: 'Complete the scheduled execution proof.',
          schedule: { kind: 'once', runAt: Date.now() + 5_000 },
          effect: { kind: 'session_resume', sessionId: fixture.sessionId },
        },
      });
      assert.equal(heartbeat.kind, 'task');
      if (heartbeat.kind !== 'task') {
        return;
      }

      const observedHeartbeat = await waitForScheduledTaskCompletion(tui, heartbeat.task.id);
      assert.ok(observedHeartbeat.runs[0]?.runId);
      assert.equal(observedHeartbeat.lastError, null);

      const deletedHeartbeat = await tui.request('scheduled-task.mutate', {
        kind: 'delete',
        taskId: heartbeat.task.id,
      });
      assert.equal(deletedHeartbeat.kind, 'deleted');
    } finally {
      await Promise.allSettled([desktop.close(), tui.close()]);
      await fixture.stopHost(host);
    }
  });
});

test('production Host fails slug-only ScheduledTask Agent runs before binding execution identity', {
  timeout: 30_000,
}, async () => {
  await withExecutionRoot(async (fixture) => {
    const seededConnection = await fixture.seedConnectionEffect(
      'http://127.0.0.1:1',
      'test-secret',
    );
    const host = await fixture.startHost();
    const desktop = await connectClient(fixture.root);
    try {
      const created = await desktop.request('scheduled-task.mutate', {
        kind: 'create',
        input: {
          title: 'legacy agent-run identity proof',
          intentBody: 'Do not execute with a replacement account.',
          schedule: { kind: 'once', runAt: Date.now() + 60_000 },
          effect: {
            kind: 'agent_run',
            execution: {
              cwd: fixture.root,
              llmConnectionId: seededConnection.connectionId,
              llmConnectionSlug: seededConnection.slug,
              model: seededConnection.enabledModelIds[0]!,
              permissionMode: 'ask',
              collaborationMode: 'agent',
              orchestrationMode: 'default',
            },
          },
        },
      });
      assert.equal(created.kind, 'task');
      if (created.kind !== 'task') return;

      // Simulate a record written by a pre-#3927 build. Legacy rows remain
      // readable, but must fail closed before a Session or AgentRun is bound.
      const database = new DatabaseSync(join(fixture.root, OPERATIONAL_STATE_DATABASE_NAME));
      try {
        database
          .prepare(
            `UPDATE workflow_scheduled_tasks
             SET record_json = json_remove(record_json, '$.effect.execution.llmConnectionId')
             WHERE task_id = ?`,
          )
          .run(created.task.id);
      } finally {
        database.close();
      }

      const fired = await desktop.request('scheduled-task.mutate', {
        kind: 'trigger_now',
        taskId: created.task.id,
      });
      assert.equal(fired.kind, 'task');
      if (fired.kind !== 'task') return;
      assert.equal(
        fired.task.lastError,
        'ScheduledTask Agent runs require an immutable model connection identity',
      );
      assert.equal(fired.task.runs.length, 1);
      assert.equal(fired.task.runs[0]?.outcome, 'failed');
      assert.equal(fired.task.runs[0]?.sessionId, undefined);
      assert.equal(fired.task.runs[0]?.runId, undefined);
    } finally {
      await desktop.close();
      await fixture.stopHost(host);
    }
  });
});

test('two UDS Clients never rebind an Agent ScheduledTask after Connection slug reuse', {
  timeout: 30_000,
}, async () => {
  await withExecutionRoot(async (fixture) => {
    const original = await fixture.seedConnectionEffect('http://127.0.0.1:1', 'test-secret');
    const model = original.enabledModelIds[0]!;
    const host = await fixture.startHost();
    const creator = await connectClient(fixture.root);
    const trigger = await connectClient(fixture.root);
    try {
      const created = await creator.request('scheduled-task.mutate', {
        kind: 'create',
        input: {
          title: 'Connection slug reuse proof',
          intentBody: 'The deleted account must never be replaced silently.',
          schedule: { kind: 'once', runAt: Date.now() + 60_000 },
          effect: {
            kind: 'agent_run',
            execution: {
              cwd: fixture.root,
              llmConnectionId: original.connectionId,
              llmConnectionSlug: original.slug,
              model,
              permissionMode: 'ask',
              collaborationMode: 'agent',
              orchestrationMode: 'default',
            },
          },
        },
      });
      assert.equal(created.kind, 'task');
      if (created.kind !== 'task') return;

      const catalog = await trigger.request('connection.catalog.query', { kind: 'start' });
      assert.equal(catalog.kind, 'page');
      if (catalog.kind !== 'page') return;
      const header = catalog.items.find(
        (item) => item.kind === 'connection' && item.connectionId === original.connectionId,
      );
      assert.equal(header?.kind, 'connection');
      if (header?.kind !== 'connection') return;

      const removed = await trigger.request('connection.catalog.remove', {
        expected: { connectionId: original.connectionId, revision: header.revision },
      });
      assert.equal(removed.kind, 'committed');
      if (removed.kind !== 'committed') return;
      const replacement = await trigger.request('connection.catalog.create', {
        expectedCatalogRevision: removed.catalogRevision,
        connection: {
          slug: original.slug,
          name: 'Replacement account',
          providerType: original.providerType,
          ...(original.baseUrl === undefined ? {} : { baseUrl: original.baseUrl }),
          enabled: true,
          enabledModelIds: [model],
        },
      });
      assert.equal(replacement.kind, 'committed');
      if (replacement.kind !== 'committed') return;
      assert.notEqual(replacement.connection.connectionId, original.connectionId);

      const fired = await trigger.request('scheduled-task.mutate', {
        kind: 'trigger_now',
        taskId: created.task.id,
      });
      assert.equal(fired.kind, 'task');
      if (fired.kind !== 'task') return;
      assert.equal(fired.task.runs[0]?.outcome, 'failed');
      assert.equal(fired.task.lastError, 'ScheduledTask model connection identity changed');
      const failedRun = fired.task.runs[0];
      assert.ok(failedRun?.sessionId);
      assert.ok(failedRun?.runId);
      const databaseAfterFire = new DatabaseSync(
        join(fixture.root, OPERATIONAL_STATE_DATABASE_NAME),
      );
      try {
        assert.equal(
          databaseAfterFire
            .prepare('SELECT 1 AS present FROM session_metadata WHERE session_id = ?')
            .get(failedRun.sessionId),
          undefined,
        );
        assert.equal(
          databaseAfterFire
            .prepare('SELECT 1 AS present FROM core_agent_runs WHERE run_id = ?')
            .get(failedRun.runId),
          undefined,
        );
      } finally {
        databaseAfterFire.close();
      }
    } finally {
      await Promise.allSettled([creator.close(), trigger.close()]);
      await fixture.stopHost(host);
    }
  });
});

test('production Host settles dispatched Client Capabilities before publishing Ready', async () => {
  await withExecutionRoot(async (fixture) => {
    const prepared = await seedDispatchedClientCapability(fixture);
    const host = await fixture.startHost({
      sessionId: fixture.sessionId,
      runId: prepared.runId,
    });
    try {
      const outcome = host.recoveryOutcome;
      assert.equal(outcome?.content?.kind, 'function_response');
      if (outcome?.content?.kind !== 'function_response') return;
      assert.equal(outcome.content.name, prepared.toolName);
      assert.equal(outcome.content.isError, true);
      assert.ok(outcome.content.result && typeof outcome.content.result === 'object');
      const recovered = outcome.content.result as {
        kind?: unknown;
        uncertainOutcome?: unknown;
      };
      assert.equal(recovered.kind, 'text');
      assert.deepEqual(recovered.uncertainOutcome, {
        code: 'outcome_unknown',
        retrySafe: false,
      });
    } finally {
      await fixture.stopHost(host);
    }
  });
});

test('dual UDS Clients query the same persisted SessionTodo snapshot across Host restart', async () => {
  await withExecutionRoot(async (fixture) => {
    const initial: SessionTodoItem[] = Array.from({ length: 129 }, (_, index) => ({
      content: `Authority acceptance todo ${index + 1}`,
      status: index === 0 ? 'in_progress' : 'pending',
    }));
    await withOwnedSessionTodoToolPort(fixture, async (_coordinator, tools) => {
      const write = requireSessionTodoWriteTool(tools);
      await write.impl(write.parameters.parse({ todos: initial }), sessionTodoToolContext(fixture));
    });

    const host = await fixture.startHost();
    const desktop = await connectClient(fixture.root);
    const tui = await connectClient(fixture.root);
    try {
      const [desktopProjection, tuiProjection] = await Promise.all([
        desktop.request('session.todo.query', { sessionId: fixture.sessionId }),
        tui.request('session.todo.query', { sessionId: fixture.sessionId }),
      ]);
      assert.deepEqual(desktopProjection, { sessionId: fixture.sessionId, items: initial });
      assert.deepEqual(tuiProjection, desktopProjection);
    } finally {
      await Promise.allSettled([desktop.close(), tui.close()]);
      await fixture.stopHost(host);
    }

    const changed = [
      { content: 'Changed after authority reacquisition', status: 'completed' },
    ] as const;
    await withOwnedSessionTodoToolPort(fixture, async (_coordinator, tools) => {
      const write = requireSessionTodoWriteTool(tools);
      await write.impl(write.parameters.parse({ todos: changed }), sessionTodoToolContext(fixture));
    });

    const successorHost = await fixture.startHost();
    const successor = await connectClient(fixture.root);
    try {
      assert.deepEqual(
        await successor.request('session.todo.query', { sessionId: fixture.sessionId }),
        {
          sessionId: fixture.sessionId,
          items: changed,
        },
      );
    } finally {
      await successor.close();
      await fixture.stopHost(successorHost);
    }
  });
});

async function seedDispatchedClientCapability(
  fixture: ExecutionFixture,
): Promise<{ runId: string; toolName: string }> {
  const owner = await tryAcquireInteractiveRootOwner(fixture.capability);
  assert.ok(owner);
  if (!owner) throw new Error('Unable to acquire execution root for Client Capability setup');
  let stores: Awaited<ReturnType<typeof openInteractiveExecutionStoresForWrite>> | undefined;
  try {
    stores = await openInteractiveExecutionStoresForWrite(owner.lease);
    const operationId = 'client-capability-before-ready';
    const invocationId = `${operationId}-invocation`;
    const runId = `${operationId}-run`;
    const turnId = `${operationId}-turn`;
    const providerToolCallId = `${operationId}-call`;
    const toolName = 'mcp__client_fixture__navigate';
    const args = { url: 'https://example.test/recovery' };
    const canonicalArgsHash = canonicalToolArgsHash(toolName, args);
    const call: RuntimeEvent = {
      id: `${operationId}_call`,
      invocationId,
      runId,
      sessionId: fixture.sessionId,
      turnId,
      ts: 10,
      partial: false,
      role: 'model',
      author: 'agent',
      content: {
        kind: 'function_call',
        id: providerToolCallId,
        name: toolName,
        args,
      },
      refs: { operationId, toolCallId: providerToolCallId },
    };
    const dispatch: RuntimeEvent = {
      id: `${operationId}_dispatch`,
      invocationId,
      runId,
      sessionId: fixture.sessionId,
      turnId,
      ts: 10,
      partial: false,
      role: 'system',
      author: 'system',
      actions: {
        toolDispatch: {
          protocol: TOOL_BOUNDARY_PROTOCOL_V1,
          operationId,
          providerToolCallId,
          toolName,
          canonicalArgsHash,
          recoveryMode: 'outcome_unknown',
        },
      },
      refs: { operationId, toolCallId: providerToolCallId },
    };
    await stores.runtimeEventStore.commitToolPrepared({
      operationId,
      journalEventId: `${operationId}_prepared`,
      runtimeEvent: call,
      dispatchRuntimeEvent: dispatch,
      providerToolCallId,
      toolName,
      canonicalArgsHash,
      recoveryMode: 'outcome_unknown',
      committedAt: 10,
    });
    return { runId, toolName };
  } finally {
    await stores?.sessionStore.close?.();
    await owner.close();
  }
}

test('two UDS Clients share one Runtime Policy authority and CAS winner', async () => {
  await withExecutionRoot(async (fixture) => {
    const host = await fixture.startHost();
    const first = await connectClient(fixture.root);
    const second = await connectClient(fixture.root);
    try {
      const initial = await first.request('runtime.policy.query', {});
      assert.deepEqual(await second.request('runtime.policy.query', {}), initial);
      const outcomes = await Promise.all([
        first.request('runtime.policy.mutate', {
          expectedRevision: initial.revision,
          operation: {
            kind: 'set_personalization',
            value: { displayName: 'Desktop', assistantTone: 'precise' },
          },
        }),
        second.request('runtime.policy.mutate', {
          expectedRevision: initial.revision,
          operation: {
            kind: 'set_memory',
            value: { enabled: false, agentReadEnabled: false },
          },
        }),
      ]);
      assert.deepEqual(outcomes.map((outcome) => outcome.kind).sort(), [
        'committed',
        'revision_conflict',
      ]);
      assert.deepEqual(
        await first.request('runtime.policy.query', {}),
        await second.request('runtime.policy.query', {}),
      );
    } finally {
      await Promise.allSettled([first.close(), second.close()]);
      await fixture.stopHost(host);
    }
  });
});

test('two UDS Clients serialize same-provider account creation through one Host lane', async () => {
  const provider = await startConnectionEffectProvider({ responseDelayMs: 50 });
  try {
    await withExecutionRoot(async (fixture) => {
      const host = await fixture.startHost();
      const desktop = await connectClient(fixture.root);
      const tui = await connectClient(fixture.root);
      const secrets = ['desktop-account-secret', 'tui-account-secret'] as const;
      let identities: Array<{ connectionId: string; slug: string }> = [];
      try {
        const results = await Promise.all(
          [desktop, tui].map((client, index) =>
            client.request('connection.onboarding.save', {
              target: { kind: 'create', providerType: 'openai-compatible' },
              apiKey: secrets[index]!,
              baseUrl: provider.baseUrl,
              enabledModelIds: [CONNECTION_EFFECT_MODEL_IDS[0]!],
            }),
          ),
        );
        assert.ok(results.every((result) => result.kind === 'saved'));
        identities = results.map((result) => {
          if (result.kind !== 'saved') throw new Error('Onboarding did not save');
          return {
            connectionId: result.connection.connectionId,
            slug: result.connection.slug,
          };
        });
        assert.notEqual(identities[0]?.connectionId, identities[1]?.connectionId);
        assert.deepEqual(identities.map(({ slug }) => slug).sort(), [
          'openai-compatible',
          'openai-compatible-2',
        ]);
      } finally {
        await Promise.allSettled([desktop.close(), tui.close()]);
        await fixture.stopHost(host);
      }

      const owner = await tryAcquireInteractiveRootOwner(fixture.capability);
      assert.ok(owner);
      if (!owner) return;
      try {
        const stores = await openInteractiveRuntimePolicyStoresForWrite(owner.lease);
        const catalog = await stores.connectionCatalog.getSnapshot();
        assert.deepEqual(
          catalog.connections
            .filter(({ providerType }) => providerType === 'openai-compatible')
            .map(({ connectionId, slug }) => ({ connectionId, slug }))
            .sort((left, right) => left.slug.localeCompare(right.slug)),
          [...identities].sort((left, right) => left.slug.localeCompare(right.slug)),
        );
        for (const [index, identity] of identities.entries()) {
          assert.equal(
            (
              await stores.operations.exportCredentialMaterial({
                scope: 'connection',
                connectionId: identity.connectionId,
                kind: 'api_key',
              })
            )?.secret,
            secrets[index],
          );
        }
      } finally {
        await owner.close();
      }
      assert.deepEqual(
        provider.requests.map(({ authorization }) => authorization).sort(),
        secrets.map((secret) => `Bearer ${secret}`).sort(),
      );
    });
  } finally {
    await provider.close();
  }
});

test('two UDS Clients await slow connection effects against one canonical catalog', async () => {
  const provider = await startConnectionEffectProvider({ responseDelayMs: 2_100 });
  try {
    await withExecutionRoot(async (fixture) => {
      const secret = 'connection-effect-secret';
      const connection = await fixture.seedConnectionEffect(provider.baseUrl, secret);
      const host = await fixture.startHost();
      const desktop = await connectClient(fixture.root);
      const tui = await connectClient(fixture.root);
      try {
        assert.equal(desktop.hostEpoch, tui.hostEpoch);
        assert.notEqual(desktop.connectionId, tui.connectionId);
        const fetchInput = { connectionId: connection.connectionId };
        const fetched = await desktop.request('connection.models.fetch', {
          ...fetchInput,
        });
        assert.equal(fetched.kind, 'committed');
        if (fetched.kind !== 'committed') return;
        assert.equal(fetched.modelCount, CONNECTION_EFFECT_MODEL_IDS.length);
        assert.equal(fetched.source, 'fetched');

        const firstPage = await tui.request('connection.catalog.query', { kind: 'start' });
        assert.equal(firstPage.kind, 'page');
        if (firstPage.kind !== 'page') return;
        type CatalogPage = Extract<ConnectionCatalogQueryResult, { readonly kind: 'page' }>;
        const pages: CatalogPage[] = [firstPage];
        let observed: CatalogPage = firstPage;
        while (observed.nextCursor) {
          const nextResult: ConnectionCatalogQueryResult = await tui.request(
            'connection.catalog.query',
            {
              kind: 'continue',
              revision: observed.revision,
              cursor: observed.nextCursor,
            },
          );
          assert.equal(nextResult.kind, 'page');
          if (nextResult.kind !== 'page') return;
          pages.push(nextResult);
          observed = nextResult;
        }
        assert.ok(pages.length > 1);
        assert.ok(pages.every((page) => page.revision === fetched.catalogRevision));
        assert.deepEqual(
          pages.flatMap((page) =>
            page.items.flatMap((item) => (item.kind === 'model' ? [item.model.id] : [])),
          ),
          CONNECTION_EFFECT_MODEL_IDS,
        );

        const testInput = {
          connectionId: connection.connectionId,
          modelId: CONNECTION_EFFECT_MODEL_IDS[0]!,
        };
        const tested = await tui.request('connection.test.run', {
          ...testInput,
        });
        assert.equal(tested.kind, 'committed');
        if (tested.kind !== 'committed') return;
        assert.equal(tested.test.kind, 'verified');

        const canonical = await desktop.request('connection.catalog.query', { kind: 'start' });
        assert.equal(canonical.kind, 'page');
        if (canonical.kind !== 'page') return;
        const header = canonical.items.find(
          (item) => item.kind === 'connection' && item.connectionId === connection.connectionId,
        );
        assert.equal(header?.kind, 'connection');
        if (header?.kind === 'connection') {
          assert.deepEqual(header.lastTest, {
            status: 'verified',
            checkedAt: tested.test.checkedAt,
          });
        }
        assert.equal(
          JSON.stringify([fetchInput, fetched, pages, testInput, tested, canonical]).includes(
            secret,
          ),
          false,
        );
        assert.equal(provider.requests.length, 2);
        assert.ok(
          provider.requests.every(({ authorization }) => authorization === `Bearer ${secret}`),
        );
        assert.deepEqual(
          provider.requests.map(({ method, url }) => ({
            method,
            url,
          })),
          [
            {
              method: 'GET',
              url: '/v1/models',
            },
            {
              method: 'POST',
              url: '/v1/chat/completions',
            },
          ],
        );
      } finally {
        await Promise.allSettled([desktop.close(), tui.close()]);
        await fixture.stopHost(host);
      }
    });
  } finally {
    await provider.close();
  }
});

test('two Clients share one execution after the starting Client disconnects', async () => {
  await withExecutionRoot(async (fixture) => {
    const host = await fixture.startHost();
    const first = await connectClient(fixture.root);
    const second = await connectClient(fixture.root);
    const turnId = randomUUID();

    const started = requireStartedTurn(
      await first.request(
        'turn.start',
        {
          sessionId: fixture.sessionId,
          turnId,
          content: { text: FAKE_ASK_USER_QUESTION_PROMPT },
        },
        PROCESS_TIMEOUT_MS,
      ),
    );
    assert.equal(started.turnId, turnId);
    const secondSubscription = await second.openSessionSubscription({
      sessionId: fixture.sessionId,
      transcript: { kind: 'tail', maxBytes: 16 * 1024 },
    });
    const transcript = await secondSubscription.loadTranscript(decodeStoredMessage);
    assert.ok(
      transcript.some(
        (message) =>
          message.type === 'user' &&
          message.turnId === turnId &&
          message.text === FAKE_ASK_USER_QUESTION_PROMPT,
      ),
    );
    const secondProbe = new SubscriptionProbe(secondSubscription);
    await assert.rejects(
      () =>
        second.request(
          'turn.start',
          {
            sessionId: fixture.sessionId,
            turnId: randomUUID(),
            content: { text: 'must stay busy' },
          },
          PROCESS_TIMEOUT_MS,
        ),
      operationError('session_busy'),
    );

    await first.close();
    const pending = await waitForPendingInteraction(secondSubscription, secondProbe, started.runId);
    assert.equal(pending.sessionId, fixture.sessionId);
    assert.equal(pending.turnId, turnId);
    assert.equal(pending.runId, started.runId);
    const questionRequest = pending.request;
    assert.ok(questionRequest.kind === 'question');
    assert.deepEqual(
      await second.request('interaction.query', {
        sessionId: fixture.sessionId,
        interactionId: pending.interactionId,
      }),
      pending,
    );
    const observed = await second.request('turn.query', {
      sessionId: fixture.sessionId,
      turnId,
    });
    assert.equal(observed.runId, started.runId);
    assert.ok(observed.status === 'running' || observed.status === 'waiting_for_user');
    const stopped = await second.request(
      'turn.stop',
      {
        sessionId: fixture.sessionId,
        turnId,
        runId: started.runId,
      },
      PROCESS_TIMEOUT_MS,
    );
    assert.equal(stopped.status, 'cancelled');
    const closed = await second.request('interaction.query', {
      sessionId: fixture.sessionId,
      interactionId: pending.interactionId,
    });
    assert.equal(closed.sessionId, fixture.sessionId);
    assert.equal(closed.turnId, turnId);
    assert.equal(closed.runId, started.runId);
    assert.equal(closed.status, 'closed');
    assert.equal(closed.outcome.kind, 'closure');
    if (closed.outcome.kind === 'closure') assert.equal(closed.outcome.reason, 'turn_stopped');
    await assert.rejects(
      () =>
        second.request('interaction.answer', {
          sessionId: fixture.sessionId,
          interactionId: pending.interactionId,
          answer: {
            kind: 'question',
            answers: questionRequest.questions.map(() => null),
          },
        }),
      operationError('already_resolved'),
    );

    const nextTurnId = randomUUID();
    const next = requireStartedTurn(
      await second.request(
        'turn.start',
        {
          sessionId: fixture.sessionId,
          turnId: nextTurnId,
          content: { text: FAKE_ASK_USER_QUESTION_PROMPT },
        },
        PROCESS_TIMEOUT_MS,
      ),
    );
    assert.deepEqual(
      requireStartedTurn(
        await second.request(
          'turn.start',
          {
            sessionId: fixture.sessionId,
            turnId,
            content: { text: FAKE_ASK_USER_QUESTION_PROMPT },
          },
          PROCESS_TIMEOUT_MS,
        ),
      ),
      stopped,
    );
    assert.deepEqual(
      await second.request('turn.stop', {
        sessionId: fixture.sessionId,
        turnId,
        runId: started.runId,
      }),
      stopped,
    );
    const nextObserved = await second.request('turn.query', {
      sessionId: fixture.sessionId,
      turnId: nextTurnId,
    });
    assert.equal(nextObserved.runId, next.runId);
    assert.ok(nextObserved.status === 'running' || nextObserved.status === 'waiting_for_user');
    await second.request(
      'turn.stop',
      {
        sessionId: fixture.sessionId,
        turnId: nextTurnId,
        runId: next.runId,
      },
      PROCESS_TIMEOUT_MS,
    );
    await secondSubscription.close();
    await secondProbe.done;
    await second.close();
    await fixture.stopHost(host);

    const ledger = await fixture.readTurn(turnId);
    assert.equal(ledger.runs.length, 1);
    assert.equal(ledger.userMessages.length, 1);
    assert.equal(ledger.terminalEvents.length, 1);
    assert.equal(ledger.classification.kind, 'fact');
    if (ledger.classification.kind === 'fact') {
      assert.equal(ledger.classification.fact.runStatus, 'cancelled');
      assert.notEqual(ledger.classification.fact.failureClass, 'app_restarted');
    }
  });
});

test('regenerate replays the durable source content with one recoverable root identity', async () => {
  await withExecutionRoot(async (fixture) => {
    const host = await fixture.startHost();
    const client = await connectClient(fixture.root);
    const sourceTurnId = randomUUID();
    const regeneratedTurnId = randomUUID();
    try {
      await client.request(
        'turn.start',
        {
          sessionId: fixture.sessionId,
          turnId: sourceTurnId,
          content: quotedContent('repeat this request'),
        },
        PROCESS_TIMEOUT_MS,
      );
      await waitForTerminalTurn(client, fixture.sessionId, sourceTurnId);

      const started = await client.request(
        'turn.regenerate',
        {
          sessionId: fixture.sessionId,
          sourceTurnId,
          turnId: regeneratedTurnId,
        },
        PROCESS_TIMEOUT_MS,
      );
      const terminal = await waitForTerminalTurn(client, fixture.sessionId, regeneratedTurnId);
      assert.equal(terminal.runId, started.runId);
      assert.deepEqual(
        await client.request('turn.regenerate', {
          sessionId: fixture.sessionId,
          sourceTurnId,
          turnId: regeneratedTurnId,
        }),
        terminal,
      );
    } finally {
      await client.close();
      await fixture.stopHost(host);
    }

    const ledger = await fixture.readTurn(regeneratedTurnId);
    assert.equal(ledger.runs.length, 1);
    assert.equal(ledger.userMessages.length, 1);
    assert.equal(ledger.runs[0]?.opening.lineage?.parentTurnId, sourceTurnId);
    assert.equal(ledger.runs[0]?.opening.lineage?.regeneratedFromTurnId, sourceTurnId);
    assert.deepEqual(
      {
        text: ledger.userMessages[0]?.text,
        quotes: ledger.userMessages[0]?.quotes,
      },
      quotedContent('repeat this request'),
    );
  });
});

test('regenerate rejects self-source and legacy target collisions without draining Host', async () => {
  await withExecutionRoot(async (fixture) => {
    const firstHost = await fixture.startHost();
    const first = await connectClient(fixture.root);
    const sourceTurnId = randomUUID();
    await first.request('turn.start', {
      sessionId: fixture.sessionId,
      turnId: sourceTurnId,
      content: { text: 'source request' },
    });
    await waitForTerminalTurn(first, fixture.sessionId, sourceTurnId);
    await assert.rejects(
      first.request('turn.regenerate', {
        sessionId: fixture.sessionId,
        sourceTurnId,
        turnId: sourceTurnId,
      }),
      operationError('operation_conflict'),
    );
    await first.close();
    await fixture.stopHost(firstHost);

    const legacy = await fixture.seedSafeBoundaryContinuationSource();
    const secondHost = await fixture.startHost();
    const second = await connectClient(fixture.root);
    try {
      await assert.rejects(
        second.request('turn.regenerate', {
          sessionId: fixture.sessionId,
          sourceTurnId,
          turnId: legacy.sourceTurnId,
        }),
        operationError('operation_conflict'),
      );
      const followingTurnId = randomUUID();
      await second.request('turn.start', {
        sessionId: fixture.sessionId,
        turnId: followingTurnId,
        content: { text: 'Host remains available' },
      });
      assert.equal(
        (await waitForTerminalTurn(second, fixture.sessionId, followingTurnId)).status,
        'completed',
      );
    } finally {
      await second.close();
      await fixture.stopHost(secondHost);
    }
    assert.deepEqual(await fixture.readTurnFootprint(legacy.sourceTurnId), {
      admitted: false,
      runCount: 1,
      userMessageCount: 0,
    });
  });
});

test('context actions share root admission and expose backend capability honestly', async () => {
  await withExecutionRoot(async (fixture) => {
    const host = await fixture.startHost();
    const first = await connectClient(fixture.root);
    const second = await connectClient(fixture.root);
    const turnId = randomUUID();
    const unavailableTurnId = randomUUID();
    try {
      assert.deepEqual(
        await first.request('context.diagnostics.query', {
          sessionId: fixture.sessionId,
        }),
        {
          status: 'unavailable',
          reason: 'no_completed_request',
        },
      );
      const started = requireStartedTurn(
        await first.request('turn.start', {
          sessionId: fixture.sessionId,
          turnId,
          content: { text: FAKE_ASK_USER_QUESTION_PROMPT },
        }),
      );
      await waitForRunningTurn(second, fixture.sessionId, turnId);
      await assert.rejects(
        second.request('context.compact', {
          sessionId: fixture.sessionId,
          turnId: randomUUID(),
        }),
        operationError('session_busy'),
      );
      await second.request('turn.stop', {
        sessionId: fixture.sessionId,
        turnId,
        runId: started.runId,
      });
      await assert.rejects(
        second.request('context.compact', {
          sessionId: fixture.sessionId,
          turnId: unavailableTurnId,
        }),
        operationError('operation_unavailable'),
      );
      await assert.rejects(
        second.request('context.diagnostics.query', { sessionId: 'missing-session' }),
        operationError('not_found'),
      );
    } finally {
      await Promise.allSettled([first.close(), second.close()]);
      await fixture.stopHost(host);
    }
    assert.deepEqual(await fixture.readTurnFootprint(unavailableTurnId), {
      admitted: false,
      runCount: 0,
      userMessageCount: 0,
    });
  });
});

test('a disconnected Client leaves a durable Interaction that another Client can answer', async () => {
  await withExecutionRoot(async (fixture) => {
    const firstHost = await fixture.startHost();
    const first = await connectClient(fixture.root);
    const turnId = randomUUID();
    const started = requireStartedTurn(
      await first.request('turn.start', {
        sessionId: fixture.sessionId,
        turnId,
        content: { text: FAKE_ASK_USER_QUESTION_PROMPT },
      }),
    );
    await first.close();

    const second = await connectClient(fixture.root);
    const subscription = await second.openSessionSubscription({
      sessionId: fixture.sessionId,
      transcript: { kind: 'none' },
    });
    const probe = new SubscriptionProbe(subscription);
    const pending = await waitForPendingInteraction(subscription, probe, started.runId);
    assert.equal(pending.sessionId, fixture.sessionId);
    assert.equal(pending.turnId, turnId);
    assert.equal(pending.runId, started.runId);
    assert.equal(pending.status, 'pending');
    const questionRequest = pending.request;
    assert.ok(questionRequest.kind === 'question');

    assert.deepEqual(
      await second.request('interaction.query', {
        sessionId: fixture.sessionId,
        interactionId: pending.interactionId,
      }),
      pending,
    );
    const answer = {
      kind: 'question' as const,
      answers: questionRequest.questions.map((question) => question.options[0]?.label ?? null),
    };
    const winner = await second.request('interaction.answer', {
      sessionId: fixture.sessionId,
      interactionId: pending.interactionId,
      answer,
    });
    assert.equal(winner.sessionId, fixture.sessionId);
    assert.equal(winner.turnId, turnId);
    assert.equal(winner.runId, started.runId);
    assert.equal(winner.status, 'answered');
    assert.equal(winner.outcome.kind, 'question_answer');
    assert.deepEqual(winner.outcome.answers, answer.answers);
    assert.deepEqual(
      await second.request('interaction.query', {
        sessionId: fixture.sessionId,
        interactionId: pending.interactionId,
      }),
      winner,
    );
    assert.deepEqual(
      await second.request('interaction.answer', {
        sessionId: fixture.sessionId,
        interactionId: pending.interactionId,
        answer,
      }),
      winner,
    );
    const resumed = await probe.waitFor(
      (frame) =>
        frame.kind === 'subscription.session_projection' &&
        frame.snapshot.session.status === 'running' &&
        frame.snapshot.rootTurn?.runId === started.runId &&
        frame.snapshot.rootTurn.status === 'running' &&
        frame.snapshot.interactions.pending.length === 0,
      'continuity did not publish the resumed Turn after the question answer',
    );
    assert.equal(resumed.kind, 'subscription.session_projection');
    const completed = await waitForTerminalTurn(second, fixture.sessionId, turnId);
    assert.equal(completed.runId, started.runId);
    assert.equal(completed.status, 'completed');
    await subscription.close();
    await probe.done;
    await second.close();
    await fixture.stopHost(firstHost);

    const secondHost = await fixture.startHost();
    const observer = await connectClient(fixture.root);
    assert.deepEqual(
      await observer.request('interaction.query', {
        sessionId: fixture.sessionId,
        interactionId: pending.interactionId,
      }),
      winner,
    );
    assert.deepEqual(
      await observer.request('interaction.answer', {
        sessionId: fixture.sessionId,
        interactionId: pending.interactionId,
        answer,
      }),
      winner,
    );
    assert.deepEqual(
      await observer.request('turn.query', { sessionId: fixture.sessionId, turnId }),
      completed,
    );
    await observer.close();
    await fixture.stopHost(secondHost);
  });
});

test('two UDS Clients settle one hosted sandbox boundary and resume its exact Run', async () => {
  await withExecutionRoot(async (fixture) => {
    const firstHost = await fixture.startHost();
    const starter = await connectClient(fixture.root);
    const first = await connectClient(fixture.root);
    const second = await connectClient(fixture.root);
    const subscription = await first.openSessionSubscription({
      sessionId: fixture.sessionId,
      transcript: { kind: 'none' },
    });
    const probe = new SubscriptionProbe(subscription);
    const turnId = randomUUID();
    const started = requireStartedTurn(
      await starter.request('turn.start', {
        sessionId: fixture.sessionId,
        turnId,
        content: { text: FAKE_ASK_SANDBOX_BOUNDARY_PROMPT },
      }),
    );
    await starter.close();

    const pending = await waitForPendingInteraction(subscription, probe, started.runId);
    assert.equal(pending.sessionId, fixture.sessionId);
    assert.equal(pending.turnId, turnId);
    assert.equal(pending.runId, started.runId);
    assert.equal(pending.status, 'pending');
    assert.equal(pending.request.kind, 'sandbox_boundary');
    if (pending.request.kind !== 'sandbox_boundary') return;
    assert.deepEqual(pending.request.expansion, { network: { enabled: true } });

    const answer = {
      sessionId: fixture.sessionId,
      interactionId: pending.interactionId,
      answer: { kind: 'sandbox_boundary', decision: 'allow' },
    } as const;
    const [firstWinner, secondWinner] = await Promise.all([
      first.request('interaction.answer', answer),
      second.request('interaction.answer', answer),
    ]);
    assert.deepEqual(firstWinner, secondWinner);
    assert.equal(firstWinner.status, 'answered');
    assert.equal(firstWinner.outcome.kind, 'sandbox_boundary_decision');
    if (firstWinner.outcome.kind !== 'sandbox_boundary_decision') return;
    assert.equal(firstWinner.outcome.decision, 'allow');
    assert.equal(firstWinner.outcome.status, 'approved');
    assert.equal(Number.isSafeInteger(firstWinner.outcome.committedAt), true);
    assert.deepEqual(
      await first.request('interaction.query', {
        sessionId: fixture.sessionId,
        interactionId: pending.interactionId,
      }),
      firstWinner,
    );
    await probe.waitFor(
      (frame) =>
        frame.kind === 'subscription.session_projection' &&
        frame.snapshot.rootTurn?.runId === started.runId &&
        frame.snapshot.interactions.pending.length === 0,
      'continuity did not publish the resumed Turn after the sandbox boundary answer',
    );
    const completed = await waitForTerminalTurn(first, fixture.sessionId, turnId);
    assert.equal(completed.runId, started.runId);
    assert.equal(completed.status, 'completed');

    await subscription.close();
    await probe.done;
    await Promise.allSettled([first.close(), second.close()]);
    await fixture.stopHost(firstHost);

    const secondHost = await fixture.startHost();
    const observer = await connectClient(fixture.root);
    assert.deepEqual(
      await observer.request('interaction.query', {
        sessionId: fixture.sessionId,
        interactionId: pending.interactionId,
      }),
      firstWinner,
    );
    assert.deepEqual(
      await observer.request('turn.query', { sessionId: fixture.sessionId, turnId }),
      completed,
    );
    await observer.close();
    await fixture.stopHost(secondHost);
  });
});

type SessionTodoWriteTool = MakaTool<{ todos: SessionTodoItem[] }, string> & {
  parameters: { parse(value: unknown): { todos: SessionTodoItem[] } };
};

async function withOwnedSessionTodoToolPort<T>(
  fixture: ExecutionFixture,
  run: (coordinator: HostSessionTodoCoordinator, tools: MakaTool[]) => Promise<T>,
): Promise<T> {
  const owner = await tryAcquireInteractiveRootOwner(fixture.capability);
  assert.ok(owner);
  if (!owner) throw new Error('Unable to acquire the interactive SessionTodo tool port');
  let writer: Awaited<ReturnType<typeof openInteractiveSessionTodoStoreForWrite>> | undefined;
  try {
    writer = await openInteractiveSessionTodoStoreForWrite(owner.lease);
    const coordinator = new HostSessionTodoCoordinator(
      writer,
      new SessionAdmissionGate(),
      { probeSessionRemoval: async () => ({ kind: 'present' }) },
      () => {},
      () => {},
    );
    return await run(coordinator, buildSessionTodoTools(coordinator));
  } finally {
    writer?.close();
    await owner.close();
  }
}

function requireSessionTodoWriteTool(tools: readonly MakaTool[]): SessionTodoWriteTool {
  const tool = tools.find((candidate) => candidate.name === 'todo_write');
  assert.ok(tool, 'Expected todo_write Runtime tool');
  return tool as SessionTodoWriteTool;
}

function sessionTodoToolContext(fixture: ExecutionFixture): MakaToolContext {
  return {
    sessionId: fixture.sessionId,
    cwd: fixture.root,
    runId: randomUUID(),
    turnId: randomUUID(),
    toolCallId: randomUUID(),
    abortSignal: new AbortController().signal,
    emitOutput: () => {},
  };
}

async function waitForScheduledTaskCompletion(
  client: RuntimeHostConnection,
  taskId: string,
): Promise<ScheduledTask> {
  const deadline = Date.now() + 20_000;
  let last: ScheduledTask | null = null;
  while (Date.now() < deadline) {
    const result = await client.request('scheduled-task.query', {
      kind: 'get',
      taskId,
    });
    if (result.kind === 'task' && result.task?.status === 'completed') return result.task;
    if (result.kind === 'task') last = result.task;
    if (result.kind === 'task' && result.task?.lastError) {
      throw new Error(`ScheduledTask execution failed: ${result.task.lastError}`);
    }
    await new Promise((resolve) => setTimeout(resolve, 50));
  }
  throw new Error(
    `ScheduledTask ${taskId} did not settle before the deadline: ${JSON.stringify(last)}`,
  );
}