// @vitest-environment node

import { mkdtemp, rm } from 'node:fs/promises';
import { tmpdir } from 'node:os';
import path from 'node:path';
import { afterEach, describe, expect, it } from 'vitest';
import { createCodingConversationStore } from '../../electron/coding-projects/conversation-store';
import { createCodingProjectAgent } from '../../electron/coding-projects/project-config';
import {
  createCodingProjectStore,
  createLocalCodingProject,
  createMemoryCodingProjectStorage,
} from '../../electron/coding-projects/project-store';
import { PiConversationRuntime } from '../../electron/coding-runtime/pi/runtime';
import { PiSessionProjectionError } from '../../electron/coding-runtime/pi/session-projector';
import { PiSessionRegistry } from '../../electron/coding-runtime/pi/session-registry';
import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host';
import {
  PiWorkerPool,
  type PiConversationWorker,
} from '../../electron/coding-runtime/pi/worker-pool';
import type { PiWorkerStopReason } from '../../electron/coding-runtime/pi/worker-process';
import { PiProcessError } from '../../electron/coding-runtime/pi/process-errors';
import type {
  PiRpcCommand,
  PiRpcEvent,
  PiRpcRequestOptions,
  PiRpcResponse,
} from '../../electron/coding-runtime/pi/rpc-client';
import { PI_084_TEXT_TURN } from '../fixtures/pi-0.84.2-projector-fixtures';
import { unknownManagedModelCapability, type ManagedModelRequest } from '../../shared/managed-model-capabilities';

const roots: string[] = [];
const NOW = '2026-08-22T15:00:00.000Z';

class RuntimeFakeWorker implements PiConversationWorker {
  readonly requests: PiRpcCommand[] = [];
  readonly stopReasons: PiWorkerStopReason[] = [];
  private stateData: unknown;
  private entriesData: unknown = { entries: [], leafId: null };
  private statsData: unknown = {
    contextUsage: { tokens: 0, contextWindow: 100_000, percent: 0 },
    tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
  };
  private commandsData: unknown = { commands: [] };
  private thinkingLevelsData: unknown = {
    levels: ['off', 'minimal', 'low', 'medium', 'high'],
  };
  private failType: string | null = null;
  private readonly responseGates = new Map<string, Promise<void>>();
  private readonly events = new Set<(event: PiRpcEvent) => void>();
  private readonly invalidations = new Set<(error: PiProcessError) => void>();

  constructor(readonly id: string, readonly generation: number) {
    this.stateData = {
      sessionId: `session-${id}`,
      thinkingLevel: 'medium',
      isStreaming: false,
      isCompacting: false,
      pendingMessageCount: 0,
    };
  }

  async request<T = unknown>(
    command: PiRpcCommand,
    _options?: PiRpcRequestOptions,
  ): Promise<PiRpcResponse<T>> {
    this.requests.push(structuredClone(command));
    await this.responseGates.get(command.type);
    this.responseGates.delete(command.type);
    if (command.type === this.failType) {
      this.failType = null;
      throw new Error(`fake ${command.type} rejection`);
    }
    if (command.type === 'set_thinking_level') {
      const levels = (this.thinkingLevelsData as { levels?: unknown }).levels;
      if (Array.isArray(levels) && levels.includes(command.level)) {
        this.stateData = {
          ...(this.stateData as Record<string, unknown>),
          thinkingLevel: command.level,
        };
      }
    }
    const data = command.type === 'get_state'
      ? this.stateData
      : command.type === 'get_entries'
        ? this.entriesData
        : command.type === 'get_session_stats'
          ? this.statsData
          : command.type === 'get_commands'
            ? this.commandsData
            : command.type === 'get_available_thinking_levels'
              ? this.thinkingLevelsData
              : undefined;
    return {
      type: 'response',
      id: `${this.id}-${this.requests.length}`,
      success: true,
      ...(data === undefined ? {} : { data: structuredClone(data) as T }),
    };
  }

  async send(command: PiRpcCommand): Promise<void> {
    this.requests.push(structuredClone(command));
  }

  setSessionData(input: {
    state?: unknown;
    entries?: unknown;
    stats?: unknown;
    commands?: unknown;
    thinkingLevels?: unknown;
  }): void {
    if (input.state !== undefined) this.stateData = structuredClone(input.state);
    if (input.entries !== undefined) this.entriesData = structuredClone(input.entries);
    if (input.stats !== undefined) this.statsData = structuredClone(input.stats);
    if (input.commands !== undefined) this.commandsData = structuredClone(input.commands);
    if (input.thinkingLevels !== undefined) {
      this.thinkingLevelsData = structuredClone(input.thinkingLevels);
    }
  }

  failNext(type: string): void { this.failType = type; }

  holdNext(type: string): () => void {
    let release!: () => void;
    this.responseGates.set(type, new Promise<void>((resolve) => { release = resolve; }));
    return release;
  }

  subscribe(listener: (event: PiRpcEvent) => void): () => void {
    this.events.add(listener);
    return () => this.events.delete(listener);
  }

  subscribeInvalidation(listener: (error: PiProcessError) => void): () => void {
    this.invalidations.add(listener);
    return () => this.invalidations.delete(listener);
  }

  emit(event: PiRpcEvent): void {
    for (const listener of this.events) listener(event);
  }

  invalidate(error = new PiProcessError('PI_RPC_EXITED', 'fake worker crashed')): void {
    for (const listener of this.invalidations) listener(error);
  }

  async stop(reason: PiWorkerStopReason) {
    this.stopReasons.push(reason);
    return { mode: 'stdin-close' as const, code: 0, signal: null };
  }
}

class TrackingExtensionHost extends PiManagedExtensionHost {
  managedRequest?: ManagedModelRequest;
  readonly runs = new Map<string, { generation: number; runId: string }>();

  override async bindRun(conversationId: string, generation: number, runId: string, request?: ManagedModelRequest): Promise<void> {
    this.managedRequest = request;
    this.runs.set(conversationId, { generation, runId });
  }

  override async clearRun(conversationId: string, _generation: number, runId?: string): Promise<void> {
    if (!runId || this.runs.get(conversationId)?.runId === runId) this.runs.delete(conversationId);
  }
}

afterEach(async () => {
  await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
});

describe('Pi Conversation runtime', () => {

  it.each(['What is in this picture?', ''])('keeps submitted image references visible before Pi emits its user message (%s)', async (text) => {
    const projectPath = await mkdtemp(path.join(tmpdir(), 'makelore-image-runtime-'));
    roots.push(projectPath);
    const projectStore = createCodingProjectStore(createMemoryCodingProjectStorage(), {
      createId: () => 'project-image', now: () => NOW,
    });
    await createLocalCodingProject({ projectPath, now: NOW }, projectStore);
    const model = { accountId: 'account-image', modelId: 'deepseek-flash', thinkingLevel: 'off' as const };
    await createCodingProjectAgent(projectPath, {
      id: 'agent-image', avatarId: 'avatar-01', roleName: 'Reader', name: 'Reader',
      model, modelResolution: 'resolved',
      responsibility: { mission: 'Read images', owns: [], boundaries: [], collaborators: [], principles: [] },
    }, { now: NOW });
    const conversation = await createCodingConversationStore(projectPath).create({
      agentId: 'agent-image', title: 'Image', model, modelResolution: 'resolved',
    });
    let worker: RuntimeFakeWorker;
    const runtime = new PiConversationRuntime({
      pool: new PiWorkerPool({ openWorker: async ({ generation }) => {
        worker = new RuntimeFakeWorker('image-worker', generation);
        return { worker, session: { piSessionId: 'image-session', sessionKey: 'image-key' } };
      } }),
      registry: new PiSessionRegistry({ projectStore }),
      resolveModel: async (candidate) => ({ ...candidate, runtimeProviderId: 'image-provider', input: ['text', 'image'] }),
      resolveImages: async () => [{ type: 'image', data: 'aW1hZ2U=', mimeType: 'image/png' }],
    });
    try {
      await runtime.prepare({ conversationId: conversation.id, projectId: 'project-image',
        agentId: 'agent-image', title: 'Image', model: { model, modelResolution: 'resolved' } });
      await runtime.prompt({ conversationId: conversation.id, clientRequestId: 'image-request',
        mode: 'prompt', text, attachments: [{ attachmentId: 'image-attachment' }] });
      expect(worker!.requests.find(command => command.type === 'prompt')).toMatchObject({
        images: [{ type: 'image', data: 'aW1hZ2U=', mimeType: 'image/png' }],
      });
      const snapshot = await runtime.getSnapshot(conversation.id);
      expect(snapshot.nodes).toContainEqual(expect.objectContaining({
        role: 'user', blocks: expect.arrayContaining([expect.objectContaining({
          kind: 'image', attachmentId: 'image-attachment', mime: 'image/png',
        })]),
      }));
    } finally { await runtime.shutdown(); }
  });
  it('persists native managed choices and rejects removed effort or unsupported images before prompt', async () => {
    const projectPath = await mkdtemp(path.join(tmpdir(), 'makelore-managed-runtime-'));
    roots.push(projectPath);
    const projectStore = createCodingProjectStore(createMemoryCodingProjectStorage(), {
      createId: () => 'project-managed', now: () => NOW,
    });
    await createLocalCodingProject({ projectPath, now: NOW }, projectStore);
    const model = { accountId: 'niancode-user-models', modelId: 'unfamiliar-model', thinkingLevel: 'high' as const };
    await createCodingProjectAgent(projectPath, {
      id: 'agent-a', avatarId: 'avatar-01', roleName: 'Implementer', name: 'Agent A',
      model, modelResolution: 'resolved',
      responsibility: { mission: 'Implement', owns: [], boundaries: [], collaborators: [], principles: [] },
    }, { now: NOW });
    const store = createCodingConversationStore(projectPath);
    const conversation = await store.create({ agentId: 'agent-a', title: 'Managed', model, modelResolution: 'resolved' });
    let worker: RuntimeFakeWorker;
    const pool = new PiWorkerPool({ openWorker: async ({generation}) => {
      worker = new RuntimeFakeWorker('managed-worker', generation);
      return { worker, session: { piSessionId: 'session', sessionKey: 'key' } };
    } });
    const capability = unknownManagedModelCapability();
    capability.inputModalities = ['text'];
    capability.reasoning = { supported: true, canDisable: true, defaultEnabled: true,
      controlFormat: 'qwen', effortValues: ['xhigh', 'future-native'], defaultEffort: 'xhigh', budget: null };
    const host = new TrackingExtensionHost();
    const runtime = new PiConversationRuntime({
      pool, registry: new PiSessionRegistry({ projectStore }), extensionHost: host,
      resolveImages: async () => [],
      resolveModel: async candidate => ({ ...candidate, input: ['text'],
        runtimeProviderId: 'managed', managedCapability: structuredClone(capability) }),
    });
    try {
      await runtime.prepare({ conversationId: conversation.id, projectId: 'project-managed', agentId: 'agent-a',
        title: conversation.title, model: { model, modelResolution: 'resolved' } });
      expect((await runtime.getSnapshot(conversation.id)).conversation.model.model?.reasoningChoice).toEqual({ mode: 'default' });
      await runtime.setThinking({ conversationId: conversation.id, thinkingLevel: 'off',
        reasoningChoice: { mode: 'enabled', effort: 'future-native' } });
      expect((await store.get(conversation.id))?.model?.reasoningChoice).toEqual({ mode: 'enabled', effort: 'future-native' });
      capability.reasoning.effortValues = ['xhigh'];
      const prompt = { conversationId: conversation.id, clientRequestId: 'request-managed', mode: 'prompt' as const, text: 'Hello', attachments: [] };
      await expect(runtime.prompt(prompt)).rejects.toThrow('思考选项已不可用');
      expect(worker!.requests.filter(r => r.type === 'prompt')).toHaveLength(0);
      await runtime.setThinking({ conversationId: conversation.id, thinkingLevel: 'off', reasoningChoice: { mode: 'default' } });
      await expect(runtime.prompt({ ...prompt, attachments: [{ attachmentId: 'image' }] })).rejects.toThrow('图片输入');
      await runtime.prompt(prompt);
      expect(host.managedRequest).toEqual({ modelId: 'unfamiliar-model', choice: { mode: 'default' }, reasoningFields: {} });
      for (const event of PI_084_TEXT_TURN.events) worker!.emit(structuredClone(event));
      await runtime.getSnapshot(conversation.id);
    } finally { await runtime.shutdown(); }
  });

  it('separates RPC acceptance from settle and changes only the target Conversation model', async () => {
    const projectPath = await mkdtemp(path.join(tmpdir(), 'makelore-pi-runtime-'));
    roots.push(projectPath);
    const projectStore = createCodingProjectStore(createMemoryCodingProjectStorage(), {
      createId: () => 'project-a',
      now: () => NOW,
    });
    await createLocalCodingProject({ projectPath, now: NOW }, projectStore);
    await createCodingProjectAgent(projectPath, {
      id: 'agent-a',
      avatarId: 'avatar-01',
      roleName: 'Implementer',
      name: 'Agent A',
      model: { accountId: 'account-a', modelId: 'model-a', thinkingLevel: 'medium' },
      modelResolution: 'resolved',
      responsibility: { mission: 'Implement', owns: [], boundaries: [], collaborators: [], principles: [] },
    }, { now: NOW });
    const store = createCodingConversationStore(projectPath, {
      createId: (() => {
        const ids = [
          'f47ac10b-58cc-4372-a567-0e02b2c3d479',
          '8b1a9953-c461-4d88-9c3e-7e1f8f3f2c11',
          '3d594650-3436-4a8c-8b38-7d1c5e3f9a20',
        ];
        return () => ids.shift() as string;
      })(),
      now: () => NOW,
    });
    const model = { accountId: 'account-a', modelId: 'model-a', thinkingLevel: 'medium' as const };
    const left = await store.create({ agentId: 'agent-a', title: 'Left', model, modelResolution: 'resolved' });
    const right = await store.create({ agentId: 'agent-a', title: 'Right', model, modelResolution: 'resolved' });
    const forkTarget = await store.create({ agentId: 'agent-a', title: 'Fork', model, modelResolution: 'resolved' });
    const inputs = [left, right, forkTarget].map((item) => ({
      conversationId: item.id,
      projectId: 'project-a',
      agentId: 'agent-a',
      title: item.title,
      model: { model: item.model, modelResolution: item.modelResolution },
    }));
    const workers = new Map<string, RuntimeFakeWorker>();
    const workerHistory = new Map<string, RuntimeFakeWorker[]>();
    const durableSessions = new Map<string, Parameters<RuntimeFakeWorker['setSessionData']>[0]>();
    const openInputs: Array<{ conversationId: string; forkSource?: string; sourceEntryId?: string }> = [];
    const pool = new PiWorkerPool({
      maxIdle: 4,
      openWorker: async ({ conversation, fork, generation, existingSession }) => {
        openInputs.push({
          conversationId: conversation.conversationId,
          ...(fork ? {
            forkSource: fork.sourceSession.piSessionId,
            sourceEntryId: fork.sourceEntryId,
          } : {}),
        });
        const worker = new RuntimeFakeWorker(`worker-${conversation.conversationId}-${generation}`, generation);
        const durable = durableSessions.get(conversation.conversationId);
        if (durable) worker.setSessionData(durable);
        workers.set(conversation.conversationId, worker);
        workerHistory.set(conversation.conversationId, [
          ...(workerHistory.get(conversation.conversationId) ?? []),
          worker,
        ]);
        return {
          worker,
          session: existingSession ?? {
            piSessionId: `session-${conversation.conversationId}`,
            sessionKey: `key-${conversation.conversationId}`,
          },
        };
      },
    });
    const trackingHost = new TrackingExtensionHost();
    let acquiredBackgroundLeases = 0;
    let releasedBackgroundLeases = 0;
    let activeBackgroundLeases = 0;
    const runtime = new PiConversationRuntime({
      pool,
      registry: new PiSessionRegistry({ projectStore }),
      createId: (kind) => `${kind}-fixed`,
      extensionHost: trackingHost,
      acquireBackgroundLease: () => {
        acquiredBackgroundLeases += 1;
        activeBackgroundLeases += 1;
        let released = false;
        return () => {
          if (released) throw new Error('background lease released twice');
          released = true;
          releasedBackgroundLeases += 1;
          activeBackgroundLeases -= 1;
        };
      },
      resolveModel: async (candidate) => {
        if (candidate.accountId !== 'account-b'
          || !['model-b', 'model-c'].includes(candidate.modelId)) {
          throw new Error('model unavailable');
        }
        return {
          accountId: candidate.accountId,
          runtimeProviderId: 'runtime-account-b',
          modelId: candidate.modelId,
          thinkingLevel: candidate.thinkingLevel,
          input: ['text'],
        };
      },
    });
    await Promise.all(inputs.slice(0, 2).map((input) => runtime.prepare(input)));
    workers.get(left.id)!.setSessionData({
      commands: {
        commands: [{ name: 'custom', description: 'Custom command', provider: 'raw-provider' }],
        apiKey: 'raw-secret',
      },
    });
    expect(await runtime.listCommands(left.id)).toEqual([
      { name: 'custom', description: 'Custom command' },
    ]);
    expect(await runtime.listCommands('00000000-0000-4000-8000-000000000000')).toEqual([]);
    const leftGenerationOneSeqs: number[] = [];
    const crashPatchOps: string[] = [];
    const unsubscribe = runtime.subscribe((envelope) => {
      if (envelope.conversationId === left.id && envelope.workerGeneration === 1) {
        leftGenerationOneSeqs.push(envelope.seq);
      }
      if (envelope.conversationId === left.id
        && envelope.workerGeneration === 3
        && (envelope.patch.op === 'run.state' || envelope.patch.op === 'worker.state')) {
        crashPatchOps.push(envelope.patch.op);
      }
    });

    const beforeNoopPrepare = await runtime.getSnapshot(left.id);
    await runtime.prepare(inputs[0]!);
    expect(await runtime.getSnapshot(left.id)).toEqual(beforeNoopPrepare);
    expect(leftGenerationOneSeqs).toEqual([]);

    const releasePromptAcceptance = workers.get(left.id)!.holdNext('prompt');
    let acceptanceResolved = false;
    const acceptance = runtime.prompt({
      clientRequestId: 'request-left',
      conversationId: left.id,
      mode: 'prompt',
      text: 'Implement the change',
      attachments: [],
    }).then((value) => {
      acceptanceResolved = true;
      return value;
    });
    await expect.poll(() => workers.get(left.id)!.requests.at(-1)?.type).toBe('prompt');
    expect(workers.get(left.id)!.requests).toHaveLength(6);
    expect(acceptanceResolved).toBe(false);
    releasePromptAcceptance();
    const accepted = await acceptance;
    expect(accepted).toMatchObject({ accepted: true, runId: 'run-fixed', mode: 'prompt' });
    expect(activeBackgroundLeases).toBe(1);
    expect((await runtime.getSnapshot(left.id)).run.status).toBe('running');
    for (const event of PI_084_TEXT_TURN.events) {
      workers.get(left.id)!.emit(structuredClone(event));
    }
    const streamed = await runtime.getSnapshot(left.id);
    expect(streamed.nodes).toContainEqual(expect.objectContaining({
      kind: 'message',
      id: 'client:request-left',
      clientRequestId: 'request-left',
      status: 'complete',
    }));
    expect(JSON.stringify(streamed.nodes)).toContain('Implemented');
    workers.get(left.id)!.emit({
      type: 'extension_ui_request',
      id: 'question-runtime',
      method: 'select',
      title: 'Choose implementation',
      options: ['Small seam', 'Large seam'],
    });
    await expect.poll(async () => (await runtime.getSnapshot(left.id)).pendingInteractions)
      .toContainEqual(expect.objectContaining({ id: 'question-runtime', status: 'pending' }));
    await runtime.respondInteraction(left.id, {
      interactionId: 'question-runtime',
      optionId: 'question-runtime:option:0',
    });
    expect(workers.get(left.id)!.requests.at(-1)).toEqual({
      type: 'extension_ui_response', id: 'question-runtime', value: 'Small seam',
    });
    expect((await runtime.getSnapshot(left.id)).pendingInteractions).not.toContainEqual(
      expect.objectContaining({ id: 'question-runtime' }),
    );
    const durable = {
      state: {
        sessionId: `session-${left.id}`,
        thinkingLevel: 'medium',
        isStreaming: false,
        isCompacting: false,
        pendingMessageCount: 0,
      },
      entries: PI_084_TEXT_TURN.entries,
      stats: PI_084_TEXT_TURN.stats,
    };
    durableSessions.set(left.id, durable);
    workers.get(left.id)!.setSessionData(durable);
    workers.get(left.id)!.emit({ type: 'agent_end' });
    expect((await runtime.getSnapshot(left.id)).run.status).toBe('running');
    const checkpoint = await runtime.getSnapshot(left.id);
    expect(checkpoint.cursor.leafEntryId).toBe('entry-assistant-a');
    expect(checkpoint.nodes).toContainEqual(expect.objectContaining({
      kind: 'message',
      id: 'client:request-left',
      sourceEntryId: 'entry-user-a',
    }));
    const releaseSettledHydration = workers.get(left.id)!.holdNext('get_entries');
    workers.get(left.id)!.emit({ type: 'agent_settled' });
    let continuationAccepted = false;
    const continuation = runtime.followUp({
      clientRequestId: 'request-after-settled',
      conversationId: left.id,
      text: 'Continue only after the checkpoint',
      attachments: [],
    }).then((result) => {
      continuationAccepted = true;
      return result;
    });
    await expect.poll(() => workers.get(left.id)!.requests.filter(
      ({ type }) => type === 'get_entries',
    ).length).toBe(3);
    expect(workers.get(left.id)!.requests.some(({ type }) => type === 'follow_up')).toBe(false);
    expect(continuationAccepted).toBe(false);
    releaseSettledHydration();
    await expect(continuation).resolves.toMatchObject({ mode: 'follow-up', queuePosition: 1 });
    workers.get(left.id)!.emit({ type: 'agent_settled' });
    await expect.poll(async () => (await runtime.getSnapshot(left.id)).queue.items).toEqual([]);
    expect((await runtime.getSnapshot(left.id)).run.status).toBe('idle');
    expect(activeBackgroundLeases).toBe(0);
    const settledNodes = (await runtime.getSnapshot(left.id)).nodes;
    expect(settledNodes).toEqual(checkpoint.nodes);
    expect(leftGenerationOneSeqs).toEqual(leftGenerationOneSeqs.map((_, index) => index + 1));
    expect((await runtime.getSnapshot(left.id)).cursor.seq).toBe(leftGenerationOneSeqs.at(-1));

    const changed = await runtime.setModel({
      conversationId: left.id,
      accountId: 'account-b',
      modelId: 'model-b',
    });
    expect(changed.model).toEqual({ accountId: 'account-b', modelId: 'model-b', thinkingLevel: 'medium' });
    expect(workerHistory.get(left.id)).toHaveLength(2);
    expect(workerHistory.get(left.id)![0]!.requests).not.toContainEqual(expect.objectContaining({
      type: 'set_model',
    }));
    expect(workers.get(left.id)!.requests.map(({ type }) => type).sort()).toEqual([
      'get_available_thinking_levels',
      'get_entries',
      'get_session_stats',
      'get_state',
    ]);
    expect(pool.getState(left.id)).toMatchObject({ generation: 2, state: 'ready' });
    expect(pool.getState(right.id)).toMatchObject({ generation: 1, state: 'ready' });
    expect((await runtime.getSnapshot(right.id)).conversation.model.model).toEqual(model);
    expect((await runtime.getSnapshot(left.id)).nodes).toEqual(settledNodes);

    const generationBeforeSameAccountModelChange = pool.getState(left.id)!.generation;
    const sameAccountChanged = await runtime.setModel({
      conversationId: left.id,
      accountId: 'account-b',
      modelId: 'model-c',
    });
    expect(sameAccountChanged.model).toEqual({
      accountId: 'account-b',
      modelId: 'model-c',
      thinkingLevel: 'medium',
    });
    expect(pool.getState(left.id)!.generation).toBe(generationBeforeSameAccountModelChange);
    expect(workerHistory.get(left.id)).toHaveLength(2);
    expect(workers.get(left.id)!.requests.slice(-3).map(({ type }) => type)).toEqual([
      'set_model',
      'get_available_thinking_levels',
      'get_state',
    ]);

    await expect(runtime.setModel({
      conversationId: left.id,
      accountId: 'account-missing',
      modelId: 'model-missing',
    })).rejects.toThrow('model unavailable');
    expect((await runtime.getSnapshot(left.id)).conversation.model).toEqual(sameAccountChanged);

    await runtime.prompt({
      clientRequestId: 'request-left-active',
      conversationId: left.id,
      mode: 'prompt',
      text: 'Keep working',
      attachments: [],
    });
    await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('running');
    workers.get(left.id)!.failNext('steer');
    await expect(runtime.steer({
      clientRequestId: 'request-left-rejected-steer',
      conversationId: left.id,
      text: 'This command is rejected',
      attachments: [],
    })).rejects.toThrow('fake steer rejection');
    expect((await runtime.getSnapshot(left.id)).queue.items).toEqual([]);
    await expect(runtime.steer({
      clientRequestId: 'request-left-steer',
      conversationId: left.id,
      text: 'Use the smaller seam',
      attachments: [],
    })).resolves.toMatchObject({ accepted: true, mode: 'steer', queuePosition: 1 });
    await expect(runtime.followUp({
      clientRequestId: 'request-left-follow-up',
      conversationId: left.id,
      text: 'Then run the focused test',
      attachments: [],
    })).resolves.toMatchObject({ accepted: true, mode: 'follow-up', queuePosition: 2 });
    const generationBeforeThinking = pool.getState(left.id)!.generation;
    await expect(runtime.setThinking({
      conversationId: left.id,
      thinkingLevel: 'high',
    })).resolves.toMatchObject({
      model: { accountId: 'account-b', modelId: 'model-c', thinkingLevel: 'high' },
    });
    expect(pool.getState(left.id)!.generation).toBe(generationBeforeThinking);
    workers.get(left.id)!.failNext('abort');
    await expect(runtime.abort(left.id)).rejects.toThrow('fake abort rejection');
    expect((await runtime.getSnapshot(left.id)).run.status).toBe('running');
    await runtime.abort(left.id);
    expect((await runtime.getSnapshot(left.id)).run.status).toBe('aborting');
    expect(workers.get(left.id)!.requests.slice(-8).map(({ type }) => type)).toEqual([
      'steer',
      'steer',
      'follow_up',
      'get_available_thinking_levels',
      'set_thinking_level',
      'get_state',
      'abort',
      'abort',
    ]);
    expect(workers.get(right.id)!.requests.map(({ type }) => type)).toEqual([
      'get_state',
      'get_entries',
      'get_session_stats',
      'get_available_thinking_levels',
    ]);
    workers.get(left.id)!.emit({ type: 'agent_settled' });
    await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('idle');
    expect(activeBackgroundLeases).toBe(0);
    expect((await runtime.getSnapshot(left.id)).queue.items).toEqual([]);

    workers.get(right.id)!.failNext('compact');
    await expect(runtime.compact(right.id)).rejects.toThrow('fake compact rejection');
    expect((await runtime.getSnapshot(right.id)).run.status).toBe('error');
    expect(activeBackgroundLeases).toBe(0);
    await runtime.compact(right.id);
    await expect.poll(async () => (await runtime.getSnapshot(right.id)).run.status).toBe('idle');
    expect(activeBackgroundLeases).toBe(0);
    expect(workers.get(right.id)!.requests.at(-1)).toEqual({ type: 'compact' });
    expect((await runtime.getSnapshot(left.id)).run.status).toBe('idle');
    const rightDurable = {
      entries: {
        leafId: 'entry-right-user',
        entries: [{
          type: 'message',
          id: 'entry-right-user',
          parentId: null,
          timestamp: NOW,
          message: { role: 'user', content: 'Right history', timestamp: 1 },
        }],
      },
    };
    durableSessions.set(right.id, rightDurable);
    workers.get(right.id)!.setSessionData(rightDurable);
    workers.get(right.id)!.emit({ type: 'agent_end' });
    expect((await runtime.getSnapshot(right.id)).run.status).toBe('idle');
    workers.get(right.id)!.emit({ type: 'agent_settled' });
    await expect.poll(async () => (await runtime.getSnapshot(right.id)).run.status).toBe('idle');
    expect(activeBackgroundLeases).toBe(0);

    const leftBeforeFailedRecovery = await runtime.getSnapshot(left.id);
    const rightBeforeFailedRecovery = await runtime.getSnapshot(right.id);
    const corruptRightSession = {
      entries: {
        leafId: 'entry-corrupt',
        entries: [{
          type: 'message',
          id: 'entry-corrupt',
          parentId: 'missing-parent',
          timestamp: NOW,
          message: { role: 'user', content: 'must not replace history', timestamp: 1 },
        }],
      },
    };
    durableSessions.set(right.id, corruptRightSession);
    await expect(runtime.recover(right.id)).rejects.toBeInstanceOf(PiSessionProjectionError);
    expect((await runtime.getSnapshot(right.id)).nodes).toEqual(rightBeforeFailedRecovery.nodes);
    expect((await runtime.getSnapshot(right.id)).worker).toMatchObject({
      status: 'error',
      generation: 2,
      error: { code: 'CODING_SESSION_UNREADABLE', recoverable: true },
    });
    expect(await runtime.getSnapshot(left.id)).toEqual(leftBeforeFailedRecovery);

    durableSessions.set(right.id, rightDurable);
    await expect(runtime.recover(right.id)).resolves.toMatchObject({
      conversationId: right.id,
      status: 'ready',
      workerGeneration: 3,
    });
    expect((await runtime.getSnapshot(right.id)).cursor).toMatchObject({
      workerGeneration: 3,
      seq: 0,
    });
    expect(workers.get(right.id)!.requests.map(({ type }) => type).sort()).toEqual([
      'get_available_thinking_levels',
      'get_entries',
      'get_session_stats',
      'get_state',
    ]);
    expect(pool.getState(left.id)).toMatchObject({ generation: 2, state: 'idle' });

    const sourceRequestCount = workers.get(left.id)!.requests.length;
    const forked = await runtime.fork({
      sourceConversationId: left.id,
      sourceEntryId: 'entry-a',
      conversation: inputs[2]!,
    });
    expect(forked.conversationId).toBe(forkTarget.id);
    expect(workers.get(forkTarget.id)?.id).toBe(`worker-${forkTarget.id}-1`);
    expect(workers.get(left.id)!.requests).toHaveLength(sourceRequestCount);
    expect(openInputs.at(-1)).toEqual({
      conversationId: forkTarget.id,
      forkSource: `session-${left.id}`,
      sourceEntryId: 'entry-a',
    });

    await runtime.dispose(forkTarget.id, 'test_injection');
    expect(pool.getState(forkTarget.id)).toBeNull();
    expect(pool.getState(left.id)).toMatchObject({ generation: 2, state: 'idle' });
    await expect(runtime.getSnapshot(forkTarget.id)).rejects.toMatchObject({
      publicError: { code: 'CODING_CONVERSATION_NOT_FOUND' },
    });
    await runtime.prompt({
      clientRequestId: 'request-active-recover',
      conversationId: left.id,
      mode: 'prompt',
      text: 'Recover this active run',
      attachments: [],
    });
    expect(trackingHost.runs.get(left.id)?.runId).toBe('run-fixed');
    await runtime.recover(left.id);
    expect(trackingHost.runs.has(left.id)).toBe(false);
    expect((await runtime.getSnapshot(left.id)).run).toEqual({ status: 'idle' });
    expect(activeBackgroundLeases).toBe(0);

    workers.get(left.id)!.setSessionData({
      thinkingLevels: { levels: ['high'] },
      state: {
        sessionId: `session-${left.id}`,
        thinkingLevel: 'high',
        isStreaming: false,
        isCompacting: false,
        pendingMessageCount: 0,
      },
    });
    await expect(runtime.setThinking({
      conversationId: left.id,
      thinkingLevel: 'off',
    })).rejects.toMatchObject({
      publicError: { code: 'CODING_MODEL_UNAVAILABLE', recoverable: true },
    });
    expect((await runtime.getSnapshot(left.id)).conversation.model).toMatchObject({
      model: { thinkingLevel: 'medium' },
      availableThinkingLevels: ['high'],
    });

    const rightBeforeCrash = await runtime.getSnapshot(right.id);
    const requestsBeforeCrash = workers.get(left.id)!.requests.length;
    await runtime.prompt({
      clientRequestId: 'request-worker-exit',
      conversationId: left.id,
      mode: 'prompt',
      text: 'This accepted prompt must not be replayed',
      attachments: [],
    });
    expect(trackingHost.runs.has(left.id)).toBe(true);
    crashPatchOps.length = 0;
    workers.get(left.id)!.invalidate();
    await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('error');
    expect(activeBackgroundLeases).toBe(0);
    expect((await runtime.getSnapshot(left.id))).toMatchObject({
      run: {
        status: 'error',
        terminalReason: 'failed',
        error: {
          code: 'CODING_RUNTIME_START_FAILED',
          message: '本地 Agent 已中断，原请求未自动重发。',
          recoverable: true,
        },
      },
      worker: { status: 'error', generation: 3 },
    });
    expect(trackingHost.runs.has(left.id)).toBe(false);
    expect(crashPatchOps).toEqual(['run.state', 'worker.state']);
    await expect(runtime.abort(left.id)).resolves.toBeUndefined();
    expect((await runtime.getSnapshot(left.id)).run.status).toBe('error');
    expect(await runtime.getSnapshot(right.id)).toEqual(rightBeforeCrash);

    await runtime.recover(left.id);
    expect(workers.get(left.id)!.requests.slice(0, requestsBeforeCrash)).not.toContainEqual({
      type: 'prompt',
      message: 'This accepted prompt must not be replayed',
    });

    await runtime.prompt({
      clientRequestId: 'request-settled-before-close',
      conversationId: left.id,
      mode: 'prompt',
      text: 'Settle before close',
      attachments: [],
    });
    workers.get(left.id)!.emit({ type: 'agent_settled' });
    workers.get(left.id)!.invalidate();
    await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('idle');
    expect((await runtime.getSnapshot(left.id)).run.terminalReason).toBe('completed');

    await runtime.recover(left.id);
    await runtime.prompt({
      clientRequestId: 'request-protocol-invalidation',
      conversationId: left.id,
      mode: 'prompt',
      text: 'Fail this generation without replay',
      attachments: [],
    });
    workers.get(left.id)!.invalidate(new PiProcessError(
      'PI_RPC_PROTOCOL_ERROR',
      'strict JSONL protocol failure',
    ));
    await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('error');
    expect(activeBackgroundLeases).toBe(0);
    expect((await runtime.getSnapshot(left.id)).run).toMatchObject({
      terminalReason: 'failed',
      error: {
        code: 'CODING_RUNTIME_PROTOCOL_ERROR',
        message: '本地 Agent 通信异常，当前对话已停止。',
        recoverable: true,
      },
    });
    workers.get(left.id)!.emit({ type: 'agent_settled' });
    await new Promise((resolve) => setImmediate(resolve));
    expect((await runtime.getSnapshot(left.id)).run.status).toBe('error');

    await runtime.recover(left.id);
    const siblingBeforeForcedDispose = await runtime.getSnapshot(right.id);
    const bindingBeforeForcedDispose = await store.get(left.id);
    const forcedDisposePatches: string[] = [];
    const unsubscribeForcedDispose = runtime.subscribe((envelope) => {
      if (envelope.conversationId === left.id && envelope.patch.op === 'run.state') {
        forcedDisposePatches.push(JSON.stringify(envelope.patch));
      }
    });
    const forcedWorker = workers.get(left.id)!;
    await runtime.prompt({
      clientRequestId: 'request-project-deactivated',
      conversationId: left.id,
      mode: 'prompt',
      text: 'Do not replay this accepted prompt',
      attachments: [],
    });
    const forcedPromptCount = forcedWorker.requests.filter(({ type }) => type === 'prompt').length;
    await runtime.dispose(left.id, 'project_deactivated');
    unsubscribeForcedDispose();
    expect(forcedDisposePatches.at(-1)).toContain('CODING_RUNTIME_START_FAILED');
    expect(forcedDisposePatches.at(-1)).toContain('"recoverable":true');
    expect(forcedWorker.stopReasons.at(-1)).toBe('project_deactivated');
    expect(activeBackgroundLeases).toBe(0);
    expect(await runtime.getSnapshot(right.id)).toEqual(siblingBeforeForcedDispose);
    expect((await store.get(left.id))?.sessionKey).toBe(bindingBeforeForcedDispose?.sessionKey);

    const persistedAfterForcedDispose = (await store.get(left.id))!;
    await runtime.prepare({
      ...inputs[0]!,
      model: {
        model: persistedAfterForcedDispose.model,
        modelResolution: persistedAfterForcedDispose.modelResolution,
      },
    });
    expect(forcedWorker.requests.filter(({ type }) => type === 'prompt')).toHaveLength(
      forcedPromptCount,
    );
    expect(acquiredBackgroundLeases).toBe(releasedBackgroundLeases);
    expect(activeBackgroundLeases).toBe(0);
    unsubscribe();
  });
});