import { createHash } from 'node:crypto';
import {
  ContinuationPolicySchema,
  DEFAULT_CONTINUATION_POLICY,
  type Assignment,
  type CommandId,
  type ContinuationPolicy,
  type ConversationBinding,
  type ConversationState,
  type EventType,
  type ExecutionAttempt,
  type ExecutionAttemptOutcome,
  type ExecutionAttemptReason,
  type ExecutionId,
  type ExecutorCommand,
  type ExecutorCommandRecord,
  type Message,
  type MessageDelivery,
  type MessagePriority,
  type OrchestratorErrorCode,
  type OrchestratorEvent,
  type Run,
  type RunId,
  type Worker,
  type WorkerContinuityState,
  type WorkerId,
} from '@platform-modules/chatgpt-orchestrator-protocol';
import {
  defaultWorkerContinuity,
  effectiveContinuationPolicy,
  pruneTerminalRuns as pruneStoredTerminalRuns,
  type OrchestratorRepository,
  type StoredState,
} from '@platform-modules/chatgpt-orchestrator-persistence';

export class OrchestratorError extends Error {
  readonly code: OrchestratorErrorCode;
  constructor(code: OrchestratorErrorCode, message: string) {
    super(message);
    this.name = 'OrchestratorError';
    this.code = code;
  }
}

type MessageType = import('@platform-modules/chatgpt-orchestrator-protocol').Message['type'];

export interface AssignmentInput {
  objective: string;
  constraints?: string[];
  acceptanceCriteria?: string[];
  dependencies?: WorkerId[];
  allowedScope?: string[];
  forbiddenScope?: string[];
  artifacts?: string[];
}

export interface WaitResult {
  cursor: number;
  events: OrchestratorEvent[];
  timedOut: boolean;
}

export interface WorkerContinuityView {
  continuity: WorkerContinuityState;
  effectivePolicy: ContinuationPolicy;
  policySource: 'worker' | 'run' | 'default';
  latestAttempt: ExecutionAttempt | null;
}

export interface ContinuationEvaluation {
  action: 'stop' | 'wait' | 'schedule';
  reason: string;
  attempt: ExecutionAttempt | null;
}

export interface ContinueNowResult {
  status: 'scheduled' | 'queued_after_current_turn' | 'already_pending';
  attempt: ExecutionAttempt | null;
}

const TERMINAL_WORKER_STATES = new Set(['completed', 'failed', 'cancelled']);
const ACTIVE_ATTEMPT_STATES = new Set(['scheduled', 'submitted', 'generating']);
const ORCHESTRATOR_CHECKPOINT_BODY = `Orchestrator checkpoint.

Send the root conversation a durable update about this assignment now. Include:
- latest checkpoint commit(s), if any;
- what has been completed since the previous report;
- what you are working on now;
- what remains before the full assignment is complete;
- blockers, risks, dependency changes, or cross-lane contract changes.

Then continue working on the same assignment to completion.
Do not stop merely because you sent the update.
If fully complete, call worker.complete with the terminal handoff.
If genuinely blocked, report the blocker and enter the correct durable blocking state.`;

function isTerminalWorker(worker: Worker): boolean {
  return TERMINAL_WORKER_STATES.has(worker.state);
}

function executionFingerprint(input: { workerId: WorkerId; reason: ExecutionAttemptReason; resumeOfExecutionId: ExecutionId | null }): string {
  return fingerprint(input);
}

function id(prefix: string): string { return `${prefix}_${crypto.randomUUID()}`; }
function now(): string { return new Date().toISOString(); }
function fingerprint(value: unknown): string {
  return createHash('sha256').update(JSON.stringify(value)).digest('hex');
}

function summarizeObjective(objective: string, maxLength = 220): string {
  const compact = objective.replace(/\s+/g, ' ').trim();
  if (compact.length <= maxLength) return compact;
  return `${compact.slice(0, maxLength - 1).trimEnd()}…`;
}

function inferredLane(name: string): string | null {
  const token = name.split(/[^a-zA-Z0-9]+/).find((part) => /^[a-zA-Z]$/.test(part));
  return token ? token.toUpperCase() : null;
}

function defaultWorkerDisplayName(taskName: string, internalName: string): string {
  const lane = inferredLane(internalName);
  return lane ? `${taskName} Lane ${lane}` : `${taskName} — ${internalName}`;
}

class EventNotifier {
  private readonly waiters = new Map<string, Set<() => void>>();

  notify(runId: string): void {
    const set = this.waiters.get(runId);
    if (!set) return;
    this.waiters.delete(runId);
    for (const wake of set) wake();
  }

  async wait(runId: string, timeoutMs: number): Promise<void> {
    await new Promise<void>((resolve) => {
      const set = this.waiters.get(runId) ?? new Set<() => void>();
      let timer: NodeJS.Timeout;
      const wake = () => {
        clearTimeout(timer);
        set.delete(wake);
        if (set.size === 0) this.waiters.delete(runId);
        resolve();
      };
      set.add(wake);
      this.waiters.set(runId, set);
      timer = setTimeout(wake, timeoutMs);
    });
  }
}

export class OrchestratorService {
  private readonly repository: OrchestratorRepository;
  private readonly notifier = new EventNotifier();

  constructor(repository: OrchestratorRepository) {
    this.repository = repository;
  }

  private workerOrThrow(state: Readonly<StoredState>, workerId: WorkerId): Worker {
    const worker = state.workers[workerId];
    if (!worker) throw new OrchestratorError('NOT_FOUND', `Worker ${workerId} not found.`);
    return worker;
  }

  private continuityFor(state: StoredState, worker: Worker): WorkerContinuityState {
    const existing = state.workerContinuity[worker.workerId];
    if (existing) return existing;
    const created = defaultWorkerContinuity(worker);
    state.workerContinuity[worker.workerId] = created;
    return created;
  }

  private clearWakeIntentInState(state: StoredState, worker: Worker, reason: string, timestamp: string): void {
    const continuity = this.continuityFor(state, worker);
    if (continuity.pendingWakeReasons.length === 0 && !continuity.wakeNotBefore) return;
    const pendingWakeReasons = [...continuity.pendingWakeReasons];
    continuity.pendingWakeReasons = [];
    continuity.wakeNotBefore = null;
    continuity.updatedAt = timestamp;
    this.appendEvent(state, {
      runId: worker.runId,
      type: 'continuation.wake_cancelled',
      workerId: worker.workerId,
      payload: { reason, pendingWakeReasons },
    });
  }

  private ensureKeepaliveInState(state: StoredState, worker: Worker, at: string, random = Math.random): boolean {
    const continuity = this.continuityFor(state, worker);
    const policy = effectiveContinuationPolicy(state, worker.workerId);
    const run = state.runs[worker.runId];
    const runnable = run?.state === 'active'
      && worker.workerId !== run.rootWorkerId
      && !isTerminalWorker(worker)
      && !['awaiting_human', 'awaiting_dependency', 'paused', 'terminal'].includes(continuity.disposition)
      && policy.mode === 'auto';
    if (!runnable) {
      const changed = continuity.activeKeepaliveDueAt !== null;
      continuity.activeKeepaliveDueAt = null;
      if (changed) continuity.updatedAt = at;
      return changed;
    }
    if (continuity.activeKeepaliveDueAt || continuity.pendingWakeReasons.includes('active_keepalive')) return false;
    const span = Math.max(0, policy.activeKeepaliveMaxMs - policy.activeKeepaliveMinMs);
    const jitter = policy.activeKeepaliveMinMs + Math.floor(Math.min(0.999999999, Math.max(0, random())) * (span + 1));
    continuity.activeKeepaliveDueAt = new Date(Date.parse(at) + jitter).toISOString();
    continuity.updatedAt = at;
    this.appendEvent(state, {
      runId: worker.runId,
      type: 'continuation.keepalive_scheduled',
      workerId: worker.workerId,
      payload: { dueAt: continuity.activeKeepaliveDueAt, minMs: policy.activeKeepaliveMinMs, maxMs: policy.activeKeepaliveMaxMs },
    });
    return true;
  }

  private queueWakeInState(state: StoredState, worker: Worker, reason: string, at: string, allowManual = false): boolean {
    const run = state.runs[worker.runId];
    const continuity = this.continuityFor(state, worker);
    const policy = effectiveContinuationPolicy(state, worker.workerId);
    if (!run || run.state !== 'active' || isTerminalWorker(worker)) return false;
    if (['awaiting_human', 'paused', 'terminal'].includes(continuity.disposition)) return false;
    if (policy.mode === 'manual' && !allowManual) return false;
    const floorAt = continuity.lastContinuationSubmittedAt
      ? new Date(Date.parse(continuity.lastContinuationSubmittedAt) + policy.minContinuationSpacingMs).toISOString()
      : at;
    const candidateNotBefore = Date.parse(floorAt) > Date.parse(at) ? floorAt : at;
    const nextNotBefore = continuity.wakeNotBefore && Date.parse(continuity.wakeNotBefore) > Date.parse(candidateNotBefore)
      ? continuity.wakeNotBefore
      : candidateNotBefore;
    const reasonAdded = !continuity.pendingWakeReasons.includes(reason);
    const notBeforeChanged = continuity.wakeNotBefore !== nextNotBefore;
    if (!reasonAdded && !notBeforeChanged) return false;
    if (reasonAdded) continuity.pendingWakeReasons = [...continuity.pendingWakeReasons, reason].slice(-32);
    continuity.wakeNotBefore = nextNotBefore;
    continuity.updatedAt = at;
    this.appendEvent(state, {
      runId: worker.runId,
      type: 'continuation.wake_queued',
      workerId: worker.workerId,
      payload: { reason, reasons: continuity.pendingWakeReasons, notBefore: continuity.wakeNotBefore },
    });
    return true;
  }

  private createMessageInState(state: StoredState, input: {
    runId: RunId;
    fromWorkerId: WorkerId;
    toWorkerId: WorkerId;
    type: MessageType;
    body: string;
    delivery: MessageDelivery;
    priority: MessagePriority;
    idempotencyKey?: string;
    notBefore?: string | null;
  }, createdAt = now()): { message: Message; changed: boolean } {
    const from = state.workers[input.fromWorkerId];
    const to = state.workers[input.toWorkerId];
    if (!from || !to) throw new OrchestratorError('NOT_FOUND', 'Message sender or recipient does not exist.');
    if (from.runId !== input.runId || to.runId !== input.runId) {
      throw new OrchestratorError('WRONG_RUN', 'Message sender and recipient must belong to the run.');
    }
    if (input.delivery === 'steer_now' && !input.idempotencyKey) {
      throw new OrchestratorError('INVALID_STATE', 'steer_now message delivery requires an idempotencyKey.');
    }
    const key = input.idempotencyKey ? `message.send:${input.runId}:${input.idempotencyKey}` : undefined;
    const fp = fingerprint({
      fromWorkerId: input.fromWorkerId,
      toWorkerId: input.toWorkerId,
      type: input.type,
      body: input.body,
      delivery: input.delivery,
      priority: input.priority,
    });
    if (key && state.idempotency[key]) {
      const prior = state.idempotency[key];
      if (prior.fingerprint !== fp) throw new OrchestratorError('IDEMPOTENCY_CONFLICT', 'Message idempotency key reused with different message input.');
      const existing = state.messages[prior.resourceId];
      if (!existing) throw new OrchestratorError('INTERNAL_ERROR', 'Idempotent message record is missing.');
      return { message: existing, changed: false };
    }

    const deliveryState = input.delivery === 'record_only' ? 'recorded' as const : 'pending' as const;
    const message: Message = {
      messageId: id('msg'),
      runId: input.runId,
      fromWorkerId: input.fromWorkerId,
      toWorkerId: input.toWorkerId,
      type: input.type,
      body: input.body,
      delivery: input.delivery,
      priority: input.priority,
      idempotencyKey: input.idempotencyKey ?? null,
      deliveryState,
      notBefore: input.delivery === 'record_only' ? null : input.notBefore ?? null,
      deliveryCommandId: null,
      deliveryAttempts: 0,
      deliveredAt: null,
      lastDeliveryError: null,
      createdAt,
    };
    state.messages[message.messageId] = message;
    this.appendEvent(state, {
      runId: input.runId,
      type: 'worker.message',
      workerId: input.toWorkerId,
      payload: {
        messageId: message.messageId,
        fromWorkerId: input.fromWorkerId,
        type: input.type,
        body: input.body,
        delivery: input.delivery,
        priority: input.priority,
        notBefore: message.notBefore,
      },
    });
    if (message.notBefore && Date.parse(message.notBefore) > Date.parse(createdAt)) {
      this.appendEvent(state, {
        runId: input.runId,
        type: 'message.delivery_deferred',
        workerId: input.toWorkerId,
        payload: { messageId: message.messageId, notBefore: message.notBefore, reason: 'initial_not_before' },
      });
    }
    if (key) state.idempotency[key] = { resourceId: message.messageId, fingerprint: fp };
    return { message, changed: true };
  }

  private settleMessageDeliveryInState(
    state: StoredState,
    commandId: CommandId,
    success: boolean,
    timestamp: string,
    error: string | null,
  ): Message | null {
    const message = Object.values(state.messages).find((candidate) => candidate.deliveryCommandId === commandId);
    if (!message || ['delivered', 'ambiguous'].includes(message.deliveryState)) return message ?? null;
    if (success) {
      const delivered: Message = {
        ...message,
        deliveryState: 'delivered',
        deliveredAt: timestamp,
        notBefore: null,
        lastDeliveryError: null,
      };
      state.messages[message.messageId] = delivered;
      const command = state.executorCommands[commandId]?.command;
      if (message.delivery === 'steer_now' && message.priority === 'normal' && command?.type === 'conversation.steer') {
        const worker = state.workers[message.toWorkerId];
        if (worker) {
          const continuity = this.continuityFor(state, worker);
          continuity.lastRoutineSteerAt = timestamp;
          continuity.updatedAt = timestamp;
        }
      }
      this.appendEvent(state, {
        runId: message.runId,
        type: 'message.delivery_delivered',
        workerId: message.toWorkerId,
        payload: { messageId: message.messageId, commandId, deliveredAt: timestamp },
      });
      return delivered;
    }

    const ambiguous = Boolean(error && /AMBIGUOUS_SUBMISSION/i.test(error));
    const next: Message = {
      ...message,
      deliveryState: ambiguous ? 'ambiguous' : 'retry_wait',
      notBefore: ambiguous ? null : new Date(Date.parse(timestamp) + 5_000).toISOString(),
      lastDeliveryError: error ?? 'Executor delivery failed.',
    };
    state.messages[message.messageId] = next;
    this.appendEvent(state, {
      runId: message.runId,
      type: ambiguous ? 'message.delivery_ambiguous' : 'message.delivery_failed',
      workerId: message.toWorkerId,
      payload: {
        messageId: message.messageId,
        commandId,
        error: next.lastDeliveryError,
        retryNotBefore: next.notBefore,
      },
    });
    return next;
  }

  private latestAttemptInState(state: Readonly<StoredState>, workerId: WorkerId): ExecutionAttempt | null {
    const attempts = Object.values(state.executionAttempts)
      .filter((attempt) => attempt.workerId === workerId)
      .sort((a, b) => b.sequence - a.sequence);
    return attempts[0] ?? null;
  }

  private activeAttemptInState(state: Readonly<StoredState>, workerId: WorkerId): ExecutionAttempt | null {
    return Object.values(state.executionAttempts)
      .filter((attempt) => attempt.workerId === workerId && ACTIVE_ATTEMPT_STATES.has(attempt.state))
      .sort((a, b) => b.sequence - a.sequence)[0] ?? null;
  }

  private scheduleAttemptInState(
    state: StoredState,
    worker: Worker,
    reason: ExecutionAttemptReason,
    idempotencyKey: string,
    resumeOfExecutionId: ExecutionId | null,
    scheduledAt = now(),
    wakeReasons: string[] = [],
  ): { attempt: ExecutionAttempt; changed: boolean } {
    if (isTerminalWorker(worker)) throw new OrchestratorError('INVALID_STATE', `Worker cannot schedule execution from ${worker.state}.`);
    const run = state.runs[worker.runId];
    if (!run || run.state !== 'active') throw new OrchestratorError('INVALID_STATE', 'Execution may only be scheduled for an active run.');
    const fp = executionFingerprint({ workerId: worker.workerId, reason, resumeOfExecutionId });
    const priorIdempotency = state.executionIdempotency[idempotencyKey];
    if (priorIdempotency) {
      if (priorIdempotency.fingerprint !== fp) {
        throw new OrchestratorError('IDEMPOTENCY_CONFLICT', 'Execution idempotency key reused with different execution input.');
      }
      const prior = state.executionAttempts[priorIdempotency.executionId];
      if (!prior) throw new OrchestratorError('INTERNAL_ERROR', 'Idempotent execution attempt is missing.');
      return { attempt: prior, changed: false };
    }
    const active = this.activeAttemptInState(state, worker.workerId);
    if (active) throw new OrchestratorError('ALREADY_RUNNING', `Worker already has execution ${active.executionId} in state ${active.state}.`);

    const sequence = (state.executionSequences[worker.workerId] ?? 0) + 1;
    const executionId = id('exe') as ExecutionId;
    const continuity = this.continuityFor(state, worker);
    const attempt: ExecutionAttempt = {
      executionId,
      runId: worker.runId,
      workerId: worker.workerId,
      conversationId: state.conversations[worker.workerId]?.conversationId ?? null,
      sequence,
      reason,
      resumeOfExecutionId,
      state: 'scheduled',
      scheduledAt,
      submittedAt: null,
      generatingAt: null,
      idleAt: null,
      terminalAt: null,
      progressCursorAtStart: continuity.lastProgressCursor,
      lastProgressCursor: continuity.lastProgressCursor,
      wakeReasons,
      continuationCommandId: null,
      idempotencyKey,
      outcome: null,
      errorCode: null,
      errorDetail: null,
    };
    state.executionAttempts[executionId] = attempt;
    state.executionSequences[worker.workerId] = sequence;
    state.executionIdempotency[idempotencyKey] = { executionId, fingerprint: fp };
    state.workers[worker.workerId] = { ...worker, executionId, updatedAt: scheduledAt };
    continuity.pendingExecutionId = executionId;
    if (wakeReasons.length > 0) {
      continuity.pendingWakeReasons = [];
      continuity.wakeNotBefore = null;
    }
    continuity.disposition = 'awaiting_execution';
    continuity.idleGraceDueAt = null;
    continuity.updatedAt = scheduledAt;
    this.appendEvent(state, {
      runId: worker.runId,
      type: 'execution.scheduled',
      workerId: worker.workerId,
      payload: { executionId, sequence, reason, resumeOfExecutionId, wakeReasons },
    });
    if (['auto_resume', 'manual_resume', 'recovery'].includes(reason)) {
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'execution.resume_scheduled',
        workerId: worker.workerId,
        payload: { executionId, sequence, reason, resumeOfExecutionId, wakeReasons },
      });
    }
    return { attempt, changed: true };
  }

  private linkPendingAttemptToCommand(state: StoredState, command: ExecutorCommand): void {
    const attempt = this.activeAttemptInState(state, command.workerId);
    if (!attempt || attempt.state !== 'scheduled' || attempt.continuationCommandId) return;
    if (command.type === 'conversation.create' && attempt.reason !== 'initial') return;
    if (command.type === 'conversation.send' && attempt.reason === 'initial') return;
    state.executionAttempts[attempt.executionId] = { ...attempt, continuationCommandId: command.commandId };
  }

  private markAttemptSubmittedInState(
    state: StoredState,
    executionId: ExecutionId,
    commandId?: CommandId,
    conversationId?: string,
    timestamp = now(),
  ): ExecutionAttempt {
    const attempt = state.executionAttempts[executionId];
    if (!attempt) throw new OrchestratorError('NOT_FOUND', `Execution attempt ${executionId} not found.`);
    if (['completed', 'blocked', 'failed', 'cancelled'].includes(attempt.state)) return attempt;
    if (attempt.state === 'submitted' && !commandId && !conversationId) return attempt;
    const next: ExecutionAttempt = {
      ...attempt,
      state: attempt.state === 'generating' || attempt.state === 'idle' ? attempt.state : 'submitted',
      submittedAt: attempt.submittedAt ?? timestamp,
      continuationCommandId: commandId ?? attempt.continuationCommandId,
      conversationId: conversationId ?? attempt.conversationId,
    };
    state.executionAttempts[executionId] = next;
    if (!attempt.submittedAt) {
      const worker = state.workers[attempt.workerId];
      if (worker && ['auto_resume', 'manual_resume', 'recovery'].includes(attempt.reason)) {
        const continuity = this.continuityFor(state, worker);
        continuity.lastContinuationSubmittedAt = timestamp;
        continuity.pendingWakeReasons = [];
        continuity.wakeNotBefore = null;
        continuity.activeKeepaliveDueAt = null;
        continuity.updatedAt = timestamp;
        this.ensureKeepaliveInState(state, worker, timestamp);
      }
      this.appendEvent(state, {
        runId: attempt.runId,
        type: 'execution.submitted',
        workerId: attempt.workerId,
        payload: { executionId, sequence: attempt.sequence, reason: attempt.reason, commandId: next.continuationCommandId },
      });
      if (attempt.reason !== 'initial' && attempt.reason !== 'followup') {
        this.appendEvent(state, {
          runId: attempt.runId,
          type: 'execution.resumed',
          workerId: attempt.workerId,
          payload: { executionId, sequence: attempt.sequence, reason: attempt.reason },
        });
      }
    }
    return next;
  }

  private markAttemptGeneratingInState(state: StoredState, executionId: ExecutionId, timestamp = now()): ExecutionAttempt {
    const attempt = state.executionAttempts[executionId];
    if (!attempt) throw new OrchestratorError('NOT_FOUND', `Execution attempt ${executionId} not found.`);
    if (['completed', 'blocked', 'failed', 'cancelled', 'idle'].includes(attempt.state)) return attempt;
    if (attempt.state === 'generating') return attempt;
    const next: ExecutionAttempt = { ...attempt, state: 'generating', generatingAt: attempt.generatingAt ?? timestamp };
    state.executionAttempts[executionId] = next;
    const worker = state.workers[attempt.workerId];
    if (worker) {
      const continuity = this.continuityFor(state, worker);
      if (!['awaiting_human', 'awaiting_dependency', 'paused', 'terminal'].includes(continuity.disposition)) {
        continuity.disposition = 'running';
      }
      continuity.pendingExecutionId = executionId;
      continuity.idleGraceDueAt = null;
      continuity.updatedAt = timestamp;
    }
    this.appendEvent(state, {
      runId: attempt.runId,
      type: 'execution.generating',
      workerId: attempt.workerId,
      payload: { executionId, sequence: attempt.sequence, reason: attempt.reason },
    });
    return next;
  }

  private finishAttemptInState(
    state: StoredState,
    worker: Worker,
    attemptState: 'completed' | 'blocked' | 'failed' | 'cancelled',
    outcome: ExecutionAttemptOutcome,
    timestamp: string,
    errorCode: string | null = null,
    errorDetail: string | null = null,
  ): ExecutionAttempt | null {
    const attempt = this.latestAttemptInState(state, worker.workerId);
    if (!attempt || ['completed', 'blocked', 'failed', 'cancelled'].includes(attempt.state)) return attempt;
    const next: ExecutionAttempt = {
      ...attempt,
      state: attemptState,
      terminalAt: timestamp,
      outcome,
      errorCode,
      errorDetail,
    };
    state.executionAttempts[attempt.executionId] = next;
    const continuity = this.continuityFor(state, worker);
    if (continuity.pendingExecutionId === attempt.executionId) continuity.pendingExecutionId = null;
    const eventType: EventType = attemptState === 'completed'
      ? 'execution.completed'
      : attemptState === 'blocked'
        ? 'execution.blocked'
        : 'execution.failed';
    this.appendEvent(state, {
      runId: worker.runId,
      type: eventType,
      workerId: worker.workerId,
      payload: { executionId: attempt.executionId, sequence: attempt.sequence, outcome, errorCode, errorDetail },
    });
    return next;
  }

  private applyStallPolicyInState(state: StoredState, worker: Worker, attempt: ExecutionAttempt, timestamp: string): void {
    const continuity = this.continuityFor(state, worker);
    const policy = effectiveContinuationPolicy(state, worker.workerId);
    const noProgressAttempts = continuity.consecutiveExecutionAttemptsWithoutProgress;
    const baselineMs = Date.parse(continuity.lastProgressAt ?? attempt.scheduledAt);
    const noProgressMs = Math.max(0, Date.parse(timestamp) - baselineMs);
    const warningByAttempts = policy.stallWarningAfterNoProgressAttempts !== null
      && noProgressAttempts >= policy.stallWarningAfterNoProgressAttempts;
    const warningByTime = policy.stallWarningAfterNoProgressMs !== null
      && noProgressMs >= policy.stallWarningAfterNoProgressMs;
    const pauseByAttempts = policy.stallPauseAfterNoProgressAttempts !== null
      && noProgressAttempts >= policy.stallPauseAfterNoProgressAttempts;
    const pauseByTime = policy.stallPauseAfterNoProgressMs !== null
      && noProgressMs >= policy.stallPauseAfterNoProgressMs;

    if ((warningByAttempts || warningByTime) && !continuity.stallWarningAt) {
      continuity.stallWarningAt = timestamp;
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'continuation.warning',
        workerId: worker.workerId,
        payload: { noProgressAttempts, noProgressMs },
      });
    }
    if ((pauseByAttempts || pauseByTime) && !continuity.stallPausedAt) {
      continuity.stallPausedAt = timestamp;
      continuity.pauseReason = 'stall';
      continuity.disposition = 'paused';
      continuity.idleGraceDueAt = null;
      continuity.resumeAfterCurrentTurn = false;
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'continuation.paused_for_stall',
        workerId: worker.workerId,
        payload: { noProgressAttempts, noProgressMs },
      });
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'execution.stalled',
        workerId: worker.workerId,
        payload: { executionId: attempt.executionId, noProgressAttempts, noProgressMs },
      });
    }
  }

  private markAttemptIdleInState(state: StoredState, worker: Worker, executionId: ExecutionId, timestamp = now()): ExecutionAttempt {
    const attempt = state.executionAttempts[executionId];
    if (!attempt) throw new OrchestratorError('NOT_FOUND', `Execution attempt ${executionId} not found.`);
    if (attempt.idleAt) return attempt;
    const attemptWasTerminal = ['completed', 'blocked', 'failed', 'cancelled'].includes(attempt.state);
    const next: ExecutionAttempt = {
      ...attempt,
      state: attemptWasTerminal ? attempt.state : 'idle',
      idleAt: timestamp,
      terminalAt: attempt.terminalAt ?? timestamp,
    };
    state.executionAttempts[executionId] = next;
    const continuity = this.continuityFor(state, worker);
    if (continuity.pendingExecutionId === executionId) continuity.pendingExecutionId = null;
    const shouldAccountProgress = !attemptWasTerminal
      && !isTerminalWorker(worker)
      && !['awaiting_human', 'awaiting_dependency'].includes(continuity.disposition)
      && !(continuity.disposition === 'paused' && continuity.pauseReason === 'user')
      && !continuity.pauseAfterCurrentTurn;
    if (shouldAccountProgress) {
      const madeProgress = next.lastProgressCursor !== next.progressCursorAtStart;
      continuity.consecutiveExecutionAttemptsWithoutProgress = madeProgress
        ? 0
        : continuity.consecutiveExecutionAttemptsWithoutProgress + 1;
    }

    if (isTerminalWorker(worker)) {
      continuity.disposition = 'terminal';
      continuity.idleGraceDueAt = null;
      continuity.resumeAfterCurrentTurn = false;
      continuity.pauseAfterCurrentTurn = false;
    } else if (continuity.pauseAfterCurrentTurn) {
      continuity.pauseAfterCurrentTurn = false;
      continuity.resumeAfterCurrentTurn = false;
      continuity.pauseReason = 'user';
      continuity.disposition = 'paused';
      continuity.idleGraceDueAt = null;
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'execution.paused',
        workerId: worker.workerId,
        payload: { executionId, reason: 'pause_after_current_turn' },
      });
    } else if (!['awaiting_human', 'awaiting_dependency', 'paused'].includes(continuity.disposition)) {
      continuity.disposition = 'awaiting_execution';
      const policy = effectiveContinuationPolicy(state, worker.workerId);
      continuity.idleGraceDueAt = continuity.resumeAfterCurrentTurn
        ? timestamp
        : new Date(Date.parse(timestamp) + policy.idleGraceMs).toISOString();
    }
    continuity.updatedAt = timestamp;
    this.appendEvent(state, {
      runId: worker.runId,
      type: 'execution.idle',
      workerId: worker.workerId,
      payload: { executionId, sequence: next.sequence, reason: next.reason },
    });
    if (continuity.disposition === 'awaiting_execution') {
      this.applyStallPolicyInState(state, worker, next, timestamp);
    }
    return next;
  }

  private appendEvent(state: StoredState, input: {
    runId: RunId; type: EventType; workerId?: WorkerId; payload?: Record<string, unknown>;
  }): OrchestratorEvent {
    const cursor = (state.nextCursor[input.runId] ?? 0) + 1;
    state.nextCursor[input.runId] = cursor;
    const event: OrchestratorEvent = {
      eventId: id('evt'), runId: input.runId, cursor, type: input.type, timestamp: now(), payload: input.payload ?? {},
      ...(input.workerId ? { workerId: input.workerId } : {}),
    };
    (state.events[input.runId] ??= []).push(event);
    return event;
  }

  async createRun(input: { title: string; taskName?: string; rootConversationName?: string; idempotencyKey?: string }): Promise<{ run: Run; rootWorker: Worker }> {
    const fp = fingerprint({ title: input.title, taskName: input.taskName ?? input.title, rootConversationName: input.rootConversationName ?? input.title });
    const result = await this.repository.mutate((state) => {
      const key = input.idempotencyKey ? `run.create:${input.idempotencyKey}` : undefined;
      if (key && state.idempotency[key]) {
        const prior = state.idempotency[key];
        if (prior.fingerprint !== fp) throw new OrchestratorError('IDEMPOTENCY_CONFLICT', 'Idempotency key reused with different run input.');
        const run = state.runs[prior.resourceId];
        if (!run) throw new OrchestratorError('INTERNAL_ERROR', 'Idempotent run record is missing.');
        const rootWorker = state.workers[run.rootWorkerId];
        if (!rootWorker) throw new OrchestratorError('INTERNAL_ERROR', 'Root worker is missing.');
        return { run, rootWorker, changed: false };
      }

      const runId = id('run') as RunId;
      const workerId = id('wrk') as WorkerId;
      const assignmentId = id('asg');
      const timestamp = now();
      const assignment: Assignment = {
        assignmentId,
        objective: `Coordinate run: ${input.title}`,
        constraints: [], acceptanceCriteria: [], dependencies: [], allowedScope: [], forbiddenScope: [], artifacts: [],
      };
      const taskName = input.taskName ?? input.title;
      const rootConversationName = input.rootConversationName ?? input.title;
      const rootWorker: Worker = {
        workerId, runId, name: '/root', displayName: `${taskName} Root`, taskSummary: summarizeObjective(assignment.objective),
        parentWorkerId: null, executor: 'chatgpt-web', state: 'running',
        assignmentId, executionId: null, createdAt: timestamp, updatedAt: timestamp,
      };
      const run: Run = { runId, title: input.title, taskName, rootConversationName, rootWorkerId: workerId, state: 'active', createdAt: timestamp, updatedAt: timestamp };
      state.assignments[assignmentId] = assignment;
      state.workers[workerId] = rootWorker;
      state.runs[runId] = run;
      state.runContinuationPolicies[runId] = structuredClone(DEFAULT_CONTINUATION_POLICY);
      state.workerContinuity[workerId] = defaultWorkerContinuity(rootWorker);
      this.appendEvent(state, { runId, type: 'run.created', workerId, payload: { title: input.title, taskName, rootConversationName } });
      this.appendEvent(state, { runId, type: 'worker.created', workerId, payload: { name: '/root', displayName: rootWorker.displayName } });
      const initial = this.scheduleAttemptInState(state, rootWorker, 'initial', `execution.initial:${workerId}`, null, timestamp);
      this.markAttemptGeneratingInState(state, initial.attempt.executionId, timestamp);
      if (key) state.idempotency[key] = { resourceId: runId, fingerprint: fp };
      return { run, rootWorker: state.workers[workerId] ?? rootWorker, changed: true };
    });
    if (result.changed) this.notifier.notify(result.run.runId);
    return { run: result.run, rootWorker: result.rootWorker };
  }

  async getRun(runId: RunId): Promise<Run> {
    return this.repository.read((state) => {
      const run = state.runs[runId];
      if (!run) throw new OrchestratorError('NOT_FOUND', `Run ${runId} not found.`);
      return run;
    });
  }

  async getWorkerAssignmentContext(workerId: WorkerId): Promise<{ run: Run; worker: Worker; assignment: Assignment }> {
    return this.repository.read((state) => {
      const worker = state.workers[workerId];
      if (!worker) throw new OrchestratorError('NOT_FOUND', `Worker ${workerId} not found.`);
      const run = state.runs[worker.runId];
      if (!run) throw new OrchestratorError('NOT_FOUND', `Run ${worker.runId} not found.`);
      const assignment = state.assignments[worker.assignmentId];
      if (!assignment) throw new OrchestratorError('INTERNAL_ERROR', `Assignment ${worker.assignmentId} is missing.`);
      return { run, worker, assignment };
    });
  }

  async updateRunContext(runId: RunId, input: { taskName?: string; rootConversationName?: string }): Promise<{ run: Run; workers: Worker[] }> {
    const result = await this.repository.mutate((state) => {
      const run = state.runs[runId];
      if (!run) throw new OrchestratorError('NOT_FOUND', `Run ${runId} not found.`);
      const taskName = input.taskName?.trim() || run.taskName || run.title;
      const rootConversationName = input.rootConversationName?.trim() || run.rootConversationName || run.title;
      const updatedRun: Run = { ...run, taskName, rootConversationName, updatedAt: now() };
      state.runs[runId] = updatedRun;
      const workers = Object.values(state.workers).filter((worker) => worker.runId === runId);
      for (const worker of workers) {
        const internalName = worker.name.split('/').at(-1) || 'root';
        const displayName = worker.name === '/root' ? `${taskName} Root` : defaultWorkerDisplayName(taskName, internalName);
        state.workers[worker.workerId] = { ...worker, displayName, taskSummary: worker.taskSummary ?? summarizeObjective(state.assignments[worker.assignmentId]?.objective ?? displayName), updatedAt: now() };
      }
      this.appendEvent(state, { runId, type: 'run.metadata_changed', workerId: run.rootWorkerId, payload: { taskName, rootConversationName } });
      return { run: updatedRun, workers: Object.values(state.workers).filter((worker) => worker.runId === runId) };
    });
    this.notifier.notify(runId);
    return result;
  }

  async spawnWorker(input: {
    runId: RunId; parentWorkerId: WorkerId; name: string; displayName?: string; taskSummary?: string; assignment: AssignmentInput; idempotencyKey?: string;
  }): Promise<{ worker: Worker; assignment: Assignment }> {
    if (!/^[a-zA-Z0-9._-]+$/.test(input.name)) throw new OrchestratorError('INVALID_PARENT', 'Worker name must be one path-safe segment.');
    const fp = fingerprint(input);
    const result = await this.repository.mutate((state) => {
      const run = state.runs[input.runId];
      if (!run) throw new OrchestratorError('NOT_FOUND', `Run ${input.runId} not found.`);
      if (run.state !== 'active') throw new OrchestratorError('INVALID_STATE', 'Workers may only be spawned into an active run.');
      const parent = state.workers[input.parentWorkerId];
      if (!parent) throw new OrchestratorError('INVALID_PARENT', 'Parent worker does not exist.');
      if (parent.runId !== input.runId) throw new OrchestratorError('WRONG_RUN', 'Parent worker belongs to another run.');

      const key = input.idempotencyKey ? `worker.spawn:${input.runId}:${input.idempotencyKey}` : undefined;
      if (key && state.idempotency[key]) {
        const prior = state.idempotency[key];
        if (prior.fingerprint !== fp) throw new OrchestratorError('IDEMPOTENCY_CONFLICT', 'Idempotency key reused with different worker input.');
        const worker = state.workers[prior.resourceId];
        if (!worker) throw new OrchestratorError('INTERNAL_ERROR', 'Idempotent worker record is missing.');
        const assignment = state.assignments[worker.assignmentId];
        if (!assignment) throw new OrchestratorError('INTERNAL_ERROR', 'Worker assignment is missing.');
        return { worker, assignment, changed: false };
      }

      for (const dependency of input.assignment.dependencies ?? []) {
        const dependencyWorker = state.workers[dependency];
        if (!dependencyWorker || dependencyWorker.runId !== input.runId) {
          throw new OrchestratorError('WRONG_RUN', `Dependency ${dependency} is not a worker in this run.`);
        }
      }

      const workerId = id('wrk') as WorkerId;
      const assignmentId = id('asg');
      const timestamp = now();
      const assignment: Assignment = {
        assignmentId,
        objective: input.assignment.objective,
        constraints: input.assignment.constraints ?? [],
        acceptanceCriteria: input.assignment.acceptanceCriteria ?? [],
        dependencies: input.assignment.dependencies ?? [],
        allowedScope: input.assignment.allowedScope ?? [],
        forbiddenScope: input.assignment.forbiddenScope ?? [],
        artifacts: input.assignment.artifacts ?? [],
      };
      const taskName = run.taskName ?? run.title;
      const worker: Worker = {
        workerId, runId: input.runId, name: `${parent.name}/${input.name}`,
        displayName: input.displayName ?? defaultWorkerDisplayName(taskName, input.name),
        taskSummary: input.taskSummary ?? summarizeObjective(input.assignment.objective),
        parentWorkerId: parent.workerId, executor: 'chatgpt-web', state: 'launch_requested', assignmentId, executionId: null,
        createdAt: timestamp, updatedAt: timestamp,
      };
      state.assignments[assignmentId] = assignment;
      state.workers[workerId] = worker;
      state.workerContinuity[workerId] = defaultWorkerContinuity(worker);
      this.appendEvent(state, { runId: input.runId, type: 'worker.created', workerId, payload: { name: worker.name, displayName: worker.displayName, taskSummary: worker.taskSummary } });
      this.appendEvent(state, { runId: input.runId, type: 'worker.launch_requested', workerId });
      this.scheduleAttemptInState(state, worker, 'initial', `execution.initial:${workerId}`, null, timestamp);
      if (key) state.idempotency[key] = { resourceId: workerId, fingerprint: fp };
      return { worker: state.workers[workerId] ?? worker, assignment, changed: true };
    });
    if (result.changed) this.notifier.notify(input.runId);
    return { worker: result.worker, assignment: result.assignment };
  }

  async attachWorker(workerId: WorkerId): Promise<{ worker: Worker; assignment: Assignment }> {
    const result = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      const assignment = state.assignments[worker.assignmentId];
      if (!assignment) throw new OrchestratorError('INTERNAL_ERROR', 'Worker assignment is missing.');
      if (isTerminalWorker(worker)) throw new OrchestratorError('INVALID_STATE', `Worker cannot attach from state ${worker.state}.`);
      const continuity = this.continuityFor(state, worker);
      if (['awaiting_human', 'awaiting_dependency', 'paused'].includes(continuity.disposition)) {
        throw new OrchestratorError('CONTINUATION_BLOCKED', `Worker cannot attach while disposition is ${continuity.disposition}.`);
      }
      const timestamp = now();
      let updated = worker;
      let changed = false;
      if (!['running', 'waiting'].includes(worker.state)) {
        if (!['created', 'launch_requested', 'launching'].includes(worker.state)) {
          throw new OrchestratorError('INVALID_STATE', `Worker cannot attach from state ${worker.state}.`);
        }
        updated = { ...worker, state: 'running', updatedAt: timestamp };
        state.workers[workerId] = updated;
        this.appendEvent(state, { runId: worker.runId, type: 'worker.bound', workerId, payload: { executionId: updated.executionId } });
        this.appendEvent(state, { runId: worker.runId, type: 'worker.state_changed', workerId, payload: { from: worker.state, to: 'running' } });
        changed = true;
      }
      const attempt = this.activeAttemptInState(state, workerId);
      if (attempt && attempt.state !== 'generating') {
        this.markAttemptGeneratingInState(state, attempt.executionId, timestamp);
        changed = true;
      }
      continuity.disposition = 'running';
      continuity.idleGraceDueAt = null;
      continuity.updatedAt = timestamp;
      return { worker: state.workers[workerId] ?? updated, assignment, changed };
    });
    if (result.changed) this.notifier.notify(result.worker.runId);
    return { worker: result.worker, assignment: result.assignment };
  }

  async listWorkers(runId: RunId): Promise<Worker[]> {
    return this.repository.read((state) => {
      if (!state.runs[runId]) throw new OrchestratorError('NOT_FOUND', `Run ${runId} not found.`);
      return Object.values(state.workers)
        .filter((worker) => worker.runId === runId)
        .sort((a, b) => a.createdAt.localeCompare(b.createdAt));
    });
  }

  async sendMessage(input: {
    runId: RunId;
    fromWorkerId: WorkerId;
    toWorkerId: WorkerId;
    type: MessageType;
    body: string;
    delivery?: MessageDelivery;
    priority?: MessagePriority;
    idempotencyKey?: string;
    notBefore?: string | null;
  }): Promise<Message> {
    const result = await this.repository.mutate((state) => this.createMessageInState(state, {
      ...input,
      delivery: input.delivery ?? 'next_turn',
      priority: input.priority ?? 'normal',
    }));
    if (result.changed) this.notifier.notify(input.runId);
    return result.message;
  }

  async listPendingMessageDeliveries(at = now()): Promise<Message[]> {
    return this.repository.read((state) => Object.values(state.messages)
      .filter((message) => ['pending', 'retry_wait'].includes(message.deliveryState))
      .filter((message) => !message.notBefore || Date.parse(message.notBefore) <= Date.parse(at))
      .filter((message) => {
        const run = state.runs[message.runId];
        const worker = state.workers[message.toWorkerId];
        return run?.state === 'active' && Boolean(worker) && !isTerminalWorker(worker!);
      })
      .sort((a, b) => a.createdAt.localeCompare(b.createdAt)));
  }

  async deferMessageDelivery(messageId: string, notBefore: string, reason: string): Promise<Message> {
    const result = await this.repository.mutate((state) => {
      const message = state.messages[messageId];
      if (!message) throw new OrchestratorError('NOT_FOUND', `Message ${messageId} not found.`);
      if (['recorded', 'delivered', 'ambiguous'].includes(message.deliveryState)) return { message, changed: false };
      const next: Message = {
        ...message,
        deliveryState: 'retry_wait',
        notBefore,
        lastDeliveryError: reason,
      };
      state.messages[messageId] = next;
      this.appendEvent(state, {
        runId: message.runId,
        type: 'message.delivery_deferred',
        workerId: message.toWorkerId,
        payload: { messageId, notBefore, reason },
      });
      return { message: next, changed: true };
    });
    if (result.changed) this.notifier.notify(result.message.runId);
    return result.message;
  }

  async markMessageDeliveryDispatched(messageId: string, commandId: CommandId, at = now()): Promise<Message> {
    const result = await this.repository.mutate((state) => {
      const message = state.messages[messageId];
      if (!message) throw new OrchestratorError('NOT_FOUND', `Message ${messageId} not found.`);
      if (message.deliveryState === 'delivered' || message.deliveryState === 'ambiguous') return { message, changed: false };
      if (message.deliveryState === 'dispatching' && message.deliveryCommandId === commandId) return { message, changed: false };
      const next: Message = {
        ...message,
        deliveryState: 'dispatching',
        notBefore: null,
        deliveryCommandId: commandId,
        deliveryAttempts: message.deliveryAttempts + 1,
        lastDeliveryError: null,
      };
      state.messages[messageId] = next;
      this.appendEvent(state, {
        runId: message.runId,
        type: 'message.delivery_dispatched',
        workerId: message.toWorkerId,
        payload: { messageId, commandId, delivery: message.delivery, priority: message.priority, attempt: next.deliveryAttempts, at },
      });
      return { message: next, changed: true };
    });
    if (result.changed) this.notifier.notify(result.message.runId);
    return result.message;
  }

  async getMessageDeliveryContext(messageId: string): Promise<{ message: Message; binding: ConversationBinding | null; continuity: WorkerContinuityView }> {
    const message = await this.repository.read((state) => state.messages[messageId] ?? null);
    if (!message) throw new OrchestratorError('NOT_FOUND', `Message ${messageId} not found.`);
    return {
      message,
      binding: await this.getConversationBinding(message.toWorkerId),
      continuity: await this.getWorkerContinuity(message.toWorkerId),
    };
  }

  async progress(workerId: WorkerId, detail: string): Promise<OrchestratorEvent> {
    const event = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      if (!['running', 'waiting'].includes(worker.state)) throw new OrchestratorError('INVALID_STATE', 'Only active workers may report progress.');
      const timestamp = now();
      state.workers[workerId] = { ...worker, updatedAt: timestamp };
      const event = this.appendEvent(state, { runId: worker.runId, type: 'worker.progress', workerId, payload: { detail } });
      const continuity = this.continuityFor(state, worker);
      continuity.lastProgressCursor = event.cursor;
      continuity.lastProgressAt = event.timestamp;
      continuity.consecutiveExecutionAttemptsWithoutProgress = 0;
      continuity.stallWarningAt = null;
      continuity.updatedAt = timestamp;
      const attempt = this.activeAttemptInState(state, workerId) ?? this.latestAttemptInState(state, workerId);
      if (attempt && !attempt.terminalAt) {
        state.executionAttempts[attempt.executionId] = { ...attempt, lastProgressCursor: event.cursor };
      }
      const run = state.runs[worker.runId];
      const root = run ? state.workers[run.rootWorkerId] : undefined;
      if (root && root.workerId !== workerId && !isTerminalWorker(root)) {
        this.createMessageInState(state, {
          runId: worker.runId,
          fromWorkerId: workerId,
          toWorkerId: root.workerId,
          type: 'information',
          body: detail,
          delivery: 'next_turn',
          priority: 'normal',
          idempotencyKey: `progress:${event.eventId}`,
        }, event.timestamp);
      }
      return event;
    });
    this.notifier.notify(event.runId);
    return event;
  }

  async completeWorker(workerId: WorkerId, summary: string): Promise<Worker> {
    const updated = await this.repository.mutate((state) => {
      const worker = state.workers[workerId];
      if (!worker) throw new OrchestratorError('NOT_FOUND', `Worker ${workerId} not found.`);
      if (!['running', 'waiting'].includes(worker.state)) throw new OrchestratorError('INVALID_STATE', `Worker cannot complete from ${worker.state}.`);
      if (worker.parentWorkerId === null) {
        const unfinished = Object.values(state.workers).filter((candidate) => candidate.runId === worker.runId && candidate.workerId !== workerId && !['completed', 'failed', 'cancelled'].includes(candidate.state));
        if (unfinished.length > 0) throw new OrchestratorError('INVALID_STATE', 'Root worker cannot complete while child workers are non-terminal.');
      }
      const timestamp = now();
      const next: Worker = { ...worker, state: 'completed', updatedAt: timestamp };
      state.workers[workerId] = next;
      const continuity = this.continuityFor(state, worker);
      continuity.disposition = 'terminal';
      continuity.pendingExecutionId = null;
      continuity.idleGraceDueAt = null;
      continuity.pauseAfterCurrentTurn = false;
      continuity.resumeAfterCurrentTurn = false;
      continuity.updatedAt = timestamp;
      this.appendEvent(state, { runId: worker.runId, type: 'worker.completed', workerId, payload: { summary } });
      this.appendEvent(state, { runId: worker.runId, type: 'worker.state_changed', workerId, payload: { from: worker.state, to: 'completed' } });
      this.finishAttemptInState(state, worker, 'completed', 'worker_completed', timestamp);
      if (worker.parentWorkerId === null) {
        const run = state.runs[worker.runId];
        if (run && !['completed', 'failed', 'cancelled'].includes(run.state)) {
          state.runs[worker.runId] = { ...run, state: 'completed', updatedAt: timestamp };
          this.appendEvent(state, { runId: worker.runId, type: 'run.state_changed', payload: { from: run.state, to: 'completed' } });
        }
      }
      return next;
    });
    this.notifier.notify(updated.runId);
    return updated;
  }

  async failWorker(workerId: WorkerId, reason: string): Promise<Worker> {
    const updated = await this.repository.mutate((state) => {
      const worker = state.workers[workerId];
      if (!worker) throw new OrchestratorError('NOT_FOUND', `Worker ${workerId} not found.`);
      if (['completed', 'failed', 'cancelled'].includes(worker.state)) throw new OrchestratorError('INVALID_STATE', `Worker cannot fail from ${worker.state}.`);
      const timestamp = now();
      const next: Worker = { ...worker, state: 'failed', updatedAt: timestamp };
      state.workers[workerId] = next;
      const continuity = this.continuityFor(state, worker);
      continuity.disposition = 'terminal';
      continuity.pendingExecutionId = null;
      continuity.idleGraceDueAt = null;
      continuity.pauseAfterCurrentTurn = false;
      continuity.resumeAfterCurrentTurn = false;
      continuity.updatedAt = timestamp;
      this.appendEvent(state, { runId: worker.runId, type: 'worker.failed', workerId, payload: { reason } });
      this.appendEvent(state, { runId: worker.runId, type: 'worker.state_changed', workerId, payload: { from: worker.state, to: 'failed' } });
      this.finishAttemptInState(state, worker, 'failed', 'failed', timestamp, 'WORKER_FAILED', reason);
      if (worker.parentWorkerId === null) {
        const run = state.runs[worker.runId];
        if (run && !['completed', 'failed', 'cancelled'].includes(run.state)) {
          state.runs[worker.runId] = { ...run, state: 'failed', updatedAt: timestamp };
          this.appendEvent(state, { runId: worker.runId, type: 'run.state_changed', payload: { from: run.state, to: 'failed', reason } });
          for (const child of Object.values(state.workers)) {
            if (child.runId !== worker.runId || child.workerId === workerId || ['completed', 'failed', 'cancelled'].includes(child.state)) continue;
            state.workers[child.workerId] = { ...child, state: 'cancelled', updatedAt: timestamp };
            const childContinuity = this.continuityFor(state, child);
            childContinuity.disposition = 'terminal';
            childContinuity.pendingExecutionId = null;
            childContinuity.idleGraceDueAt = null;
            childContinuity.updatedAt = timestamp;
            this.finishAttemptInState(state, child, 'cancelled', 'cancelled', timestamp, 'ROOT_FAILED', 'root worker failed');
            this.appendEvent(state, { runId: worker.runId, type: 'worker.state_changed', workerId: child.workerId, payload: { from: child.state, to: 'cancelled', reason: 'root worker failed' } });
          }
        }
      }
      return next;
    });
    this.notifier.notify(updated.runId);
    return updated;
  }

  async listEvents(runId: RunId, afterCursor = 0): Promise<OrchestratorEvent[]> {
    return this.repository.read((state) => {
      if (!state.runs[runId]) throw new OrchestratorError('NOT_FOUND', `Run ${runId} not found.`);
      return (state.events[runId] ?? []).filter((event) => event.cursor > afterCursor);
    });
  }

  async waitEvents(runId: RunId, afterCursor: number, timeoutMs: number): Promise<WaitResult> {
    const boundedTimeout = Math.max(1, Math.min(timeoutMs, 60_000));
    const immediate = await this.listEvents(runId, afterCursor);
    if (immediate.length > 0) {
      return { cursor: immediate.at(-1)?.cursor ?? afterCursor, events: immediate, timedOut: false };
    }

    await this.notifier.wait(runId, boundedTimeout);
    const events = await this.listEvents(runId, afterCursor);
    return { cursor: events.at(-1)?.cursor ?? afterCursor, events, timedOut: events.length === 0 };
  }

  async pruneTerminalRuns(before: string): Promise<string[]> {
    return this.repository.mutate((state) => pruneStoredTerminalRuns(state, before));
  }

  async cancelRun(runId: RunId, reason: string): Promise<Run> {
    const updated = await this.repository.mutate((state) => {
      const run = state.runs[runId];
      if (!run) throw new OrchestratorError('NOT_FOUND', `Run ${runId} not found.`);
      if (['completed', 'failed', 'cancelled'].includes(run.state)) return run;
      const timestamp = now();
      const next: Run = { ...run, state: 'cancelled', updatedAt: timestamp };
      state.runs[runId] = next;
      this.appendEvent(state, { runId, type: 'run.state_changed', payload: { from: run.state, to: 'cancelled', reason } });
      for (const worker of Object.values(state.workers)) {
        if (worker.runId !== runId || ['completed', 'failed', 'cancelled'].includes(worker.state)) continue;
        state.workers[worker.workerId] = { ...worker, state: 'cancelled', updatedAt: timestamp };
        const continuity = this.continuityFor(state, worker);
        continuity.disposition = 'terminal';
        continuity.pendingExecutionId = null;
        continuity.idleGraceDueAt = null;
        continuity.pauseAfterCurrentTurn = false;
        continuity.resumeAfterCurrentTurn = false;
        continuity.updatedAt = timestamp;
        this.finishAttemptInState(state, worker, 'cancelled', 'cancelled', timestamp, 'RUN_CANCELLED', reason);
        this.appendEvent(state, { runId, type: 'worker.state_changed', workerId: worker.workerId, payload: { from: worker.state, to: 'cancelled', reason } });
      }
      return next;
    });
    this.notifier.notify(runId);
    return updated;
  }

  async interruptWorker(workerId: WorkerId, reason: string): Promise<Worker> {
    const updated = await this.repository.mutate((state) => {
      const worker = state.workers[workerId];
      if (!worker) throw new OrchestratorError('NOT_FOUND', `Worker ${workerId} not found.`);
      if (['completed', 'failed', 'cancelled'].includes(worker.state)) return worker;
      const timestamp = now();
      const next: Worker = { ...worker, state: 'cancelled', updatedAt: timestamp };
      state.workers[workerId] = next;
      const continuity = this.continuityFor(state, worker);
      continuity.disposition = 'terminal';
      continuity.pendingExecutionId = null;
      continuity.idleGraceDueAt = null;
      continuity.pauseAfterCurrentTurn = false;
      continuity.resumeAfterCurrentTurn = false;
      continuity.updatedAt = timestamp;
      this.finishAttemptInState(state, worker, 'cancelled', 'cancelled', timestamp, 'WORKER_INTERRUPTED', reason);
      this.appendEvent(state, { runId: worker.runId, type: 'worker.state_changed', workerId, payload: { from: worker.state, to: 'cancelled', reason } });
      return next;
    });
    this.notifier.notify(updated.runId);
    return updated;
  }

  async followupWorker(workerId: WorkerId, assignmentInput: AssignmentInput): Promise<{ worker: Worker; assignment: Assignment; requiresLaunch: boolean }> {
    const result = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      const run = state.runs[worker.runId];
      if (!run || run.state !== 'active') throw new OrchestratorError('INVALID_STATE', 'Follow-up work requires an active run.');
      const failedExecutorCommand = Object.values(state.executorCommands)
        .some((record) => record.command.workerId === workerId && record.status === 'failed');
      const recoverableLaunchRequest = worker.state === 'launch_requested' && failedExecutorCommand;
      if (!['completed', 'waiting', 'running', 'launch_failed'].includes(worker.state) && !recoverableLaunchRequest) {
        throw new OrchestratorError('INVALID_STATE', `Worker cannot receive follow-up from ${worker.state}.`);
      }
      const binding = state.conversations[workerId];
      const missingConversation = !binding || binding.state === 'closed';
      const requiresLaunch = worker.state === 'launch_failed' || recoverableLaunchRequest || missingConversation;
      for (const dependency of assignmentInput.dependencies ?? []) {
        const dependencyWorker = state.workers[dependency];
        if (!dependencyWorker || dependencyWorker.runId !== worker.runId) {
          throw new OrchestratorError('WRONG_RUN', `Dependency ${dependency} is not a worker in this run.`);
        }
      }
      const assignmentId = id('asg');
      const assignment: Assignment = {
        assignmentId,
        objective: assignmentInput.objective,
        constraints: assignmentInput.constraints ?? [],
        acceptanceCriteria: assignmentInput.acceptanceCriteria ?? [],
        dependencies: assignmentInput.dependencies ?? [],
        allowedScope: assignmentInput.allowedScope ?? [],
        forbiddenScope: assignmentInput.forbiddenScope ?? [],
        artifacts: assignmentInput.artifacts ?? [],
      };
      const timestamp = now();
      const activeAttempt = this.activeAttemptInState(state, workerId);
      if (activeAttempt) this.finishAttemptInState(state, worker, 'cancelled', 'cancelled', timestamp, 'FOLLOWUP_REPLACED_EXECUTION', 'Assignment replaced by follow-up.');
      const next: Worker = { ...worker, assignmentId, state: 'launch_requested', updatedAt: timestamp };
      state.assignments[assignmentId] = assignment;
      state.workers[workerId] = next;
      const continuity = this.continuityFor(state, next);
      continuity.disposition = 'awaiting_execution';
      continuity.humanWait = null;
      continuity.dependencyWait = null;
      continuity.pauseReason = null;
      continuity.pauseAfterCurrentTurn = false;
      continuity.resumeAfterCurrentTurn = false;
      continuity.pendingExecutionId = null;
      continuity.idleGraceDueAt = null;
      continuity.stallPausedAt = null;
      continuity.updatedAt = timestamp;
      this.appendEvent(state, { runId: worker.runId, type: 'worker.launch_requested', workerId, payload: { followup: true, assignmentId } });
      this.appendEvent(state, { runId: worker.runId, type: 'worker.state_changed', workerId, payload: { from: worker.state, to: 'launch_requested' } });
      const previous = this.latestAttemptInState(state, workerId);
      this.scheduleAttemptInState(state, next, 'followup', `execution.followup:${assignmentId}`, previous?.executionId ?? null, timestamp);
      return { worker: state.workers[workerId] ?? next, assignment, requiresLaunch };
    });
    this.notifier.notify(result.worker.runId);
    return result;
  }

  async broadcastMessage(input: { runId: RunId; fromWorkerId: WorkerId; type: MessageType; body: string }): Promise<Message[]> {
    const recipients = await this.listWorkers(input.runId);
    const messages: Message[] = [];
    for (const recipient of recipients) {
      if (recipient.workerId === input.fromWorkerId) continue;
      messages.push(await this.sendMessage({ ...input, toWorkerId: recipient.workerId }));
    }
    return messages;
  }

  async enqueueExecutorCommand(command: ExecutorCommand): Promise<ExecutorCommandRecord> {
    const record = await this.repository.mutate((state) => {
      const existing = Object.values(state.executorCommands).find((candidate) => candidate.command.idempotencyKey === command.idempotencyKey);
      if (existing) {
        if (JSON.stringify(existing.command) !== JSON.stringify(command)) {
          throw new OrchestratorError('IDEMPOTENCY_CONFLICT', 'Executor command idempotency key reused with different command.');
        }
        return existing;
      }
      const worker = state.workers[command.workerId];
      if (!worker) throw new OrchestratorError('NOT_FOUND', `Worker ${command.workerId} not found.`);
      const timestamp = now();
      const created: ExecutorCommandRecord = {
        command,
        status: 'pending',
        attempts: 0,
        executorId: null,
        createdAt: timestamp,
        updatedAt: timestamp,
        result: null,
        error: null,
      };
      state.executorCommands[command.commandId] = created;
      this.linkPendingAttemptToCommand(state, command);
      return created;
    });
    this.notifier.notify((await this.repository.read((state) => state.workers[command.workerId]?.runId)) as RunId);
    return record;
  }

  async listDispatchableExecutorCommands(): Promise<ExecutorCommandRecord[]> {
    return this.repository.read((state) => Object.values(state.executorCommands)
      .filter((record) => ['pending', 'sent', 'acked'].includes(record.status))
      .filter((record) => Date.parse(record.command.deadline) > Date.now())
      .sort((a, b) => a.createdAt.localeCompare(b.createdAt)));
  }

  async expirePendingExecutorCommands(at = now()): Promise<ExecutorCommandRecord[]> {
    const result = await this.repository.mutate((state) => {
      const timestampMs = Date.parse(at);
      const expired: ExecutorCommandRecord[] = [];
      const runIds = new Set<RunId>();
      for (const record of Object.values(state.executorCommands)) {
        if (record.status !== 'pending' || Date.parse(record.command.deadline) > timestampMs) continue;
        const failed: ExecutorCommandRecord = {
          ...record,
          status: 'failed',
          updatedAt: at,
          error: 'Executor command expired before submission.',
        };
        state.executorCommands[record.command.commandId] = failed;
        this.settleMessageDeliveryInState(state, record.command.commandId, false, at, failed.error);
        expired.push(failed);
        const worker = state.workers[record.command.workerId];
        if (!worker) continue;
        runIds.add(worker.runId);
        const linkedAttempt = Object.values(state.executionAttempts)
          .find((attempt) => attempt.continuationCommandId === record.command.commandId);
        if (linkedAttempt?.state === 'scheduled') {
          this.finishAttemptInState(
            state,
            worker,
            'failed',
            'executor_unavailable',
            at,
            'EXECUTOR_UNAVAILABLE',
            'Executor command expired before submission.',
          );
          const continuity = this.continuityFor(state, worker);
          continuity.disposition = 'awaiting_execution';
          continuity.pauseReason = null;
          continuity.pauseAfterCurrentTurn = false;
          const binding = state.conversations[worker.workerId];
          const policy = effectiveContinuationPolicy(state, worker.workerId);
          continuity.idleGraceDueAt = binding && ['idle', 'ready'].includes(binding.state) && policy.mode === 'auto'
            ? at
            : null;
          continuity.updatedAt = at;
          if (record.command.type === 'conversation.create' && !isTerminalWorker(worker)) {
            state.workers[worker.workerId] = { ...worker, state: 'launch_failed', updatedAt: at };
            this.appendEvent(state, {
              runId: worker.runId,
              type: 'worker.state_changed',
              workerId: worker.workerId,
              payload: { from: worker.state, to: 'launch_failed', error: 'Executor command expired before submission.' },
            });
          }
        }
      }
      return { expired, runIds: [...runIds] };
    });
    for (const runId of result.runIds) this.notifier.notify(runId);
    return result.expired;
  }

  async markExecutorCommandSent(commandId: string, executorId: string): Promise<ExecutorCommandRecord> {
    return this.repository.mutate((state) => {
      const record = state.executorCommands[commandId];
      if (!record) throw new OrchestratorError('NOT_FOUND', `Executor command ${commandId} not found.`);
      if (['completed', 'failed'].includes(record.status)) return record;
      if (record.status === 'pending') {
        const linkedAttempt = Object.values(state.executionAttempts).find((attempt) => attempt.continuationCommandId === commandId);
        if (linkedAttempt) {
          const worker = state.workers[linkedAttempt.workerId];
          const continuity = worker ? this.continuityFor(state, worker) : null;
          const invalid = linkedAttempt.state !== 'scheduled'
            || !worker
            || isTerminalWorker(worker)
            || !continuity
            || ['awaiting_human', 'awaiting_dependency', 'paused', 'terminal'].includes(continuity.disposition);
          if (invalid) {
            const failed: ExecutorCommandRecord = {
              ...record,
              status: 'failed',
              executorId,
              updatedAt: now(),
              error: 'Execution attempt is no longer dispatchable.',
            };
            state.executorCommands[commandId] = failed;
            return failed;
          }
        }
      }
      const next: ExecutorCommandRecord = { ...record, status: 'sent', attempts: record.attempts + 1, executorId, updatedAt: now() };
      state.executorCommands[commandId] = next;
      return next;
    });
  }

  async acknowledgeExecutorCommand(commandId: string, executorId: string): Promise<ExecutorCommandRecord> {
    const result = await this.repository.mutate((state) => {
      const record = state.executorCommands[commandId];
      if (!record) throw new OrchestratorError('NOT_FOUND', `Executor command ${commandId} not found.`);
      if (record.executorId && record.executorId !== executorId) throw new OrchestratorError('UNAUTHORIZED', 'Command is assigned to another executor.');
      if (['completed', 'failed'].includes(record.status)) return { record, runId: state.workers[record.command.workerId]?.runId, changed: false };
      const next: ExecutorCommandRecord = { ...record, status: 'acked', executorId, updatedAt: now() };
      state.executorCommands[commandId] = next;
      const worker = state.workers[record.command.workerId];
      if (worker) this.appendEvent(state, { runId: worker.runId, type: 'executor.command_acked', workerId: worker.workerId, payload: { commandId, executorId } });
      return { record: next, runId: worker?.runId, changed: true };
    });
    if (result.changed && result.runId) this.notifier.notify(result.runId);
    return result.record;
  }

  async completeExecutorCommand(commandId: string, executorId: string, success: boolean, result: Record<string, unknown>, error?: string): Promise<ExecutorCommandRecord> {
    const output = await this.repository.mutate((state) => {
      const record = state.executorCommands[commandId];
      if (!record) throw new OrchestratorError('NOT_FOUND', `Executor command ${commandId} not found.`);
      if (record.executorId && record.executorId !== executorId) throw new OrchestratorError('UNAUTHORIZED', 'Command is assigned to another executor.');
      const timestamp = now();
      const next: ExecutorCommandRecord = {
        ...record,
        status: success ? 'completed' : 'failed',
        executorId,
        updatedAt: timestamp,
        result,
        error: error ?? null,
      };
      state.executorCommands[commandId] = next;
      this.settleMessageDeliveryInState(state, record.command.commandId, success, timestamp, error ?? null);
      const worker = state.workers[record.command.workerId];
      const linkedAttempt = Object.values(state.executionAttempts).find((attempt) => attempt.continuationCommandId === commandId);
      if (success && linkedAttempt) {
        this.markAttemptSubmittedInState(state, linkedAttempt.executionId, record.command.commandId, state.conversations[record.command.workerId]?.conversationId ?? undefined);
      }
      if (worker && !success) {
        if (linkedAttempt) this.finishAttemptInState(state, worker, 'failed', 'failed', now(), 'EXECUTOR_COMMAND_FAILED', error ?? 'Executor command failed.');
        if (worker.state === 'launch_requested' && ['conversation.create', 'conversation.send'].includes(record.command.type)) {
          state.workers[worker.workerId] = { ...worker, state: 'launch_failed', updatedAt: now() };
          const continuity = this.continuityFor(state, worker);
          continuity.disposition = 'awaiting_execution';
          continuity.pendingExecutionId = null;
          continuity.updatedAt = now();
          this.appendEvent(state, { runId: worker.runId, type: 'worker.state_changed', workerId: worker.workerId, payload: { from: worker.state, to: 'launch_failed', error: error ?? 'Executor command failed.' } });
        }
      }
      return { record: next, runId: worker?.runId };
    });
    if (output.runId) this.notifier.notify(output.runId);
    return output.record;
  }

  async recordConversationState(input: {
    workerId: WorkerId; executorId: string; state: ConversationState; tabId?: number; conversationId?: string; url?: string; error?: string;
  }): Promise<ConversationBinding> {
    const result = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, input.workerId);
      const previous = state.conversations[input.workerId];
      if (isTerminalWorker(worker) && previous) return { binding: previous, runId: worker.runId, changed: false };
      const lifecycleAttempt = this.activeAttemptInState(state, input.workerId) ?? this.latestAttemptInState(state, input.workerId);
      const needsLifecycleTransition = Boolean(lifecycleAttempt && (
        (input.state === 'idle' && !lifecycleAttempt.idleAt && ['scheduled', 'submitted', 'generating'].includes(lifecycleAttempt.state))
        || (input.state === 'generating' && lifecycleAttempt.state !== 'generating' && !lifecycleAttempt.terminalAt)
      ));
      const unchanged = previous
        && previous.executorId === input.executorId
        && previous.state === input.state
        && previous.tabId === (input.tabId ?? previous.tabId)
        && previous.conversationId === (input.conversationId ?? previous.conversationId)
        && previous.url === (input.url ?? previous.url)
        && !input.error
        && !needsLifecycleTransition;
      if (unchanged) return { binding: previous, runId: worker.runId, changed: false };
      const timestamp = now();
      const binding: ConversationBinding = {
        workerId: input.workerId,
        executorId: input.executorId,
        tabId: input.tabId ?? previous?.tabId ?? null,
        conversationId: input.conversationId ?? previous?.conversationId ?? null,
        url: input.url ?? previous?.url ?? null,
        state: input.state,
        updatedAt: timestamp,
      };
      state.conversations[input.workerId] = binding;
      this.appendEvent(state, {
        runId: worker.runId,
        type: input.state === 'loading' && !previous ? 'conversation.created' : 'conversation.state_changed',
        workerId: worker.workerId,
        payload: { state: input.state, tabId: binding.tabId, conversationId: binding.conversationId, url: binding.url, ...(input.error ? { error: input.error } : {}) },
      });
      const attempt = this.activeAttemptInState(state, input.workerId) ?? this.latestAttemptInState(state, input.workerId);
      if (attempt && binding.conversationId && !attempt.conversationId) {
        state.executionAttempts[attempt.executionId] = { ...attempt, conversationId: binding.conversationId };
      }
      if (attempt && input.state === 'generating') {
        this.markAttemptGeneratingInState(state, attempt.executionId, timestamp);
      }
      if (attempt && input.state === 'idle' && !attempt.idleAt) {
        this.markAttemptIdleInState(state, worker, attempt.executionId, timestamp);
      }
      return { binding, runId: worker.runId, changed: true };
    });
    if (result.changed) this.notifier.notify(result.runId);
    return result.binding;
  }

  async getRunContinuationPolicy(runId: RunId): Promise<ContinuationPolicy> {
    return this.repository.read((state) => {
      if (!state.runs[runId]) throw new OrchestratorError('NOT_FOUND', `Run ${runId} not found.`);
      return state.runContinuationPolicies[runId] ?? DEFAULT_CONTINUATION_POLICY;
    });
  }

  async setRunContinuationPolicy(runId: RunId, patch: Partial<ContinuationPolicy>): Promise<ContinuationPolicy> {
    const result = await this.repository.mutate((state) => {
      const run = state.runs[runId];
      if (!run) throw new OrchestratorError('NOT_FOUND', `Run ${runId} not found.`);
      const previous = state.runContinuationPolicies[runId] ?? DEFAULT_CONTINUATION_POLICY;
      const next = ContinuationPolicySchema.parse({ ...previous, ...patch });
      if (JSON.stringify(previous) === JSON.stringify(next)) return { policy: previous, changed: false };
      state.runContinuationPolicies[runId] = next;
      this.appendEvent(state, {
        runId,
        type: 'continuation.policy_changed',
        payload: { target: 'run', previous, next },
      });
      const timestamp = now();
      for (const worker of Object.values(state.workers)) {
        if (worker.runId !== runId) continue;
        const continuity = this.continuityFor(state, worker);
        if (continuity.policyOverride || isTerminalWorker(worker)) continue;
        const binding = state.conversations[worker.workerId];
        if (next.mode === 'auto') {
          if (binding && ['idle', 'ready'].includes(binding.state) && continuity.disposition === 'awaiting_execution') {
            continuity.idleGraceDueAt = new Date(Date.parse(timestamp) + next.idleGraceMs).toISOString();
          }
          this.ensureKeepaliveInState(state, worker, timestamp);
          continuity.updatedAt = timestamp;
        } else {
          continuity.activeKeepaliveDueAt = null;
        }
      }
      return { policy: next, changed: true };
    });
    if (result.changed) this.notifier.notify(runId);
    return result.policy;
  }

  async getWorkerContinuity(workerId: WorkerId): Promise<WorkerContinuityView> {
    return this.repository.read((state) => {
      const worker = this.workerOrThrow(state, workerId);
      const continuity = state.workerContinuity[workerId] ?? defaultWorkerContinuity(worker);
      const effectivePolicy = effectiveContinuationPolicy(state, workerId);
      const policySource: WorkerContinuityView['policySource'] = continuity.policyOverride
        ? 'worker'
        : state.runContinuationPolicies[worker.runId]
          ? 'run'
          : 'default';
      return { continuity, effectivePolicy, policySource, latestAttempt: this.latestAttemptInState(state, workerId) };
    });
  }

  async setWorkerContinuationPolicy(workerId: WorkerId, patch: Partial<ContinuationPolicy> | null): Promise<WorkerContinuityView> {
    const result = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      const continuity = this.continuityFor(state, worker);
      const previous = continuity.policyOverride;
      const next = patch === null ? null : ContinuationPolicySchema.parse({ ...effectiveContinuationPolicy(state, workerId), ...patch });
      if (JSON.stringify(previous) === JSON.stringify(next)) return { changed: false, runId: worker.runId };
      continuity.policyOverride = next;
      continuity.updatedAt = now();
      const effective = effectiveContinuationPolicy(state, workerId);
      const binding = state.conversations[workerId];
      if (effective.mode === 'auto' && binding && ['idle', 'ready'].includes(binding.state) && continuity.disposition === 'awaiting_execution') {
        continuity.idleGraceDueAt = new Date(Date.now() + effective.idleGraceMs).toISOString();
      }
      if (effective.mode === 'manual' && !continuity.resumeAfterCurrentTurn) {
        continuity.idleGraceDueAt = null;
        continuity.activeKeepaliveDueAt = null;
      } else if (effective.mode === 'auto') {
        this.ensureKeepaliveInState(state, worker, continuity.updatedAt);
      }
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'continuation.policy_changed',
        workerId,
        payload: { target: 'worker', previous, next, effective },
      });
      return { changed: true, runId: worker.runId };
    });
    if (result.changed) this.notifier.notify(result.runId);
    return this.getWorkerContinuity(workerId);
  }

  async scheduleExecutionAttempt(input: {
    workerId: WorkerId;
    reason: ExecutionAttemptReason;
    idempotencyKey: string;
    resumeOfExecutionId?: ExecutionId | null;
  }): Promise<ExecutionAttempt> {
    const result = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, input.workerId);
      return this.scheduleAttemptInState(state, worker, input.reason, input.idempotencyKey, input.resumeOfExecutionId ?? null);
    });
    if (result.changed) this.notifier.notify(result.attempt.runId);
    return result.attempt;
  }

  async listExecutionAttempts(workerId: WorkerId): Promise<ExecutionAttempt[]> {
    return this.repository.read((state) => {
      this.workerOrThrow(state, workerId);
      return Object.values(state.executionAttempts)
        .filter((attempt) => attempt.workerId === workerId)
        .sort((a, b) => a.sequence - b.sequence);
    });
  }

  async listPendingExecutionAttempts(): Promise<ExecutionAttempt[]> {
    return this.repository.read((state) => Object.values(state.executionAttempts)
      .filter((attempt) => attempt.state === 'scheduled')
      .filter((attempt) => {
        const worker = state.workers[attempt.workerId];
        if (!worker || isTerminalWorker(worker) || state.runs[worker.runId]?.state !== 'active') return false;
        const continuity = state.workerContinuity[worker.workerId] ?? defaultWorkerContinuity(worker);
        return !['awaiting_human', 'awaiting_dependency', 'paused', 'terminal'].includes(continuity.disposition);
      })
      .sort((a, b) => a.scheduledAt.localeCompare(b.scheduledAt)));
  }

  async markExecutionSubmitted(executionId: ExecutionId, input: { commandId?: CommandId; conversationId?: string } = {}): Promise<ExecutionAttempt> {
    const result = await this.repository.mutate((state) => {
      const attempt = state.executionAttempts[executionId];
      if (!attempt) throw new OrchestratorError('NOT_FOUND', `Execution attempt ${executionId} not found.`);
      const next = this.markAttemptSubmittedInState(state, executionId, input.commandId, input.conversationId);
      return { attempt: next, runId: attempt.runId };
    });
    this.notifier.notify(result.runId);
    return result.attempt;
  }

  async markExecutionGenerating(executionId: ExecutionId): Promise<ExecutionAttempt> {
    const result = await this.repository.mutate((state) => {
      const attempt = state.executionAttempts[executionId];
      if (!attempt) throw new OrchestratorError('NOT_FOUND', `Execution attempt ${executionId} not found.`);
      return { attempt: this.markAttemptGeneratingInState(state, executionId), runId: attempt.runId };
    });
    this.notifier.notify(result.runId);
    return result.attempt;
  }

  async markExecutionIdle(executionId: ExecutionId): Promise<ExecutionAttempt> {
    const result = await this.repository.mutate((state) => {
      const attempt = state.executionAttempts[executionId];
      if (!attempt) throw new OrchestratorError('NOT_FOUND', `Execution attempt ${executionId} not found.`);
      const worker = this.workerOrThrow(state, attempt.workerId);
      return { attempt: this.markAttemptIdleInState(state, worker, executionId), runId: attempt.runId };
    });
    this.notifier.notify(result.runId);
    return result.attempt;
  }

  async awaitHuman(workerId: WorkerId, input: {
    reason: string;
    request: string;
    choices?: string[];
    contextRefs?: string[];
  }): Promise<WorkerContinuityView> {
    const runId = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      if (isTerminalWorker(worker)) throw new OrchestratorError('INVALID_STATE', `Worker cannot await human from ${worker.state}.`);
      const timestamp = now();
      const continuity = this.continuityFor(state, worker);
      continuity.disposition = 'awaiting_human';
      continuity.humanWait = {
        reason: input.reason,
        request: input.request,
        requestedAt: timestamp,
        choices: input.choices ?? [],
        contextRefs: input.contextRefs ?? [],
      };
      continuity.dependencyWait = null;
      continuity.pauseReason = null;
      continuity.pauseAfterCurrentTurn = false;
      continuity.resumeAfterCurrentTurn = false;
      continuity.pendingExecutionId = null;
      continuity.idleGraceDueAt = null;
      continuity.updatedAt = timestamp;
      state.workers[workerId] = { ...worker, updatedAt: timestamp };
      this.finishAttemptInState(state, worker, 'blocked', 'awaiting_human', timestamp);
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'worker.awaiting_human',
        workerId,
        payload: { reason: input.reason, request: input.request, choices: input.choices ?? [], contextRefs: input.contextRefs ?? [] },
      });
      return worker.runId;
    });
    this.notifier.notify(runId);
    return this.getWorkerContinuity(workerId);
  }

  async awaitDependency(workerId: WorkerId, input: { reason: string; dependencies?: WorkerId[] }): Promise<WorkerContinuityView> {
    const runId = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      if (isTerminalWorker(worker)) throw new OrchestratorError('INVALID_STATE', `Worker cannot await dependency from ${worker.state}.`);
      const assignment = state.assignments[worker.assignmentId];
      const dependencies = input.dependencies ?? assignment?.dependencies ?? [];
      for (const dependency of dependencies) {
        const dependencyWorker = state.workers[dependency];
        if (!dependencyWorker || dependencyWorker.runId !== worker.runId) {
          throw new OrchestratorError('WRONG_RUN', `Dependency ${dependency} is not a worker in this run.`);
        }
      }
      const timestamp = now();
      const continuity = this.continuityFor(state, worker);
      continuity.disposition = 'awaiting_dependency';
      continuity.dependencyWait = { reason: input.reason, dependencies, requestedAt: timestamp };
      continuity.humanWait = null;
      continuity.pauseReason = null;
      continuity.pauseAfterCurrentTurn = false;
      continuity.resumeAfterCurrentTurn = false;
      continuity.pendingExecutionId = null;
      continuity.idleGraceDueAt = null;
      continuity.updatedAt = timestamp;
      state.workers[workerId] = { ...worker, updatedAt: timestamp };
      this.finishAttemptInState(state, worker, 'blocked', 'awaiting_dependency', timestamp);
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'worker.awaiting_dependency',
        workerId,
        payload: { reason: input.reason, dependencies },
      });
      return worker.runId;
    });
    this.notifier.notify(runId);
    return this.getWorkerContinuity(workerId);
  }

  async pauseWorker(workerId: WorkerId): Promise<WorkerContinuityView> {
    const runId = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      if (isTerminalWorker(worker)) throw new OrchestratorError('INVALID_STATE', `Worker cannot pause from ${worker.state}.`);
      const timestamp = now();
      const continuity = this.continuityFor(state, worker);
      const active = this.activeAttemptInState(state, workerId);
      const command = active?.continuationCommandId ? state.executorCommands[active.continuationCommandId] : undefined;
      const commandInFlight = Boolean(command && ['sent', 'acked'].includes(command.status));
      const generating = state.conversations[workerId]?.state === 'generating' || active?.state === 'generating' || commandInFlight;
      continuity.pauseReason = 'user';
      continuity.idleGraceDueAt = null;
      continuity.resumeAfterCurrentTurn = false;
      if (generating) {
        continuity.pauseAfterCurrentTurn = true;
      } else {
        if (active?.state === 'scheduled') {
          this.finishAttemptInState(state, worker, 'cancelled', 'paused', timestamp, 'PAUSED_BEFORE_SUBMISSION', 'Execution paused before submission.');
          if (command?.status === 'pending') {
            state.executorCommands[command.command.commandId] = {
              ...command,
              status: 'failed',
              updatedAt: timestamp,
              error: 'Execution paused before submission.',
            };
          }
        }
        continuity.pendingExecutionId = null;
        continuity.pauseAfterCurrentTurn = false;
        continuity.disposition = 'paused';
      }
      continuity.updatedAt = timestamp;
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'execution.paused',
        workerId,
        payload: { afterCurrentTurn: generating },
      });
      return worker.runId;
    });
    this.notifier.notify(runId);
    return this.getWorkerContinuity(workerId);
  }

  async resumeWorker(workerId: WorkerId): Promise<WorkerContinuityView> {
    const runId = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      if (isTerminalWorker(worker)) throw new OrchestratorError('INVALID_STATE', `Worker cannot resume from ${worker.state}.`);
      const timestamp = now();
      const continuity = this.continuityFor(state, worker);
      continuity.humanWait = null;
      continuity.dependencyWait = null;
      continuity.pauseReason = null;
      continuity.pauseAfterCurrentTurn = false;
      continuity.resumeAfterCurrentTurn = false;
      continuity.stallWarningAt = null;
      continuity.stallPausedAt = null;
      continuity.consecutiveExecutionAttemptsWithoutProgress = 0;
      const binding = state.conversations[workerId];
      const active = this.activeAttemptInState(state, workerId);
      continuity.disposition = binding?.state === 'generating' || active?.state === 'generating' ? 'running' : 'awaiting_execution';
      const policy = effectiveContinuationPolicy(state, workerId);
      continuity.idleGraceDueAt = binding && ['idle', 'ready'].includes(binding.state) && policy.mode === 'auto'
        ? new Date(Date.parse(timestamp) + policy.idleGraceMs).toISOString()
        : null;
      continuity.updatedAt = timestamp;
      this.appendEvent(state, { runId: worker.runId, type: 'worker.resumed', workerId, payload: { disposition: continuity.disposition } });
      return worker.runId;
    });
    this.notifier.notify(runId);
    return this.getWorkerContinuity(workerId);
  }

  async continueNow(workerId: WorkerId, options: { overrideBlock?: boolean; idempotencyKey?: string } = {}): Promise<ContinueNowResult> {
    const result = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      if (isTerminalWorker(worker)) throw new OrchestratorError('INVALID_STATE', `Worker cannot continue from ${worker.state}.`);
      const continuity = this.continuityFor(state, worker);
      if (['awaiting_human', 'awaiting_dependency'].includes(continuity.disposition) && !options.overrideBlock) {
        throw new OrchestratorError('CONTINUATION_BLOCKED', `Worker is ${continuity.disposition}; resolve or explicitly override the wait.`);
      }
      if (options.overrideBlock) {
        continuity.humanWait = null;
        continuity.dependencyWait = null;
      }
      continuity.pauseReason = null;
      continuity.pauseAfterCurrentTurn = false;
      continuity.stallWarningAt = null;
      continuity.stallPausedAt = null;
      continuity.consecutiveExecutionAttemptsWithoutProgress = 0;
      const active = this.activeAttemptInState(state, workerId);
      const binding = state.conversations[workerId];
      if (binding?.state === 'generating' || active?.state === 'generating') {
        continuity.disposition = 'running';
        continuity.resumeAfterCurrentTurn = true;
        continuity.idleGraceDueAt = null;
        continuity.updatedAt = now();
        return { status: 'queued_after_current_turn' as const, attempt: null, runId: worker.runId, changed: true };
      }
      if (active) {
        return { status: 'already_pending' as const, attempt: active, runId: worker.runId, changed: false };
      }
      if (!binding || !['idle', 'ready'].includes(binding.state)) {
        throw new OrchestratorError('INVALID_STATE', 'Continue Now requires an idle/ready managed conversation or a currently generating attempt.');
      }
      const latest = this.latestAttemptInState(state, workerId);
      const key = options.idempotencyKey ?? `execution.manual:${workerId}:${latest?.executionId ?? 'none'}`;
      continuity.disposition = 'awaiting_execution';
      continuity.resumeAfterCurrentTurn = false;
      continuity.idleGraceDueAt = null;
      const scheduled = this.scheduleAttemptInState(state, worker, 'manual_resume', key, latest?.executionId ?? null);
      return { status: 'scheduled' as const, attempt: scheduled.attempt, runId: worker.runId, changed: scheduled.changed };
    });
    if (result.changed) this.notifier.notify(result.runId);
    return { status: result.status, attempt: result.attempt };
  }

  async reconcileWakeIntents(at = now(), random = Math.random): Promise<WorkerId[]> {
    const result = await this.repository.mutate((state) => {
      const changed = new Set<WorkerId>();
      for (const worker of Object.values(state.workers)) {
        const continuity = this.continuityFor(state, worker);
        if (this.ensureKeepaliveInState(state, worker, at, random)) changed.add(worker.workerId);
        if (continuity.activeKeepaliveDueAt && Date.parse(continuity.activeKeepaliveDueAt) <= Date.parse(at)) {
          const dueAt = continuity.activeKeepaliveDueAt;
          const run = state.runs[worker.runId];
          const root = run ? state.workers[run.rootWorkerId] : undefined;
          if (root && root.workerId !== worker.workerId && !isTerminalWorker(root)) {
            const checkpoint = this.createMessageInState(state, {
              runId: worker.runId,
              fromWorkerId: root.workerId,
              toWorkerId: worker.workerId,
              type: 'information',
              body: ORCHESTRATOR_CHECKPOINT_BODY,
              delivery: 'steer_now',
              priority: 'normal',
              idempotencyKey: `checkpoint:${worker.workerId}:${dueAt}`,
            }, at);
            if (checkpoint.changed) changed.add(worker.workerId);
          }
          continuity.activeKeepaliveDueAt = null;
          continuity.updatedAt = at;
          if (this.ensureKeepaliveInState(state, worker, at, random)) changed.add(worker.workerId);
        }
      }
      return [...changed];
    });
    for (const workerId of result) {
      const worker = await this.repository.read((state) => state.workers[workerId]);
      if (worker) this.notifier.notify(worker.runId);
    }
    return result;
  }

  async evaluateContinuation(workerId: WorkerId, at = now()): Promise<ContinuationEvaluation> {
    const result = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      const run = state.runs[worker.runId];
      const continuity = this.continuityFor(state, worker);
      const stop = (reason: string): ContinuationEvaluation & { changed: boolean } => ({ action: 'stop', reason, attempt: null, changed: false });
      const wait = (reason: string): ContinuationEvaluation & { changed: boolean } => ({ action: 'wait', reason, attempt: null, changed: false });
      if (!run || run.state !== 'active') return stop(`run_${run?.state ?? 'missing'}`);
      if (isTerminalWorker(worker)) return stop(`worker_${worker.state}`);
      if (continuity.disposition === 'awaiting_human') return stop('awaiting_human');
      if (continuity.disposition === 'awaiting_dependency') return stop('awaiting_dependency');
      if (continuity.disposition === 'paused') return stop(continuity.pauseReason === 'stall' ? 'paused_for_stall' : 'paused');
      const active = this.activeAttemptInState(state, workerId);
      if (active) return wait('execution_in_flight');
      const binding = state.conversations[workerId];
      const idleOrReady = Boolean(binding && ['idle', 'ready'].includes(binding.state));
      const idleDue = idleOrReady && Boolean(continuity.idleGraceDueAt) && Date.parse(continuity.idleGraceDueAt!) <= Date.parse(at);
      const latest = this.latestAttemptInState(state, workerId);
      if (continuity.resumeAfterCurrentTurn) {
        if (!idleDue) return wait(idleOrReady ? 'idle_grace' : 'conversation_not_idle');
        continuity.resumeAfterCurrentTurn = false;
        const scheduled = this.scheduleAttemptInState(
          state,
          worker,
          'manual_resume',
          `execution.postturn:${workerId}:${latest?.executionId ?? 'none'}`,
          latest?.executionId ?? null,
          at,
          ['operator:continue_now'],
        );
        return { action: 'schedule' as const, reason: 'manual_resume', attempt: scheduled.attempt, changed: scheduled.changed };
      }
      const policy = effectiveContinuationPolicy(state, workerId);
      if (policy.mode === 'manual') {
        continuity.disposition = 'awaiting_execution';
        continuity.updatedAt = at;
        return wait('manual_policy');
      }
      if (continuity.stallPausedAt) return stop('paused_for_stall');
      const hasWake = continuity.pendingWakeReasons.length > 0;
      const wakeDue = hasWake && (!continuity.wakeNotBefore || Date.parse(continuity.wakeNotBefore) <= Date.parse(at));
      const idleBypass = idleDue && policy.idleMayBypassMinSpacing;
      if (!idleOrReady) return wait(hasWake ? 'wake_queued_until_idle' : 'conversation_not_idle');
      if (!idleDue && !wakeDue) {
        if (hasWake && continuity.wakeNotBefore && Date.parse(continuity.wakeNotBefore) > Date.parse(at)) return wait('wake_not_before');
        return wait(continuity.idleGraceDueAt ? 'idle_grace' : 'no_idle_boundary');
      }
      if (hasWake && !wakeDue && !idleBypass) return wait('wake_not_before');
      const wakeReasons = hasWake ? [...continuity.pendingWakeReasons] : ['idle'];
      const scheduled = this.scheduleAttemptInState(
        state,
        worker,
        'auto_resume',
        `execution.auto:${workerId}:${latest?.executionId ?? 'none'}`,
        latest?.executionId ?? null,
        at,
        wakeReasons,
      );
      return { action: 'schedule' as const, reason: wakeReasons.includes('active_keepalive') ? 'active_keepalive' : 'auto_resume', attempt: scheduled.attempt, changed: scheduled.changed };
    });
    if (result.changed) {
      const worker = await this.repository.read((state) => state.workers[workerId]);
      if (worker) this.notifier.notify(worker.runId);
    }
    return { action: result.action, reason: result.reason, attempt: result.attempt };
  }

  async listContinuationCandidates(at = now()): Promise<WorkerId[]> {
    return this.repository.read((state) => Object.values(state.workers)
      .filter((worker) => {
        if (isTerminalWorker(worker) || state.runs[worker.runId]?.state !== 'active') return false;
        const continuity = state.workerContinuity[worker.workerId] ?? defaultWorkerContinuity(worker);
        if (['awaiting_human', 'awaiting_dependency', 'paused', 'terminal'].includes(continuity.disposition)) return false;
        if (this.activeAttemptInState(state, worker.workerId)) return false;
        const binding = state.conversations[worker.workerId];
        if (!binding || !['idle', 'ready'].includes(binding.state)) return false;
        const policy = effectiveContinuationPolicy(state, worker.workerId);
        const idleDue = Boolean(continuity.idleGraceDueAt) && Date.parse(continuity.idleGraceDueAt!) <= Date.parse(at);
        const wakeDue = continuity.pendingWakeReasons.length > 0
          && (!continuity.wakeNotBefore || Date.parse(continuity.wakeNotBefore) <= Date.parse(at));
        const idleBypass = idleDue && policy.idleMayBypassMinSpacing;
        return continuity.resumeAfterCurrentTurn || (policy.mode === 'auto' && (idleDue || wakeDue || idleBypass));
      })
      .map((worker) => worker.workerId));
  }

  async recordExecutionError(workerId: WorkerId, input: { code: string; detail: string }): Promise<WorkerContinuityView> {
    const runId = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      if (isTerminalWorker(worker)) return worker.runId;
      const timestamp = now();
      const outcome: ExecutionAttemptOutcome = input.code === 'HOST_UI_CHANGED'
        ? 'host_ui_changed'
        : input.code === 'AMBIGUOUS_SUBMISSION'
          ? 'ambiguous_submission'
          : input.code === 'EXECUTOR_UNAVAILABLE'
            ? 'executor_unavailable'
            : 'failed';
      this.finishAttemptInState(state, worker, 'failed', outcome, timestamp, input.code, input.detail);
      const continuity = this.continuityFor(state, worker);
      continuity.disposition = 'paused';
      continuity.pauseReason = 'error';
      continuity.pauseAfterCurrentTurn = false;
      continuity.resumeAfterCurrentTurn = false;
      continuity.pendingExecutionId = null;
      continuity.idleGraceDueAt = null;
      continuity.updatedAt = timestamp;
      this.appendEvent(state, {
        runId: worker.runId,
        type: 'execution.paused',
        workerId,
        payload: { reason: 'error', errorCode: input.code, errorDetail: input.detail },
      });
      return worker.runId;
    });
    this.notifier.notify(runId);
    return this.getWorkerContinuity(workerId);
  }

  async recordDeliveryTimeout(workerId: WorkerId): Promise<ExecutionAttempt | null> {
    const result = await this.repository.mutate((state) => {
      const worker = this.workerOrThrow(state, workerId);
      if (isTerminalWorker(worker)) throw new OrchestratorError('INVALID_STATE', 'Terminal worker cannot recover a delivery timeout.');
      const timestamp = now();
      const latest = this.latestAttemptInState(state, workerId);
      if (latest && latest.outcome !== 'message_delivery_timeout') {
        this.finishAttemptInState(state, worker, 'failed', 'message_delivery_timeout', timestamp, 'MESSAGE_DELIVERY_TIMEOUT', 'Message delivery timed out.');
      }
      const continuity = this.continuityFor(state, worker);
      continuity.pendingExecutionId = null;
      continuity.idleGraceDueAt = timestamp;
      continuity.updatedAt = timestamp;
      if (['awaiting_human', 'awaiting_dependency', 'paused'].includes(continuity.disposition)) {
        return { attempt: null, runId: worker.runId, changed: true };
      }
      continuity.disposition = 'awaiting_execution';
      const policy = effectiveContinuationPolicy(state, workerId);
      if (policy.mode !== 'auto') return { attempt: null, runId: worker.runId, changed: true };
      const prior = this.latestAttemptInState(state, workerId);
      const scheduled = this.scheduleAttemptInState(
        state,
        worker,
        'recovery',
        `execution.delivery-timeout:${workerId}:${prior?.executionId ?? 'none'}`,
        prior?.executionId ?? null,
        timestamp,
      );
      return { attempt: scheduled.attempt, runId: worker.runId, changed: true };
    });
    if (result.changed) this.notifier.notify(result.runId);
    return result.attempt;
  }

  async getConversationBinding(workerId: WorkerId): Promise<ConversationBinding | null> {
    return this.repository.read((state) => state.conversations[workerId] ?? null);
  }

}