长程任务恢复

export interface DurableAgentState {  runId: string;  task: string;  cwd: string;  status: AgentState["status"];  plan: TaskPlan;  activeStepId?: string | undefined;  planHistory: AgentState["planHistory"];  messages: AgentMessage[];  contextSources: PersistedContextSource[];  skills: PersistedSkillState;  compaction: CompactionState;  changedFiles: string[];  changedFileHashes: Record<string, string>;  mutationRevision: number;  validations: ValidationRecord[];  requiredCriterionIds: string[];  budget: AgentBudget;  pendingUserInput?: UserInputRequest | undefined;  goalVersion: number;  constraints: string[];  failedAttempts: string[];  events: AgentEvent[];  nextEventSequence: number;  stopReason: string | undefined;  providerCursor?:    | { provider: string; model: string; previousResponseId: string }    | undefined;}export interface AgentRuntime {  model: ModelDriver & {    canResume?(cursor: DurableAgentState["providerCursor"]): Promise<boolean>;  };  planner: Planner;  tools: ToolRegistry;  trace: TraceSink;  store: RunStore;  clock: Clock;  skillsDirectory: string;  toolTimeoutMs: number;  signal?: AbortSignal;}

export type DurableEvent =  | { type: "run_started"; task: string }  | { type: "plan_updated"; plan: TaskPlan }  | { type: "model_completed"; responseId: string; turn: PersistedModelTurn }  | { type: "tool_intent"; call: PersistedToolCall; idempotencyKey: string }  | { type: "tool_result"; callId: string; idempotencyKey: string; output: string }  | { type: "compaction_completed"; snapshot: CompactionSnapshot }  | { type: "run_waiting"; reason: string }  | { type: "run_stopped"; reason: StopReason };export interface EventRecord {  schemaVersion: 1;  runId: string;  sequence: number;  eventId: string;  recordedAt: string;  event: DurableEvent;  checksum: string;}export interface RunCheckpoint {  schemaVersion: 1;  runId: string;  throughSequence: number;  savedAt: string;  state: DurableAgentState;  checksum: string;}

export interface RunStore {  append(input: {    runId: string;    expectedSequence: number;    ownerId: string;    epoch: number;    event: DurableEvent;  }): Promise<EventRecord>;  loadCheckpointHistory(runId: string, limit: number): Promise<RunCheckpoint[]>;  saveCheckpoint(checkpoint: RunCheckpoint, lease: RunLease): Promise<void>;  readEvents(runId: string, afterSequence: number): Promise<EventRecord[]>;  acquireLease(runId: string, ownerId: string, ttlMs: number): Promise<RunLease>;  renewLease(lease: RunLease, ttlMs: number): Promise<RunLease>;  releaseLease(lease: RunLease): Promise<void>;}

export async function atomicWriteJson(path: string, value: unknown): Promise<void> {  const directory = dirname(path);  const temporary = join(directory, `.checkpoint-${crypto.randomUUID()}.tmp`);  const handle = await open(temporary, "wx", 0o600);  try {    await handle.writeFile(`${canonicalJson(value)}\n`, "utf8");    await handle.sync(); // fsync file contents  } finally {    await handle.close();  }  await rename(temporary, path);  const directoryHandle = await open(directory, "r");  try {    await directoryHandle.sync(); // fsync directory entry  } finally {    await directoryHandle.close();  }}

export async function executeDurably(options: {  state: DurableAgentState;  call: PersistedToolCall;  tool: ToolDefinition;  store: RunStore;  lease: RunLease;}): Promise<string> {  const idempotencyKey = `${options.state.runId}:${options.call.callId}`;  await options.store.append({    runId: options.state.runId,    expectedSequence: nextSequence(options.state),    ownerId: options.lease.ownerId,    epoch: options.lease.epoch,    event: { type: "tool_intent", call: options.call, idempotencyKey },  });  const output = await options.tool.execute(options.call.arguments, { idempotencyKey });  await options.store.append({    runId: options.state.runId,    expectedSequence: nextSequence(options.state),    ownerId: options.lease.ownerId,    epoch: options.lease.epoch,    event: { type: "tool_result", callId: options.call.callId, idempotencyKey, output },  });  return output;}

export type RecoveryPolicy =  | { kind: "replay_safe" }  | { kind: "idempotent"; keyLocation: "header" | "argument" }  | { kind: "verify_then_continue"; verify: ToolStateVerifier }  | { kind: "manual_reconciliation" };

export async function restoreRun(options: {  runId: string;  ownerId: string;  store: RunStore;}): Promise<{ state: DurableAgentState; lease: RunLease; inFlight: PersistedToolCall[] }> {  const lease = await options.store.acquireLease(options.runId, options.ownerId, 30_000);  const checkpointHistory = await options.store.loadCheckpointHistory(options.runId, 2);  const checkpoint = firstValidCheckpoint(checkpointHistory);  if (!checkpoint) throw new Error("no_valid_checkpoint");  let state = migrateCheckpoint(checkpoint).state;  const eventsThroughCheckpoint = await options.store.readEvents(options.runId, 0);  const records = await options.store.readEvents(options.runId, checkpoint.throughSequence);  assertContiguousSequences(records, checkpoint.throughSequence + 1);  for (const record of records) {    verifyEventChecksum(record);    state = reduceDurableEvent(state, record.event);  }  return {    state,    lease,    inFlight: findToolIntentsWithoutResults([      ...eventsThroughCheckpoint.filter((record) => record.sequence <= checkpoint.throughSequence),      ...records,    ]),  };}

export async function resumeAgentRun(  runId: string,  runtime: AgentRuntime,): Promise<AgentResult> {  const restored = await restoreRun({    runId,    ownerId: crypto.randomUUID(),    store: runtime.store,  });  startLeaseHeartbeat(restored.lease, runtime.clock, runtime.signal);  for (const call of restored.inFlight) {    await reconcileInFlightTool(call, restored.state, runtime);  }  const canReuseProviderCursor = await runtime.model.canResume?.(    restored.state.providerCursor,  );  if (!canReuseProviderCursor) restored.state.providerCursor = undefined;  return runAgentLoopFromState(restored.state, runtime);}

const [command, value] = process.argv.slice(2);if (command === "run") {  await startAgentRun(value, runtime);} else if (command === "resume") {  await resumeAgentRun(value, runtime);} else {  throw new Error("usage: agent run <task> | agent resume <run-id>");}

it.each([  "after_tool_intent",  "after_tool_effect",  "after_tool_result",  "before_checkpoint_rename",  "after_checkpoint_rename",])("recovers safely from %s", async (crashPoint) => {  const fixture = createCrashFixture(crashPoint);  await expect(fixture.start()).rejects.toThrow("simulated_crash");  const result = await resumeAgentRun(fixture.runId, fixture.restartedRuntime());  expect(result.status).toBe("completed");  expect(fixture.externalEffectCount()).toBe(1);  expect(fixture.eventSequences()).toEqual(fixture.contiguousSequences());});