import assert from 'node:assert/strict';
import test from 'node:test';
import { OrchestratorService } from '@platform-modules/chatgpt-orchestrator-core';
import { InMemoryRepository } from '@platform-modules/chatgpt-orchestrator-persistence';
import type { ExecutionAttempt, WorkerId } from '@platform-modules/chatgpt-orchestrator-protocol';
import { commandForExecutionAttempt, ExecutionContinuityScheduler } from '../src/execution-continuity.js';

async function fixture(policy: 'manual' | 'auto' = 'manual') {
  const service = new OrchestratorService(new InMemoryRepository());
  const scheduler = new ExecutionContinuityScheduler(service, undefined, { reconcileIntervalMs: 10_000 });
  const { run, rootWorker } = await service.createRun({ title: 'continuity test' });
  await service.setRunContinuationPolicy(run.runId, { mode: policy, idleGraceMs: 0 });
  const { worker } = await service.spawnWorker({
    runId: run.runId,
    parentWorkerId: rootWorker.workerId,
    name: 'child',
    assignment: { objective: 'finish the test assignment' },
  });
  const initial = await scheduler.dispatchAttemptForReason(worker.workerId, 'initial');
  const createCommand = commandForExecutionAttempt(initial);
  await service.acknowledgeExecutorCommand(createCommand.commandId, 'executor-test');
  await service.recordConversationState({
    workerId: worker.workerId,
    executorId: 'executor-test',
    state: 'generating',
    conversationId: `WEB:${crypto.randomUUID()}`,
  });
  await service.attachWorker(worker.workerId);
  return { service, scheduler, run, rootWorker, worker, initial, createCommand };
}

async function becomeIdle(service: OrchestratorService, workerId: WorkerId): Promise<void> {
  await service.recordConversationState({ workerId, executorId: 'executor-test', state: 'idle' });
}

async function deliverScheduled(service: OrchestratorService, scheduler: ExecutionContinuityScheduler, attempt: ExecutionAttempt): Promise<void> {
  await scheduler.dispatchAttempt(attempt);
  const command = commandForExecutionAttempt(attempt);
  await service.completeExecutorCommand(command.commandId, 'executor-test', true, {});
  await service.recordConversationState({ workerId: attempt.workerId, executorId: 'executor-test', state: 'generating' });
}



test('root initial execution represents the already-running coordinator and does not launch a duplicate root conversation', async () => {
  const service = new OrchestratorService(new InMemoryRepository());
  const scheduler = new ExecutionContinuityScheduler(service);
  const { rootWorker } = await service.createRun({ title: 'root initial' });
  const attempts = await service.listExecutionAttempts(rootWorker.workerId);
  assert.equal(attempts.length, 1);
  assert.equal(attempts[0]?.reason, 'initial');
  assert.equal(attempts[0]?.state, 'generating');

  await scheduler.reconcile();
  assert.equal((await service.listDispatchableExecutorCommands()).length, 0);
});

test('initial child execution is represented by one durable attempt and one idempotent command', async () => {
  const { service, scheduler, worker, initial } = await fixture();
  assert.equal(initial.reason, 'initial');
  assert.equal(initial.sequence, 1);

  await scheduler.dispatchAttempt(initial);
  const attempts = await service.listExecutionAttempts(worker.workerId);
  const commands = await service.listDispatchableExecutorCommands();
  assert.equal(attempts.length, 1);
  assert.equal(commands.length, 1);
  assert.equal(commands[0]?.command.idempotencyKey, `conversation.execution:${initial.executionId}`);
  assert.equal(commands[0]?.command.type, 'conversation.create');
});



test('initial dispatch replay returns the same execution identity after command lifecycle advances', async () => {
  const { service, scheduler, worker, initial, createCommand } = await fixture();
  await service.completeExecutorCommand(createCommand.commandId, 'executor-test', true, {});
  const replay = await scheduler.dispatchAttemptForReason(worker.workerId, 'initial');
  assert.equal(replay.executionId, initial.executionId);
  assert.equal((await service.listExecutionAttempts(worker.workerId)).filter((attempt) => attempt.reason === 'initial').length, 1);
});

test('idle auto policy schedules exactly one productive continuation and duplicate reconciliation is harmless', async () => {
  const { service, scheduler, worker } = await fixture('auto');
  await becomeIdle(service, worker.workerId);

  await Promise.all([scheduler.reconcile(), scheduler.reconcile(), scheduler.reconcile()]);
  await scheduler.reconcile();

  const attempts = await service.listExecutionAttempts(worker.workerId);
  assert.deepEqual(attempts.map((attempt) => attempt.reason), ['initial', 'auto_resume']);
  const commands = await service.listDispatchableExecutorCommands();
  assert.equal(commands.length, 2);
  assert.equal(commands.at(-1)?.command.type, 'conversation.send');
});

test('terminal, awaiting-human, awaiting-dependency, and paused workers suppress auto continuation', async (t) => {
  await t.test('terminal', async () => {
    const { service, scheduler, worker } = await fixture('auto');
    await becomeIdle(service, worker.workerId);
    await service.completeWorker(worker.workerId, 'done');
    await scheduler.reconcile();
    assert.equal((await service.listExecutionAttempts(worker.workerId)).length, 1);
  });

  await t.test('awaiting human', async () => {
    const { service, scheduler, worker } = await fixture('auto');
    await becomeIdle(service, worker.workerId);
    await service.awaitHuman(worker.workerId, { reason: 'approval', request: 'Approve deployment?' });
    await scheduler.reconcile();
    assert.equal((await service.listExecutionAttempts(worker.workerId)).length, 1);
  });

  await t.test('awaiting dependency', async () => {
    const { service, scheduler, worker } = await fixture('auto');
    await becomeIdle(service, worker.workerId);
    await service.awaitDependency(worker.workerId, { reason: 'CI is still running' });
    await scheduler.reconcile();
    assert.equal((await service.listExecutionAttempts(worker.workerId)).length, 1);
  });

  await t.test('paused', async () => {
    const { service, scheduler, worker } = await fixture('auto');
    await becomeIdle(service, worker.workerId);
    await service.pauseWorker(worker.workerId);
    await scheduler.reconcile();
    assert.equal((await service.listExecutionAttempts(worker.workerId)).length, 1);
  });
});

test('Continue Now is distinct from auto-resume and dispatches one manual attempt under manual policy', async () => {
  const { service, scheduler, worker } = await fixture('manual');
  await becomeIdle(service, worker.workerId);

  const result = await scheduler.continueNow(worker.workerId, { idempotencyKey: 'manual-once' });
  assert.equal(result.status, 'scheduled');
  assert.equal(result.attempt?.reason, 'manual_resume');
  assert.equal((await service.listExecutionAttempts(worker.workerId)).length, 2);
  assert.equal((await service.listDispatchableExecutorCommands()).length, 2);

  const duplicate = await scheduler.continueNow(worker.workerId, { idempotencyKey: 'manual-once' });
  assert.equal(duplicate.status, 'already_pending');
  assert.equal((await service.listExecutionAttempts(worker.workerId)).length, 2);
});

test('Continue Now during generation queues one post-turn resume and sends it only after idle', async () => {
  const { service, scheduler, worker } = await fixture('manual');
  const queued = await scheduler.continueNow(worker.workerId);
  assert.equal(queued.status, 'queued_after_current_turn');
  assert.equal((await service.listExecutionAttempts(worker.workerId)).length, 1);

  await becomeIdle(service, worker.workerId);
  await scheduler.reconcile();
  const attempts = await service.listExecutionAttempts(worker.workerId);
  assert.deepEqual(attempts.map((attempt) => attempt.reason), ['initial', 'manual_resume']);
});

test('executor-unavailable dispatch persists one pending command for later bridge delivery', async () => {
  const { service, scheduler, worker } = await fixture('auto');
  await becomeIdle(service, worker.workerId);
  await scheduler.reconcile();

  const commands = await service.listDispatchableExecutorCommands();
  assert.equal(commands.length, 2);
  const continuation = commands.at(-1);
  assert.equal(continuation?.status, 'pending');
  assert.equal(continuation?.command.type, 'conversation.send');

  await scheduler.reconcile();
  assert.equal((await service.listDispatchableExecutorCommands()).length, 2);
});

test('stall protection pauses repeated no-progress turns and explicit resume resets stall accounting', async () => {
  const { service, scheduler, run, worker } = await fixture('auto');
  await service.setRunContinuationPolicy(run.runId, {
    mode: 'auto',
    idleGraceMs: 0,
    stallWarningAfterNoProgressAttempts: 1,
    stallPauseAfterNoProgressAttempts: 2,
    stallWarningAfterNoProgressMs: null,
    stallPauseAfterNoProgressMs: null,
  });

  await becomeIdle(service, worker.workerId);
  let view = await service.getWorkerContinuity(worker.workerId);
  assert.equal(view.continuity.consecutiveExecutionAttemptsWithoutProgress, 1);
  assert.ok(view.continuity.stallWarningAt);
  assert.equal(view.continuity.stallPausedAt, null);

  await scheduler.reconcile();
  const auto = (await service.listExecutionAttempts(worker.workerId)).at(-1);
  assert.ok(auto && auto.reason === 'auto_resume');
  await deliverScheduled(service, scheduler, auto);
  await becomeIdle(service, worker.workerId);

  view = await service.getWorkerContinuity(worker.workerId);
  assert.equal(view.continuity.disposition, 'paused');
  assert.ok(view.continuity.stallPausedAt);
  assert.equal(view.continuity.consecutiveExecutionAttemptsWithoutProgress, 2);
  await scheduler.reconcile();
  assert.equal((await service.listExecutionAttempts(worker.workerId)).length, 2);

  view = await service.resumeWorker(worker.workerId);
  assert.equal(view.continuity.stallPausedAt, null);
  assert.equal(view.continuity.consecutiveExecutionAttemptsWithoutProgress, 0);
});


test('pause invalidates a scheduled unsubmitted continuation before dispatcher submission', async () => {
  const { service, scheduler, worker } = await fixture('auto');
  await becomeIdle(service, worker.workerId);
  const evaluation = await service.evaluateContinuation(worker.workerId);
  assert.equal(evaluation.action, 'schedule');
  assert.ok(evaluation.attempt);
  const pending = evaluation.attempt;

  await service.pauseWorker(worker.workerId);
  await scheduler.dispatchAttempt(pending);

  const attempts = await service.listExecutionAttempts(worker.workerId);
  const pausedAttempt = attempts.find((attempt) => attempt.executionId === pending.executionId);
  assert.notEqual(pausedAttempt?.state, 'scheduled');
  const commands = await service.listDispatchableExecutorCommands();
  assert.equal(commands.some((record) => record.command.idempotencyKey === `conversation.execution:${pending.executionId}`), false);
});


test('startup reconciliation reconstructs a missing executor command exactly once from a durable scheduled attempt', async () => {
  const service = new OrchestratorService(new InMemoryRepository());
  const { run, rootWorker } = await service.createRun({ title: 'restart recovery' });
  const { worker } = await service.spawnWorker({
    runId: run.runId,
    parentWorkerId: rootWorker.workerId,
    name: 'restart-child',
    assignment: { objective: 'survive backend restart before launch dispatch' },
  });
  assert.equal((await service.listDispatchableExecutorCommands()).length, 0);

  const firstScheduler = new ExecutionContinuityScheduler(service);
  await firstScheduler.reconcile();
  const firstCommands = await service.listDispatchableExecutorCommands();
  assert.equal(firstCommands.length, 1);
  assert.equal(firstCommands[0]?.command.workerId, worker.workerId);
  assert.equal(firstCommands[0]?.command.type, 'conversation.create');

  const restartedScheduler = new ExecutionContinuityScheduler(service);
  await restartedScheduler.reconcile();
  const afterRestart = await service.listDispatchableExecutorCommands();
  assert.equal(afterRestart.length, 1);
  assert.equal(afterRestart[0]?.command.commandId, firstCommands[0]?.command.commandId);
});

test('pause after command enqueue but before send invalidates both attempt and pending command', async () => {
  const { service, scheduler, worker } = await fixture('auto');
  await becomeIdle(service, worker.workerId);
  const evaluation = await service.evaluateContinuation(worker.workerId);
  assert.equal(evaluation.action, 'schedule');
  assert.ok(evaluation.attempt);
  await scheduler.dispatchAttempt(evaluation.attempt);
  const command = commandForExecutionAttempt(evaluation.attempt);
  assert.equal((await service.listDispatchableExecutorCommands()).some((record) => record.command.commandId === command.commandId), true);

  await service.pauseWorker(worker.workerId);
  const attempt = (await service.listExecutionAttempts(worker.workerId)).find((candidate) => candidate.executionId === evaluation.attempt?.executionId);
  assert.equal(attempt?.state, 'cancelled');
  assert.equal(attempt?.outcome, 'paused');
  assert.equal((await service.listDispatchableExecutorCommands()).some((record) => record.command.commandId === command.commandId), false);
});

test('terminal race after enqueue fails the atomic command-send claim instead of emitting a stale continuation', async () => {
  const { service, scheduler, worker } = await fixture('manual');
  await becomeIdle(service, worker.workerId);
  const result = await service.continueNow(worker.workerId, { idempotencyKey: 'terminal-race' });
  assert.ok(result.attempt);
  await scheduler.dispatchAttempt(result.attempt);
  const command = commandForExecutionAttempt(result.attempt);

  await service.interruptWorker(worker.workerId, 'cancel before websocket send');
  const claim = await service.markExecutorCommandSent(command.commandId, 'executor-test');
  assert.equal(claim.status, 'failed');
});


test('expired pre-submission executor command fails without side effect and clears the active attempt', async () => {
  const service = new OrchestratorService(new InMemoryRepository());
  const { run, rootWorker } = await service.createRun({ title: 'expired initial' });
  const { worker } = await service.spawnWorker({
    runId: run.runId,
    parentWorkerId: rootWorker.workerId,
    name: 'expired-child',
    assignment: { objective: 'never submit an expired launch' },
  });
  const attempt = (await service.listExecutionAttempts(worker.workerId))[0]!;
  const command = commandForExecutionAttempt(attempt, -1);
  await service.enqueueExecutorCommand(command);

  const expired = await service.expirePendingExecutorCommands(new Date().toISOString());
  assert.equal(expired.length, 1);
  assert.equal(expired[0]?.status, 'failed');
  assert.equal((await service.listDispatchableExecutorCommands()).length, 0);
  const attempts = await service.listExecutionAttempts(worker.workerId);
  assert.equal(attempts[0]?.state, 'failed');
  assert.equal(attempts[0]?.outcome, 'executor_unavailable');
  assert.equal((await service.getWorkerContinuity(worker.workerId)).continuity.pendingExecutionId, null);
  assert.equal((await service.getRun(run.runId)).state, 'active');
});


test('generating auto worker materializes a durable checkpoint and dispatches internal steering without waiting for idle', async () => {
  const { service, scheduler, rootWorker, worker } = await fixture('auto');
  const base = new Date();
  const scheduledAt = base.toISOString();
  await service.reconcileWakeIntents(scheduledAt, () => 0);
  const firstView = await service.getWorkerContinuity(worker.workerId);
  assert.ok(firstView.continuity.activeKeepaliveDueAt);
  const dueAt = firstView.continuity.activeKeepaliveDueAt!;
  assert.equal(Date.parse(dueAt) - Date.parse(scheduledAt), 20 * 60_000);

  await service.reconcileWakeIntents(dueAt, () => 0);
  const pending = (await service.listPendingMessageDeliveries(dueAt))
    .find((message) => message.fromWorkerId === rootWorker.workerId && message.toWorkerId === worker.workerId);
  assert.ok(pending);
  assert.equal(pending.delivery, 'steer_now');
  assert.equal(pending.priority, 'normal');
  assert.match(pending.body, /Orchestrator checkpoint/);
  assert.match(pending.body, /Then continue working on the same assignment to completion/);

  const dispatched = await scheduler.dispatchMessage(pending);
  assert.equal(dispatched.deliveryState, 'dispatching');
  assert.ok(dispatched.deliveryCommandId);
  const command = (await service.listDispatchableExecutorCommands())
    .find((record) => record.command.commandId === dispatched.deliveryCommandId)?.command;
  assert.equal(command?.type, 'conversation.steer');
});

test('successful routine steering starts the non-idle floor and an early normal steering message is durably deferred with notBefore', async () => {
  const { service, scheduler, run, rootWorker, worker } = await fixture('manual');
  await service.setRunContinuationPolicy(run.runId, { minContinuationSpacingMs: 15 * 60_000 });
  const first = await scheduler.sendMessage({
    runId: run.runId,
    fromWorkerId: rootWorker.workerId,
    toWorkerId: worker.workerId,
    type: 'information',
    body: 'first checkpoint',
    delivery: 'steer_now',
    priority: 'normal',
    idempotencyKey: 'routine-first',
  });
  assert.ok(first.deliveryCommandId);
  await service.completeExecutorCommand(first.deliveryCommandId!, 'executor-test', true, { steeringAccepted: true });
  const delivered = await service.getMessageDeliveryContext(first.messageId);
  assert.equal(delivered.message.deliveryState, 'delivered');
  const floorAnchor = delivered.continuity.continuity.lastRoutineSteerAt;
  assert.ok(floorAnchor);

  const second = await scheduler.sendMessage({
    runId: run.runId,
    fromWorkerId: rootWorker.workerId,
    toWorkerId: worker.workerId,
    type: 'information',
    body: 'second checkpoint',
    delivery: 'steer_now',
    priority: 'normal',
    idempotencyKey: 'routine-second',
  });
  assert.equal(second.deliveryState, 'retry_wait');
  assert.equal(second.deliveryCommandId, null);
  assert.ok(second.notBefore);
  assert.equal(Date.parse(second.notBefore!), Date.parse(floorAnchor!) + 15 * 60_000);
});

test('idle steer_now bypasses the non-idle floor and routes through ordinary conversation send', async () => {
  const { service, scheduler, run, rootWorker, worker } = await fixture('manual');
  const first = await scheduler.sendMessage({
    runId: run.runId,
    fromWorkerId: rootWorker.workerId,
    toWorkerId: worker.workerId,
    type: 'information',
    body: 'checkpoint while generating',
    delivery: 'steer_now',
    priority: 'normal',
    idempotencyKey: 'idle-bypass-anchor',
  });
  assert.ok(first.deliveryCommandId);
  await service.completeExecutorCommand(first.deliveryCommandId!, 'executor-test', true, { steeringAccepted: true });
  await becomeIdle(service, worker.workerId);

  const next = await scheduler.sendMessage({
    runId: run.runId,
    fromWorkerId: rootWorker.workerId,
    toWorkerId: worker.workerId,
    type: 'information',
    body: 'deliver while idle',
    delivery: 'steer_now',
    priority: 'normal',
    idempotencyKey: 'idle-bypass-next',
  });
  assert.equal(next.deliveryState, 'dispatching');
  const command = (await service.listDispatchableExecutorCommands())
    .find((record) => record.command.commandId === next.deliveryCommandId)?.command;
  assert.equal(command?.type, 'conversation.send');
});

test('corrective steer_now bypasses the ordinary non-idle spacing floor', async () => {
  const { service, scheduler, run, rootWorker, worker } = await fixture('manual');
  const first = await scheduler.sendMessage({
    runId: run.runId,
    fromWorkerId: rootWorker.workerId,
    toWorkerId: worker.workerId,
    type: 'information',
    body: 'routine anchor',
    delivery: 'steer_now',
    priority: 'normal',
    idempotencyKey: 'corrective-anchor',
  });
  assert.ok(first.deliveryCommandId);
  await service.completeExecutorCommand(first.deliveryCommandId!, 'executor-test', true, { steeringAccepted: true });

  const correction = await scheduler.sendMessage({
    runId: run.runId,
    fromWorkerId: rootWorker.workerId,
    toWorkerId: worker.workerId,
    type: 'correction',
    body: 'stop using the stale contract and converge on the shared protocol',
    delivery: 'steer_now',
    priority: 'corrective',
    idempotencyKey: 'corrective-bypass',
  });
  assert.equal(correction.deliveryState, 'dispatching');
  const command = (await service.listDispatchableExecutorCommands())
    .find((record) => record.command.commandId === correction.deliveryCommandId)?.command;
  assert.equal(command?.type, 'conversation.steer');
  if (command?.type === 'conversation.steer') assert.equal(command.reason, 'corrective');
});

test('temporarily unavailable live delivery preserves one durable message intent for retry', async () => {
  const { service, scheduler, run, rootWorker, worker } = await fixture('manual');
  const message = await scheduler.sendMessage({
    runId: run.runId,
    fromWorkerId: worker.workerId,
    toWorkerId: rootWorker.workerId,
    type: 'information',
    body: 'root update while root binding is unavailable',
    delivery: 'steer_now',
    priority: 'normal',
    idempotencyKey: 'unavailable-root',
  });
  assert.equal(message.deliveryState, 'retry_wait');
  assert.equal(message.deliveryCommandId, null);
  assert.ok(message.notBefore);
  const replay = await service.sendMessage({
    runId: run.runId,
    fromWorkerId: worker.workerId,
    toWorkerId: rootWorker.workerId,
    type: 'information',
    body: 'root update while root binding is unavailable',
    delivery: 'steer_now',
    priority: 'normal',
    idempotencyKey: 'unavailable-root',
  });
  assert.equal(replay.messageId, message.messageId);
});

test('worker progress creates a durable next-turn root wake message', async () => {
  const { service, run, rootWorker, worker } = await fixture('manual');
  const event = await service.progress(worker.workerId, 'checkpoint commit abc123; implementation continues');
  const pending = (await service.listPendingMessageDeliveries())
    .find((message) => message.fromWorkerId === worker.workerId && message.toWorkerId === rootWorker.workerId && message.body.includes('abc123'));
  assert.ok(pending);
  assert.equal(pending.delivery, 'next_turn');
  assert.equal(pending.priority, 'normal');
  assert.equal(pending.deliveryState, 'pending');
  const durableEvents = await service.listEvents(run.runId, event.cursor - 1);
  assert.ok(durableEvents.some((candidate) => candidate.type === 'worker.message' && candidate.workerId === rootWorker.workerId));
});

test('message send is idempotent by key and conflicting reuse fails closed', async () => {
  const { service, run, rootWorker, worker } = await fixture('manual');
  const input = {
    runId: run.runId,
    fromWorkerId: rootWorker.workerId,
    toWorkerId: worker.workerId,
    type: 'information' as const,
    body: 'same durable message',
    delivery: 'record_only' as const,
    priority: 'normal' as const,
    idempotencyKey: 'message-idempotency',
  };
  const first = await service.sendMessage(input);
  const replay = await service.sendMessage(input);
  assert.equal(replay.messageId, first.messageId);
  await assert.rejects(service.sendMessage({ ...input, body: 'conflicting body' }), /idempotency key reused/i);
});

test('ambiguous live steering is terminally recorded and is never silently retried', async () => {
  const { service, scheduler, run, rootWorker, worker } = await fixture('manual');
  const message = await scheduler.sendMessage({
    runId: run.runId,
    fromWorkerId: rootWorker.workerId,
    toWorkerId: worker.workerId,
    type: 'information',
    body: 'ambiguous checkpoint',
    delivery: 'steer_now',
    priority: 'normal',
    idempotencyKey: 'ambiguous-steer',
  });
  assert.ok(message.deliveryCommandId);
  await service.completeExecutorCommand(message.deliveryCommandId!, 'executor-test', false, {}, 'AMBIGUOUS_SUBMISSION: unable to prove acceptance');
  const after = await service.getMessageDeliveryContext(message.messageId);
  assert.equal(after.message.deliveryState, 'ambiguous');
  assert.equal(after.message.notBefore, null);
  assert.equal((await service.listPendingMessageDeliveries()).some((candidate) => candidate.messageId === message.messageId), false);
});


test('dead-man fallback is inert while disabled and while any managed conversation remains alive', async () => {
  const { service, run, worker } = await fixture('auto');
  await becomeIdle(service, worker.workerId);
  await service.setRunContinuationPolicy(run.runId, { deadmanFallbackEnabled: false, deadmanThresholdMs: 1_000 });
  const live = await service.getConversationBinding(worker.workerId);
  assert.ok(live);
  const observedAt = new Date(Date.parse(live!.updatedAt) + 5_000).toISOString();
  const disabled = await service.requestDeadmanRecovery(run.runId, observedAt);
  assert.equal(disabled.fired, false);
  assert.equal(disabled.reason, 'managed_conversation_alive');

  await service.recordConversationState({ workerId: worker.workerId, executorId: 'executor-test', state: 'closed' });
  const closed = await service.getConversationBinding(worker.workerId);
  assert.ok(closed);
  const afterThreshold = new Date(Date.parse(closed!.updatedAt) + 5_000).toISOString();
  const stillDisabled = await service.requestDeadmanRecovery(run.runId, afterThreshold);
  assert.equal(stillDisabled.fired, false);
  assert.equal(stillDisabled.reason, 'fallback_disabled');
  assert.equal((await service.listExecutionAttempts(worker.workerId)).filter((attempt) => attempt.reason === 'recovery').length, 0);
});

test('dead-man fallback authorizes one idempotent recovery for an all-dead stale run and records durable evidence', async () => {
  const { service, scheduler, run, worker } = await fixture('auto');
  await becomeIdle(service, worker.workerId);
  await service.setRunContinuationPolicy(run.runId, { deadmanFallbackEnabled: true, deadmanThresholdMs: 1_000 });
  await service.recordConversationState({ workerId: worker.workerId, executorId: 'executor-test', state: 'closed' });
  const closed = await service.getConversationBinding(worker.workerId);
  assert.ok(closed);
  const observedAt = new Date(Date.parse(closed!.updatedAt) + 5_000).toISOString();

  const first = await service.requestDeadmanRecovery(run.runId, observedAt);
  assert.equal(first.fired, true);
  assert.equal(first.reason, 'authorized');
  assert.equal(first.attempts.length, 1);
  assert.equal(first.attempts[0]?.reason, 'recovery');
  const continuity = await service.getWorkerContinuity(worker.workerId);
  assert.equal(continuity.continuity.deadmanRecoveryCount, 1);
  assert.equal(continuity.continuity.lastDeadmanRecoveryAt, observedAt);

  const duplicate = await service.requestDeadmanRecovery(run.runId, observedAt);
  assert.equal(duplicate.fired, false);
  assert.equal(duplicate.reason, 'recovery_already_pending');
  assert.equal((await service.listExecutionAttempts(worker.workerId)).filter((attempt) => attempt.reason === 'recovery').length, 1);

  await scheduler.dispatchAttempt(first.attempts[0]!);
  const expected = commandForExecutionAttempt(first.attempts[0]!, undefined, { recoveryRequiresCreate: true });
  const dispatched = (await service.listDispatchableExecutorCommands()).find((record) => record.command.commandId === expected.commandId)?.command;
  assert.equal(dispatched?.type, 'conversation.create');
  if (dispatched?.type === 'conversation.create') {
    assert.match(dispatched.bootstrap, new RegExp(worker.workerId));
    assert.notEqual(dispatched.bootstrap.trim(), 'continue');
  }

  const events = await service.listEvents(run.runId);
  assert.ok(events.some((event) => event.type === 'continuation.deadman_recovery_requested'));
  assert.equal(events.filter((event) => event.type === 'continuation.deadman_recovery_fired').length, 1);
});

test('dead-man fallback waits for the backend threshold before authorizing recovery', async () => {
  const { service, run, worker } = await fixture('auto');
  await becomeIdle(service, worker.workerId);
  await service.setRunContinuationPolicy(run.runId, { deadmanFallbackEnabled: true, deadmanThresholdMs: 60_000 });
  await service.recordConversationState({ workerId: worker.workerId, executorId: 'executor-test', state: 'error', error: 'synthetic dead browser' });
  const dead = await service.getConversationBinding(worker.workerId);
  assert.ok(dead);
  const observedAt = new Date(Date.parse(dead!.updatedAt) + 30_000).toISOString();
  const result = await service.requestDeadmanRecovery(run.runId, observedAt);
  assert.equal(result.fired, false);
  assert.equal(result.reason, 'threshold_not_reached');
  assert.equal(result.attempts.length, 0);
});