{"version":3,"file":"durable-coordinator.d.ts","sourceRoot":"","sources":["../../../src/core/mission-domain/durable-coordinator.ts"],"names":[],"mappings":"AAAA;;;;;;;;;;;;;;;;;;;;;;;;GAwBG;AAGH,OAAO,EAKN,KAAK,oBAAoB,EACzB,KAAK,mBAAmB,EACxB,MAAM,oBAAoB,CAAC;AAC5B,OAAO,EAKN,KAAK,kBAAkB,EACvB,KAAK,kBAAkB,EAGvB,MAAM,0BAA0B,CAAC;AAClC,OAAO,EAEN,KAAK,cAAc,EACnB,KAAK,mBAAmB,EAIxB,MAAM,sBAAsB,CAAC;AAC9B,OAAO,KAAK,EAEX,wBAAwB,EACxB,eAAe,EACf,oBAAoB,EACpB,MAAM,uBAAuB,CAAC;AAE/B,OAAO,KAAK,EAAE,cAAc,EAAE,MAAM,sBAAsB,CAAC;AAG3D,OAAO,EAAmD,KAAK,YAAY,EAAE,MAAM,oBAAoB,CAAC;AAExG,MAAM,WAAW,gCAAgC;IAChD,GAAG,CAAC,EAAE,MAAM,MAAM,CAAC;IACnB;;;OAGG;IACH,gBAAgB,CAAC,EAAE,MAAM,MAAM,CAAC;IAChC;;;;OAIG;IACH,OAAO,CAAC,EAAE,MAAM,CAAC;IACjB,uEAAuE;IACvE,eAAe,CAAC,EAAE,MAAM,CAAC;IACzB,sCAAsC;IACtC,cAAc,CAAC,EAAE,MAAM,MAAM,CAAC;IAC9B,4EAA4E;IAC5E,mBAAmB,CAAC,EAAE,MAAM,CAAC;IAC7B,0EAA0E;IAC1E,qBAAqB,CAAC,EAAE,MAAM,CAAC;IAC/B,oEAAoE;IACpE,kBAAkB,CAAC,EAAE,kBAAkB,CAAC;IACxC,qEAAqE;IACrE,YAAY,CAAC,EAAE,MAAM,CAAC;IACtB,4FAA4F;IAC5F,iBAAiB,CAAC,EAAE,wBAAwB,CAAC;CAC7C;AAED,MAAM,WAAW,qBAAqB;IACrC,OAAO,EAAE,MAAM,CAAC;IAChB,iEAAiE;IACjE,UAAU,EAAE,MAAM,EAAE,CAAC;IACrB,iEAAiE;IACjE,gBAAgB,EAAE,MAAM,EAAE,CAAC;IAC3B,mDAAmD;IACnD,SAAS,EAAE,MAAM,EAAE,CAAC;IACpB,sEAAsE;IACtE,OAAO,EAAE;QAAE,SAAS,EAAE,MAAM,CAAC;QAAC,UAAU,EAAE,MAAM,CAAA;KAAE,EAAE,CAAC;IACrD,uCAAuC;IACvC,OAAO,EAAE,MAAM,EAAE,CAAC;CAClB;AA4CD,2FAA2F;AAC3F,MAAM,WAAW,0BAA0B;IAC1C,MAAM,EAAE,oBAAoB,CAAC;IAC7B,KAAK,EAAE,cAAc,CAAC;CACtB;AAED,qBAAa,yBAAyB;IACrC,OAAO,CAAC,QAAQ,CAAC,MAAM,CAAsB;IAC7C,OAAO,CAAC,QAAQ,CAAC,SAAS,CAAkB;IAC5C,OAAO,CAAC,QAAQ,CAAC,IAAI,CAAe;IACpC,OAAO,CAAC,QAAQ,CAAC,iBAAiB,CAAe;IACjD,OAAO,CAAC,QAAQ,CAAC,QAAQ,CAAS;IAClC,OAAO,CAAC,QAAQ,CAAC,gBAAgB,CAAS;IAC1C,OAAO,CAAC,QAAQ,CAAC,eAAe,CAAe;IAC/C,OAAO,CAAC,QAAQ,CAAC,gBAAgB,CAA0B;IAC3D,OAAO,CAAC,QAAQ,CAAC,mBAAmB,CAAqB;IACzD,OAAO,CAAC,QAAQ,CAAC,kBAAkB,CAAC,CAA2B;IAC/D,OAAO,CAAC,QAAQ,CAAC,aAAa,CAAC,CAAS;IACxC,OAAO,CAAC,QAAQ,CAAC,WAAW,CAAyC;IACrE,OAAO,CAAC,QAAQ,CAAC,iBAAiB,CAAsC;IAExE,YAAY,KAAK,EAAE,mBAAmB,EAAE,QAAQ,EAAE,eAAe,EAAE,OAAO,GAAE,gCAAqC,EAgBhH;IAED,0EAA0E;IAC1E,IAAI,KAAK,IAAI,mBAAmB,CAE/B;IAED,IAAI,QAAQ,IAAI,eAAe,CAE9B;IAED,iEAAiE;IACjE,IAAI,OAAO,IAAI,MAAM,CAEpB;IAED;;;;;OAKG;IACH,kBAAkB,CAAC,SAAS,EAAE,MAAM,GAAG,kBAAkB,GAAG,SAAS,CAEpE;IAED;;;;;OAKG;IACG,mBAAmB,CACxB,SAAS,EAAE,MAAM,EACjB,MAAM,CAAC,EAAE,MAAM,GACb,OAAO,CAAC;QAAE,MAAM,EAAE,kBAAkB,GAAG,aAAa,CAAC;QAAC,SAAS,EAAE,MAAM,CAAA;KAAE,CAAC,CAK5E;IAMD;;;;;;OAMG;IACG,aAAa,CAAC,OAAO,EAAE,cAAc,GAAG,OAAO,CAAC,oBAAoB,CAAC,CAU1E;IAMD,0EAA0E;IACpE,UAAU,CAAC,SAAS,EAAE,MAAM,GAAG,OAAO,CAAC,oBAAoB,GAAG,SAAS,CAAC,CAK7E;IAEK,YAAY,IAAI,OAAO,CAAC,MAAM,EAAE,CAAC,CAEtC;IAEK,uBAAuB,IAAI,OAAO,CAAC,MAAM,EAAE,CAAC,CAEjD;IAEK,YAAY,CAAC,eAAe,EAAE,MAAM,GAAG,OAAO,CAAC,MAAM,EAAE,CAAC,CAE7D;IAMD;;;;;;;OAOG;IACG,gBAAgB,CAAC,SAAS,EAAE,MAAM,GAAG,OAAO,CAAC,0BAA0B,CAAC,CAkE7E;IAED;;;OAGG;IACG,cAAc,CACnB,SAAS,EAAE,MAAM,EACjB,KAAK,EAAE,mBAAmB,EAC1B,OAAO,GAAE;QAAE,GAAG,CAAC,EAAE,MAAM,CAAA;KAAO,GAC5B,OAAO,CAAC;QAAE,KAAK,EAAE,cAAc,CAAC;QAAC,MAAM,EAAE,oBAAoB,CAAA;KAAE,CAAC,CAuBlE;IAED;;;;;OAKG;IACG,gBAAgB,CACrB,SAAS,EAAE,MAAM,EACjB,KAAK,EAAE,mBAAmB,EAC1B,OAAO,GAAE;QAAE,GAAG,CAAC,EAAE,MAAM,CAAA;KAAO,GAC5B,OAAO,CAAC,oBAAoB,CAAC,CAoC/B;IAMD;;;;;;;;OAQG;IACG,OAAO,CAAC,OAAO,GAAE;QAAE,GAAG,CAAC,EAAE,MAAM,CAAA;KAAO,GAAG,OAAO,CAAC,qBAAqB,CAAC,CAwE5E;IAED;;;;;;;;;;OAUG;IACG,uBAAuB,CAC5B,SAAS,EAAE,MAAM,EACjB,OAAO,EAAE;QAAE,YAAY,EAAE,MAAM,CAAC;QAAC,MAAM,CAAC,EAAE,MAAM,CAAC;QAAC,GAAG,CAAC,EAAE,MAAM,CAAA;KAAE,GAC9D,OAAO,CAAC;QAAE,MAAM,EAAE,YAAY,GAAG,WAAW,GAAG,SAAS,CAAC;QAAC,aAAa,CAAC,EAAE,YAAY,CAAA;KAAE,CAAC,CAgD3F;IAMD;;;;;;;OAOG;IACG,MAAM,CAAC,SAAS,EAAE,MAAM,EAAE,OAAO,GAAE,oBAAyB,GAAG,OAAO,CAAC,oBAAoB,CAAC,CAyMjG;IAED;;;;;OAKG;IACH,OAAO,CAAC,WAAW;IA6CnB,OAAO,CAAC,mBAAmB;IAQ3B,OAAO,CAAC,eAAe;IAoBvB,OAAO,CAAC,eAAe;YAkCT,aAAa;IA4C3B,OAAO,CAAC,eAAe;YA8BT,kBAAkB;CAoDhC","sourcesContent":["/**\n * Durable Mission Coordinator (2.4.0, ownership 2.7.0).\n *\n * Composes the canonical mission lifecycle with a DurableMissionStore at the\n * mission execution boundary. The parent/domain layer never remembers to write\n * files; it talks to this coordinator, and the coordinator persists every\n * authoritative lifecycle transition before and after the executor runs.\n *\n * Execution ownership:\n *   - Resume is explicit and always allocates a NEW execution attempt.\n *   - Before any executor work begins, the coordinator atomically acquires a\n *     first-class execution lease with a monotonically increasing fencing\n *     token. Two processes racing to resume the same mission therefore have\n *     exactly one winner; the loser receives a structured MISSION_OWNED error.\n *   - Every execution-authoritative mutation carries lease proof\n *     (`leaseId` + `fencingToken`); the store rejects a fenced owner.\n *   - Terminal transition clears the lease atomically in the same write, so a\n *     completed mission can never be modified by its former owner.\n *\n * Design constraints:\n *   - The coordinator depends only on the injected `DurableMissionStore` port\n *     and a `MissionExecutor` seam — never on process/provider/CLI/UI code.\n *   - Opening/recovering the store NEVER auto-runs mission work.\n *   - Recovery revokes only expired (dead) leases; it never steals a live lease.\n */\n\nimport { randomUUID } from \"node:crypto\";\nimport {\n\tcreateDurableMissionRecord,\n\ttype DurableExecutionAttempt,\n\ttype DurableExecutionAttemptEndReason,\n\ttype DurableMissionMutateResult,\n\ttype DurableMissionRecord,\n\ttype DurableMissionStore,\n} from \"./durable-store.js\";\nimport {\n\tdefaultHeartbeatScheduler,\n\tExecutionAuthorityLostError,\n\tExecutionHeartbeat,\n\ttype HeartbeatAuthorityLossInfo,\n\ttype HeartbeatScheduler,\n\ttype HeartbeatTelemetry,\n\ttype ResolvedHeartbeatTiming,\n\tresolveHeartbeatTiming,\n} from \"./execution-heartbeat.js\";\nimport {\n\tDEFAULT_EXECUTION_LEASE_DURATION_MS,\n\ttype ExecutionLease,\n\ttype ExecutionLeaseProof,\n\tExecutionOwnershipError,\n\tisExecutionLeaseActive,\n\tnewExecutorOwnerId,\n} from \"./execution-lease.js\";\nimport type {\n\tMissionExecutionEvent,\n\tMissionExecutionObserver,\n\tMissionExecutor,\n\tMissionLaunchOptions,\n} from \"./mission-executor.js\";\nimport type { MissionHandle } from \"./mission-handle.js\";\nimport type { MissionRequest } from \"./mission-request.js\";\nimport { validateMissionRequest } from \"./mission-request.js\";\nimport { createMissionResult, type MissionResult } from \"./mission-result.js\";\nimport { assertMissionTransition, isTerminalMissionState, type MissionState } from \"./mission-state.js\";\n\nexport interface DurableMissionCoordinatorOptions {\n\tnow?: () => number;\n\t/**\n\t * Durable attempt identity factory. Defaults to a UUID-based id. It is\n\t * coordinator-scoped and distinct from the executor's `executionId`.\n\t */\n\tattemptIdFactory?: () => string;\n\t/**\n\t * Executor owner identity. Defaults to a fresh host+UUID identity per\n\t * coordinator instance (never PID-derived). Pass one stable value per\n\t * executor/process lifetime for diagnostics.\n\t */\n\townerId?: string;\n\t/** Execution lease lifetime. Defaults to a conservative 30 minutes. */\n\tleaseDurationMs?: number;\n\t/** Lease identity factory (tests). */\n\tleaseIdFactory?: () => string;\n\t/** Heartbeat renewal cadence override (defaults to leaseDurationMs / 3). */\n\theartbeatIntervalMs?: number;\n\t/** Heartbeat safety margin override (defaults to leaseDurationMs / 6). */\n\trenewalSafetyMarginMs?: number;\n\t/** Injectable timer scheduler for deterministic heartbeat tests. */\n\theartbeatScheduler?: HeartbeatScheduler;\n\t/** Assignment identity used for execution/Governance attribution. */\n\tassignmentId?: string;\n\t/** Observer for actual attempt/execution lifecycle events; recovery/polling is excluded. */\n\texecutionObserver?: MissionExecutionObserver;\n}\n\nexport interface DurableRecoveryReport {\n\tscanned: number;\n\t/** Missions transitioned to INTERRUPTED during this recovery. */\n\treconciled: string[];\n\t/** Missions already INTERRUPTED (no additional action taken). */\n\talreadyRecovered: string[];\n\t/** Terminal or CREATED missions left unchanged. */\n\tunchanged: string[];\n\t/** Corrupt records surfaced structurally (never silently dropped). */\n\tcorrupt: { missionId: string; diagnostic: string }[];\n\t/** Human-readable recovery actions. */\n\tactions: string[];\n}\n\nconst ACTIVE_NONTERMINAL_STATES: ReadonlySet<MissionState> = new Set<MissionState>([\n\t\"QUEUED\",\n\t\"LAUNCHING\",\n\t\"RUNNING\",\n\t\"WAITING\",\n\t\"BLOCKED\",\n\t\"RETRYING\",\n]);\n\ntype AcquireOutcome =\n\t| { status: \"not_resumable\"; state: MissionState }\n\t| { status: \"owned\"; lease: ExecutionLease }\n\t| { status: \"acquired\"; lease: ExecutionLease; record: DurableMissionRecord };\n\ntype RenewOutcome =\n\t| { status: \"lease_not_found\" }\n\t| { status: \"stale_owner\"; lease: ExecutionLease }\n\t| { status: \"lease_expired\"; lease: ExecutionLease }\n\t| { status: \"renewed\"; lease: ExecutionLease; record: DurableMissionRecord };\n\ntype ReleaseOutcome =\n\t| { status: \"lease_not_found\" }\n\t| { status: \"stale_owner\"; lease: ExecutionLease }\n\t| { status: \"released\"; record: DurableMissionRecord };\n\ntype RevokeOutcome = { status: \"unchanged\" } | { status: \"reconciled\"; previousState: MissionState };\n\n/** Local live execution retained for operator cancellation. */\ninterface ActiveExecution {\n\tcancelController: AbortController;\n\thandle?: MissionHandle;\n}\n\n/**\n * Map a terminal MissionResult's executor-level outcome to an attempt end\n * reason. `MissionExecutionOutcome` is a strict subset of\n * `DurableExecutionAttemptEndReason` (INTERRUPTED is never a terminal result).\n */\nfunction attemptEndReason(result: MissionResult): DurableExecutionAttemptEndReason {\n\treturn result.executionOutcome;\n}\n\n/** The authoritative lease acquired for one mission, plus the record after acquisition. */\nexport interface AcquiredExecutionOwnership {\n\trecord: DurableMissionRecord;\n\tlease: ExecutionLease;\n}\n\nexport class DurableMissionCoordinator {\n\tprivate readonly _store: DurableMissionStore;\n\tprivate readonly _executor: MissionExecutor;\n\tprivate readonly _now: () => number;\n\tprivate readonly _attemptIdFactory: () => string;\n\tprivate readonly _ownerId: string;\n\tprivate readonly _leaseDurationMs: number;\n\tprivate readonly _leaseIdFactory: () => string;\n\tprivate readonly _heartbeatTiming: ResolvedHeartbeatTiming;\n\tprivate readonly _heartbeatScheduler: HeartbeatScheduler;\n\tprivate readonly _executionObserver?: MissionExecutionObserver;\n\tprivate readonly _assignmentId?: string;\n\tprivate readonly _heartbeats = new Map<string, ExecutionHeartbeat>();\n\tprivate readonly _activeExecutions = new Map<string, ActiveExecution>();\n\n\tconstructor(store: DurableMissionStore, executor: MissionExecutor, options: DurableMissionCoordinatorOptions = {}) {\n\t\tthis._store = store;\n\t\tthis._executor = executor;\n\t\tthis._now = options.now ?? (() => Date.now());\n\t\tthis._attemptIdFactory = options.attemptIdFactory ?? (() => `attempt_${randomUUID()}`);\n\t\tthis._ownerId = options.ownerId ?? newExecutorOwnerId();\n\t\tthis._leaseDurationMs = options.leaseDurationMs ?? DEFAULT_EXECUTION_LEASE_DURATION_MS;\n\t\tthis._leaseIdFactory = options.leaseIdFactory ?? (() => `lease_${randomUUID()}`);\n\t\tthis._heartbeatTiming = resolveHeartbeatTiming({\n\t\t\tleaseDurationMs: this._leaseDurationMs,\n\t\t\theartbeatIntervalMs: options.heartbeatIntervalMs,\n\t\t\trenewalSafetyMarginMs: options.renewalSafetyMarginMs,\n\t\t});\n\t\tthis._heartbeatScheduler = options.heartbeatScheduler ?? defaultHeartbeatScheduler;\n\t\tthis._executionObserver = options.executionObserver;\n\t\tthis._assignmentId = options.assignmentId;\n\t}\n\n\t/** The underlying persistence port (exposed for tests and load paths). */\n\tget store(): DurableMissionStore {\n\t\treturn this._store;\n\t}\n\n\tget executor(): MissionExecutor {\n\t\treturn this._executor;\n\t}\n\n\t/** Stable executor owner identity used for lease acquisition. */\n\tget ownerId(): string {\n\t\treturn this._ownerId;\n\t}\n\n\t/**\n\t * Latest heartbeat telemetry for a mission this coordinator executed. The\n\t * heartbeat object (and thus its telemetry) survives stop so callers can\n\t * inspect renewal/loss state after completion. Returns `undefined` before\n\t * the coordinator has ever run the mission.\n\t */\n\theartbeatTelemetry(missionId: string): HeartbeatTelemetry | undefined {\n\t\treturn this._heartbeats.get(missionId)?.telemetry();\n\t}\n\n\t/**\n\t * Request cancellation of a locally-active execution. This aborts the same\n\t * combined signal the executor received, so the normal fenced terminal path\n\t * persists CANCELLED. It never signals a remote/other-process owner and\n\t * never bypasses fencing.\n\t */\n\tasync requestCancellation(\n\t\tmissionId: string,\n\t\treason?: string,\n\t): Promise<{ status: \"cancel_requested\" | \"not_running\"; missionId: string }> {\n\t\tconst active = this._activeExecutions.get(missionId);\n\t\tif (!active) return { status: \"not_running\", missionId };\n\t\tactive.cancelController.abort(reason ?? \"mission cancelled by operator\");\n\t\treturn { status: \"cancel_requested\", missionId };\n\t}\n\n\t// =========================================================================\n\t// Create (durable, does NOT execute)\n\t// =========================================================================\n\n\t/**\n\t * Persist a mission at CREATED state. Does not launch anything.\n\t *\n\t * Re-creating the same missionId with an identical immutable request is\n\t * idempotent; re-creating with a different request is rejected structurally\n\t * (history is never silently overwritten).\n\t */\n\tasync createMission(request: MissionRequest): Promise<DurableMissionRecord> {\n\t\tconst validation = validateMissionRequest(request);\n\t\tif (!validation.valid) {\n\t\t\tthrow new Error(`Invalid MissionRequest: ${validation.errors.join(\", \")}`);\n\t\t}\n\t\tconst record = createDurableMissionRecord({ request, now: this._now() });\n\t\tconst result = await this._store.create(record);\n\t\tif (result.status === \"created\") return record;\n\t\tif (result.status === \"idempotent\") return result.record;\n\t\tthrow new Error(`Mission ${request.missionId} already exists with a conflicting request: ${result.error}`);\n\t}\n\n\t// =========================================================================\n\t// Read\n\t// =========================================================================\n\n\t/** Load a mission, or `undefined` when missing. Corrupt records throw. */\n\tasync getMission(missionId: string): Promise<DurableMissionRecord | undefined> {\n\t\tconst loaded = await this._store.load(missionId);\n\t\tif (loaded.status === \"ok\") return loaded.record;\n\t\tif (loaded.status === \"missing\") return undefined;\n\t\tthrow new Error(`Mission ${missionId} is corrupt: ${loaded.diagnostic}`);\n\t}\n\n\tasync listMissions(): Promise<string[]> {\n\t\treturn this._store.listMissions();\n\t}\n\n\tasync listNonterminalMissions(): Promise<string[]> {\n\t\treturn this._store.listNonterminalMissions();\n\t}\n\n\tasync listChildren(parentMissionId: string): Promise<string[]> {\n\t\treturn this._store.listChildren(parentMissionId);\n\t}\n\n\t// =========================================================================\n\t// Execution ownership (acquire / renew / release)\n\t// =========================================================================\n\n\t/**\n\t * Atomically acquire execution ownership for a CREATED or INTERRUPTED\n\t * mission. Transitions the mission to QUEUED and persists a new execution\n\t * lease with a strictly greater fencing token. Exactly one of any set of\n\t * racing cross-process callers wins; losers get structured errors.\n\t *\n\t * Does NOT invoke the executor.\n\t */\n\tasync acquireOwnership(missionId: string): Promise<AcquiredExecutionOwnership> {\n\t\tconst now = this._now();\n\t\tconst leaseId = this._leaseIdFactory();\n\n\t\tconst result = await this._store.mutate<AcquireOutcome>(missionId, (current) => {\n\t\t\tif (isTerminalMissionState(current.state)) {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"not_resumable\" as const, state: current.state } };\n\t\t\t}\n\t\t\t// A live owner exists regardless of which active non-terminal state it\n\t\t\t// has reached. This is the authoritative \"owned\" signal.\n\t\t\tif (current.lease && isExecutionLeaseActive(current.lease, now)) {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"owned\" as const, lease: current.lease } };\n\t\t\t}\n\t\t\tif (current.state !== \"CREATED\" && current.state !== \"INTERRUPTED\") {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"not_resumable\" as const, state: current.state } };\n\t\t\t}\n\n\t\t\tconst fencingToken = current.fencingToken + 1;\n\t\t\tconst lease: ExecutionLease = {\n\t\t\t\townerId: this._ownerId,\n\t\t\t\tleaseId,\n\t\t\t\tfencingToken,\n\t\t\t\tacquiredAtMs: now,\n\t\t\t\trenewedAtMs: now,\n\t\t\t\texpiresAtMs: now + this._leaseDurationMs,\n\t\t\t};\n\t\t\tconst next: DurableMissionRecord = {\n\t\t\t\t...this._withTransition(current, \"QUEUED\", {\n\t\t\t\t\treason: current.state === \"INTERRUPTED\" ? \"explicit resume (new execution attempt)\" : \"initial launch\",\n\t\t\t\t\tatMs: now,\n\t\t\t\t}),\n\t\t\t\tfencingToken,\n\t\t\t\tlease,\n\t\t\t};\n\t\t\treturn { kind: \"write\", next, value: { status: \"acquired\" as const, lease, record: next } };\n\t\t});\n\n\t\tif (result.status === \"missing\") throw new Error(`Mission not found: ${missionId}`);\n\t\tif (result.status === \"corrupt\") throw new Error(`Mission ${missionId} is corrupt: ${result.diagnostic}`);\n\n\t\tconst value = result.value;\n\t\tif (value.status === \"acquired\") return { record: value.record, lease: value.lease };\n\t\tif (value.status === \"owned\") {\n\t\t\tthrow new ExecutionOwnershipError(\n\t\t\t\t\"MISSION_OWNED\",\n\t\t\t\t`Mission ${missionId} already has an active execution owner`,\n\t\t\t\t{\n\t\t\t\t\tmissionId,\n\t\t\t\t\townerId: value.lease.ownerId,\n\t\t\t\t\tleaseId: value.lease.leaseId,\n\t\t\t\t\tfencingToken: value.lease.fencingToken,\n\t\t\t\t},\n\t\t\t);\n\t\t}\n\t\tif (isTerminalMissionState(value.state)) {\n\t\t\tthrow new ExecutionOwnershipError(\n\t\t\t\t\"MISSION_TERMINAL\",\n\t\t\t\t`Cannot resume terminal mission ${missionId} (${value.state})`,\n\t\t\t\t{ missionId, state: value.state },\n\t\t\t);\n\t\t}\n\t\tthrow new ExecutionOwnershipError(\n\t\t\t\"MISSION_NOT_RESUMABLE\",\n\t\t\t`Cannot resume mission ${missionId} from state ${value.state}; reconcile (recover) first`,\n\t\t\t{ missionId, state: value.state },\n\t\t);\n\t}\n\n\t/**\n\t * Renew a live lease. Requires current lease proof and does NOT change the\n\t * fencing token (renewal is a heartbeat, not a takeover).\n\t */\n\tasync renewOwnership(\n\t\tmissionId: string,\n\t\tproof: ExecutionLeaseProof,\n\t\toptions: { now?: number } = {},\n\t): Promise<{ lease: ExecutionLease; record: DurableMissionRecord }> {\n\t\tconst now = options.now ?? this._now();\n\n\t\tconst result = await this._store.mutate<RenewOutcome>(missionId, (current) => {\n\t\t\tconst lease = current.lease;\n\t\t\tif (!lease) return { kind: \"noop\", value: { status: \"lease_not_found\" as const } };\n\t\t\tif (lease.leaseId !== proof.leaseId || lease.fencingToken !== proof.fencingToken) {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"stale_owner\" as const, lease } };\n\t\t\t}\n\t\t\tif (!isExecutionLeaseActive(lease, now)) {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"lease_expired\" as const, lease } };\n\t\t\t}\n\t\t\tconst renewed: ExecutionLease = { ...lease, renewedAtMs: now, expiresAtMs: now + this._leaseDurationMs };\n\t\t\tconst next: DurableMissionRecord = {\n\t\t\t\t...current,\n\t\t\t\tlease: renewed,\n\t\t\t\tupdatedAtMs: now,\n\t\t\t\trevision: current.revision + 1,\n\t\t\t};\n\t\t\treturn { kind: \"write\", next, value: { status: \"renewed\" as const, lease: renewed, record: next } };\n\t\t});\n\n\t\treturn this._mapRenewResult(missionId, result);\n\t}\n\n\t/**\n\t * Release a live lease without reaching a terminal state. This is a clean\n\t * non-terminal stop: any active non-terminal state moves to INTERRUPTED\n\t * (recoverable) and ownership is cleared atomically. Terminal completion\n\t * should instead clear the lease as part of the terminal write.\n\t */\n\tasync releaseOwnership(\n\t\tmissionId: string,\n\t\tproof: ExecutionLeaseProof,\n\t\toptions: { now?: number } = {},\n\t): Promise<DurableMissionRecord> {\n\t\tconst now = options.now ?? this._now();\n\n\t\tconst result = await this._store.mutate<ReleaseOutcome>(missionId, (current) => {\n\t\t\tconst lease = current.lease;\n\t\t\tif (!lease) return { kind: \"noop\", value: { status: \"lease_not_found\" as const } };\n\t\t\tif (lease.leaseId !== proof.leaseId || lease.fencingToken !== proof.fencingToken) {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"stale_owner\" as const, lease } };\n\t\t\t}\n\n\t\t\tlet next: DurableMissionRecord;\n\t\t\tif (isTerminalMissionState(current.state) || current.state === \"CREATED\" || current.state === \"INTERRUPTED\") {\n\t\t\t\tnext = { ...current, lease: undefined, updatedAtMs: now, revision: current.revision + 1 };\n\t\t\t} else {\n\t\t\t\tnext = this._withTransition(current, \"INTERRUPTED\", {\n\t\t\t\t\treason: \"owner released without terminal result\",\n\t\t\t\t\tatMs: now,\n\t\t\t\t});\n\t\t\t\tnext = { ...next, lease: undefined, currentAttemptId: undefined, currentExecutionId: undefined };\n\t\t\t}\n\t\t\treturn { kind: \"write\", next, value: { status: \"released\" as const, record: next } };\n\t\t});\n\n\t\tif (result.status === \"missing\")\n\t\t\tthrow new ExecutionOwnershipError(\"LEASE_NOT_FOUND\", `Mission ${missionId} not found`);\n\t\tif (result.status === \"corrupt\") throw new Error(`Mission ${missionId} is corrupt: ${result.diagnostic}`);\n\t\tconst value = result.value;\n\t\tif (value.status === \"released\") return value.record;\n\t\tif (value.status === \"lease_not_found\") {\n\t\t\tthrow new ExecutionOwnershipError(\"LEASE_NOT_FOUND\", `Mission ${missionId} has no execution lease`);\n\t\t}\n\t\tthrow new ExecutionOwnershipError(\n\t\t\t\"STALE_EXECUTION_OWNER\",\n\t\t\t`Mission ${missionId} lease no longer belongs to this owner`,\n\t\t\t{ missionId, leaseId: value.lease.leaseId, fencingToken: value.lease.fencingToken },\n\t\t);\n\t}\n\n\t// =========================================================================\n\t// Restart reconciliation (no auto re-execution)\n\t// =========================================================================\n\n\t/**\n\t * Inspect every persisted non-terminal mission and reconcile lost executor\n\t * ownership. Persisted active states (QUEUED/RUNNING/WAITING/BLOCKED/\n\t * RETRYING) are never blindly trusted across a restart: when they have no\n\t * live lease they transition to INTERRUPTED with a recovery reason. A live\n\t * (non-expired) lease is never stolen. Terminal missions stay terminal;\n\t * CREATED missions stay CREATED; already-interrupted missions are left\n\t * alone. Nothing is executed.\n\t */\n\tasync recover(options: { now?: number } = {}): Promise<DurableRecoveryReport> {\n\t\tconst now = options.now ?? this._now();\n\t\tconst ids = await this._store.listMissions();\n\t\tconst report: DurableRecoveryReport = {\n\t\t\tscanned: ids.length,\n\t\t\treconciled: [],\n\t\t\talreadyRecovered: [],\n\t\t\tunchanged: [],\n\t\t\tcorrupt: [],\n\t\t\tactions: [],\n\t\t};\n\n\t\tfor (const id of ids) {\n\t\t\tconst loaded = await this._store.load(id);\n\t\t\tif (loaded.status === \"corrupt\") {\n\t\t\t\treport.corrupt.push({ missionId: id, diagnostic: loaded.diagnostic });\n\t\t\t\treport.actions.push(`mission '${id}' is corrupt; surfaced, not recovered`);\n\t\t\t\tcontinue;\n\t\t\t}\n\t\t\tif (loaded.status === \"missing\") continue;\n\n\t\t\tconst record = loaded.record;\n\t\t\tif (isTerminalMissionState(record.state) || record.state === \"CREATED\") {\n\t\t\t\treport.unchanged.push(id);\n\t\t\t\treport.actions.push(`mission '${id}' left at ${record.state}`);\n\t\t\t\tcontinue;\n\t\t\t}\n\t\t\tif (record.state === \"INTERRUPTED\") {\n\t\t\t\treport.alreadyRecovered.push(id);\n\t\t\t\treport.actions.push(`mission '${id}' already INTERRUPTED`);\n\t\t\t\tcontinue;\n\t\t\t}\n\n\t\t\t// Guard: only active nonterminal states may be reconciled here.\n\t\t\tif (!ACTIVE_NONTERMINAL_STATES.has(record.state)) {\n\t\t\t\treport.unchanged.push(id);\n\t\t\t\treport.actions.push(`mission '${id}' left at ${record.state}`);\n\t\t\t\tcontinue;\n\t\t\t}\n\n\t\t\t// A live owner must not be stolen by ordinary recovery.\n\t\t\tif (record.lease && isExecutionLeaseActive(record.lease, now)) {\n\t\t\t\treport.unchanged.push(id);\n\t\t\t\treport.actions.push(`mission '${id}' has a live lease (owner ${record.lease.ownerId}); left unchanged`);\n\t\t\t\tcontinue;\n\t\t\t}\n\n\t\t\t// Honest per-state recovery reason. QUEUED has no attempt and no\n\t\t\t// ownership (launch was never initiated); LAUNCHING has a durable\n\t\t\t// attempt but ownership was never confirmed; RUNNING/WAITING/BLOCKED/\n\t\t\t// RETRYING lost confirmed ownership.\n\t\t\tconst reason =\n\t\t\t\trecord.state === \"LAUNCHING\"\n\t\t\t\t\t? \"control_plane_restart: launch initiated but runtime ownership never confirmed\"\n\t\t\t\t\t: record.state === \"QUEUED\"\n\t\t\t\t\t\t? \"control_plane_restart: queued but never launched\"\n\t\t\t\t\t\t: \"control_plane_restart: executor ownership lost\";\n\n\t\t\tconst revoked = await this._revokeInterrupted(id, now, reason);\n\t\t\tif (revoked.status === \"reconciled\") {\n\t\t\t\treport.reconciled.push(id);\n\t\t\t\treport.actions.push(`mission '${id}' reconciled ${revoked.previousState} → INTERRUPTED`);\n\t\t\t} else if (revoked.status === \"corrupt\") {\n\t\t\t\treport.corrupt.push({ missionId: id, diagnostic: revoked.diagnostic });\n\t\t\t\treport.actions.push(`mission '${id}' became corrupt during recovery; surfaced, not recovered`);\n\t\t\t} else {\n\t\t\t\treport.unchanged.push(id);\n\t\t\t\treport.actions.push(`mission '${id}' left unchanged after atomic re-check`);\n\t\t\t}\n\t\t}\n\n\t\treturn report;\n\t}\n\n\t/**\n\t * Interrupt a stale worker's still-live execution ownership. Unlike\n\t * `recover()` (which only revokes EXPIRED leases), this closes the crash\n\t * window where a prior worker process died but its execution lease has not\n\t * yet expired. The caller must prove the stale owner identity (for example,\n\t * via a persisted assignment's `executionOwnerIdentity.ownerId`).\n\t *\n\t * Exactly like recovery, this NEVER auto-runs work and NEVER fabricates a\n\t * terminal result: it moves an active non-terminal mission to INTERRUPTED\n\t * (recoverable) and clears the lease for a future explicit resume.\n\t */\n\tasync interruptStaleExecution(\n\t\tmissionId: string,\n\t\toptions: { staleOwnerId: string; reason?: string; now?: number },\n\t): Promise<{ status: \"reconciled\" | \"unchanged\" | \"missing\"; previousState?: MissionState }> {\n\t\tconst now = options.now ?? this._now();\n\t\tconst reason = options.reason ?? \"worker restart: prior execution owner is stale\";\n\n\t\tconst result = await this._store.mutate<{\n\t\t\tstatus: \"reconciled\" | \"unchanged\";\n\t\t\tpreviousState: MissionState;\n\t\t}>(missionId, (current) => {\n\t\t\tif (isTerminalMissionState(current.state) || current.state === \"CREATED\" || current.state === \"INTERRUPTED\") {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"unchanged\" as const, previousState: current.state } };\n\t\t\t}\n\t\t\tif (!ACTIVE_NONTERMINAL_STATES.has(current.state)) {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"unchanged\" as const, previousState: current.state } };\n\t\t\t}\n\t\t\t// Only revoke when the current lease belongs to the proven-stale owner.\n\t\t\tif (!current.lease || current.lease.ownerId !== options.staleOwnerId) {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"unchanged\" as const, previousState: current.state } };\n\t\t\t}\n\n\t\t\tconst previousState = current.state;\n\t\t\tconst fencingToken = current.fencingToken + 1;\n\t\t\tlet next = this._withTransition(current, \"INTERRUPTED\", { reason, atMs: now });\n\n\t\t\tconst attempts = [...current.attempts];\n\t\t\tif (current.currentAttemptId) {\n\t\t\t\tconst index = attempts.findIndex((a) => a.attemptId === current.currentAttemptId);\n\t\t\t\tif (index >= 0) {\n\t\t\t\t\tattempts[index] = {\n\t\t\t\t\t\t...attempts[index],\n\t\t\t\t\t\tendReason: \"INTERRUPTED\",\n\t\t\t\t\t\trecovery: { reason, recoveredAtMs: now },\n\t\t\t\t\t};\n\t\t\t\t}\n\t\t\t}\n\t\t\tnext = {\n\t\t\t\t...next,\n\t\t\t\tfencingToken,\n\t\t\t\tlease: undefined,\n\t\t\t\tcurrentAttemptId: undefined,\n\t\t\t\tcurrentExecutionId: undefined,\n\t\t\t\tattempts,\n\t\t\t};\n\t\t\treturn { kind: \"write\", next, value: { status: \"reconciled\" as const, previousState } };\n\t\t});\n\n\t\tif (result.status === \"missing\") return { status: \"missing\" };\n\t\tif (result.status === \"corrupt\") throw new Error(`Mission ${missionId} is corrupt: ${result.diagnostic}`);\n\t\treturn result.value;\n\t}\n\n\t// =========================================================================\n\t// Explicit resume (new execution attempt)\n\t// =========================================================================\n\n\t/**\n\t * Explicitly start a mission (or restart a recovered one) to terminal state.\n\t *\n\t * The logical mission identity (`missionId`, `parentMissionId`, `depth`,\n\t * immutable `request`, and lifecycle history) is preserved, but the executor\n\t * allocates a NEW `executionId` for the attempt. The previous attempt remains\n\t * in `attempts` for auditability and is never overwritten.\n\t */\n\tasync resume(missionId: string, options: MissionLaunchOptions = {}): Promise<DurableMissionRecord> {\n\t\tconst { record: queued, lease } = await this.acquireOwnership(missionId);\n\t\tconst proof: ExecutionLeaseProof = { leaseId: lease.leaseId, fencingToken: lease.fencingToken };\n\t\tconst attemptNumber = queued.attempts.length + 1;\n\t\tlet record = queued;\n\n\t\t// Combine the caller's cancellation signal with the authority-loss signal\n\t\t// produced by the heartbeat and the operator-cancellation signal produced\n\t\t// by `requestCancellation`. All three abort the executor through the same\n\t\t// AbortSignal path, but authority loss keeps a distinct diagnostic reason.\n\t\tconst authorityLost = new AbortController();\n\t\tconst cancelController = new AbortController();\n\t\tconst combinedSignals = [cancelController.signal, authorityLost.signal];\n\t\tif (options.signal) combinedSignals.push(options.signal);\n\t\tconst signal = AbortSignal.any(combinedSignals);\n\t\tlet authorityLostInfo: HeartbeatAuthorityLossInfo | undefined;\n\n\t\tconst heartbeat = this._buildHeartbeat(missionId, proof, (info) => {\n\t\t\tauthorityLostInfo = info;\n\t\t\tauthorityLost.abort(info);\n\t\t});\n\t\tthis._heartbeats.set(missionId, heartbeat);\n\n\t\tconst activeExecution: ActiveExecution = { cancelController };\n\t\tthis._activeExecutions.set(missionId, activeExecution);\n\n\t\tlet handle: MissionHandle | undefined;\n\t\tlet completed = false;\n\n\t\ttry {\n\t\t\t// Allocate the durable attempt identity and persist LAUNCHING BEFORE\n\t\t\t// invoking the executor. If Jensen crashes immediately after the executor\n\t\t\t// actually launches but before RUNNING is persisted, the attempt intent\n\t\t\t// survives durably and restart reconciliation marks it interrupted.\n\t\t\tconst attemptId = this._attemptIdFactory();\n\t\t\tconst launchAtMs = this._now();\n\t\t\tconst correlation = {\n\t\t\t\tmissionId: record.missionId,\n\t\t\t\tassignmentId: this._assignmentId ?? record.missionId,\n\t\t\t\tattemptId,\n\t\t\t\tsessionId: record.request.childSessionId,\n\t\t\t};\n\t\t\tthis._emitExecutionEvent({\n\t\t\t\ttype: \"attempt_started\",\n\t\t\t\teventId: `${record.missionId}:${attemptId}:attempt_started`,\n\t\t\t\tatMs: launchAtMs,\n\t\t\t\tcorrelation,\n\t\t\t});\n\t\t\tif (attemptNumber > 1) {\n\t\t\t\tthis._emitExecutionEvent({\n\t\t\t\t\ttype: \"execution_retry\",\n\t\t\t\t\teventId: `${record.missionId}:${attemptId}:execution_retry:${attemptNumber - 1}`,\n\t\t\t\t\tatMs: launchAtMs,\n\t\t\t\t\tretryClass: \"execution\",\n\t\t\t\t\tretryIndex: attemptNumber - 1,\n\t\t\t\t\treason: \"explicit resume after prior execution attempt\",\n\t\t\t\t\tcorrelation,\n\t\t\t\t});\n\t\t\t}\n\t\t\tthis._emitExecutionEvent({\n\t\t\t\ttype: \"execution_launch_started\",\n\t\t\t\teventId: `${record.missionId}:${attemptId}:execution_launch_started`,\n\t\t\t\tatMs: launchAtMs,\n\t\t\t\tcorrelation,\n\t\t\t});\n\t\t\tlet next = this._withTransition(record, \"LAUNCHING\", {\n\t\t\t\treason: \"launch initiated\",\n\t\t\t\tattemptId,\n\t\t\t\tatMs: launchAtMs,\n\t\t\t});\n\t\t\tnext = {\n\t\t\t\t...next,\n\t\t\t\tcurrentAttemptId: attemptId,\n\t\t\t\tstartedAtMs: record.startedAtMs ?? launchAtMs,\n\t\t\t\tattempts: [...record.attempts, { attemptId, startedAtMs: launchAtMs }],\n\t\t\t};\n\t\t\trecord = await this._commitFenced(record, next, lease);\n\n\t\t\tconst executor = this._executor;\n\t\t\ttry {\n\t\t\t\thandle = await executor.launch(record.request, {\n\t\t\t\t\tsignal,\n\t\t\t\t\tattemptId,\n\t\t\t\t\tattemptNumber,\n\t\t\t\t\tassignmentId: this._assignmentId ?? record.missionId,\n\t\t\t\t\tsessionId: record.request.childSessionId,\n\t\t\t\t\tfencing: { leaseId: lease.leaseId, fencingToken: lease.fencingToken },\n\t\t\t\t});\n\t\t\t\tactiveExecution.handle = handle;\n\t\t\t} catch (error) {\n\t\t\t\tconst message = error instanceof Error ? error.message : String(error);\n\t\t\t\tthis._emitExecutionEvent({\n\t\t\t\t\ttype: \"execution_failed\",\n\t\t\t\t\teventId: `${record.missionId}:${attemptId}:execution_failed`,\n\t\t\t\t\tatMs: this._now(),\n\t\t\t\t\tcorrelation,\n\t\t\t\t\tstate: \"FAILED\",\n\t\t\t\t\texecutionOutcome: \"CRASHED\",\n\t\t\t\t});\n\t\t\t\tconst failed = await this._commitFenced(\n\t\t\t\t\trecord,\n\t\t\t\t\tthis._failLaunch(record, attemptId, message, launchAtMs),\n\t\t\t\t\tlease,\n\t\t\t\t);\n\t\t\t\tcompleted = true;\n\t\t\t\treturn failed;\n\t\t\t}\n\n\t\t\tconst executionId = handle.executionId;\n\t\t\tthis._emitExecutionEvent({\n\t\t\t\ttype: \"execution_started\",\n\t\t\t\teventId: `${record.missionId}:${attemptId}:${executionId}:execution_started`,\n\t\t\t\tatMs: this._now(),\n\t\t\t\tcorrelation: { ...correlation, executionId },\n\t\t\t});\n\t\t\tnext = this._withTransition(record, \"RUNNING\", {\n\t\t\t\treason: \"executor launched\",\n\t\t\t\texecutionId,\n\t\t\t});\n\t\t\tconst launchAttemptIndex = next.attempts.length - 1;\n\t\t\tnext = {\n\t\t\t\t...next,\n\t\t\t\tcurrentAttemptId: attemptId,\n\t\t\t\tcurrentExecutionId: executionId,\n\t\t\t\tstartedAtMs: record.startedAtMs ?? launchAtMs,\n\t\t\t\tattempts: next.attempts.map((a, i) => (i === launchAttemptIndex ? { ...a, executionId } : a)),\n\t\t\t};\n\t\t\trecord = await this._commitFenced(record, next, lease);\n\n\t\t\t// Heartbeat starts only after RUNNING ownership is durably confirmed,\n\t\t\t// so an execution that never actually launched is never renewed.\n\t\t\theartbeat.start(lease.expiresAtMs);\n\n\t\t\tconst result = await executor.awaitResult(handle, { signal });\n\n\t\t\t// If the heartbeat proved authority was lost while we awaited the\n\t\t\t// result, never write a terminal record: the fence (or the lease\n\t\t\t// expiry + recovery path) remains the authority.\n\t\t\tif (authorityLostInfo) {\n\t\t\t\tthis._emitExecutionEvent({\n\t\t\t\t\ttype: \"execution_failed\",\n\t\t\t\t\teventId: `${record.missionId}:${attemptId}:${executionId}:execution_failed`,\n\t\t\t\t\tatMs: this._now(),\n\t\t\t\t\tcorrelation: { ...correlation, executionId },\n\t\t\t\t\tstate: \"FAILED\",\n\t\t\t\t\texecutionOutcome: \"CRASHED\",\n\t\t\t\t});\n\t\t\t\tthrow new ExecutionAuthorityLostError(authorityLostInfo);\n\t\t\t}\n\n\t\t\tnext = this._withTransition(record, result.state, {\n\t\t\t\treason: \"executor result\",\n\t\t\t\texecutionId,\n\t\t\t});\n\t\t\tconst attemptIndex = next.attempts.length - 1;\n\t\t\tconst completedAttempt: DurableExecutionAttempt = {\n\t\t\t\t...next.attempts[attemptIndex],\n\t\t\t\texecutionId,\n\t\t\t\tfinishedAtMs: result.finishedAtMs,\n\t\t\t\tendReason: attemptEndReason(result),\n\t\t\t};\n\t\t\tnext = {\n\t\t\t\t...next,\n\t\t\t\tresult,\n\t\t\t\tresultExecutionId: executionId,\n\t\t\t\tfinishedAtMs: result.finishedAtMs,\n\t\t\t\tcurrentAttemptId: undefined,\n\t\t\t\tcurrentExecutionId: undefined,\n\t\t\t\tlease: undefined,\n\t\t\t\tattempts: [...next.attempts.slice(0, attemptIndex), completedAttempt],\n\t\t\t};\n\n\t\t\tconst terminal = await this._commitFenced(record, next, lease);\n\t\t\tthis._emitExecutionEvent({\n\t\t\t\ttype:\n\t\t\t\t\tresult.state === \"CANCELLED\"\n\t\t\t\t\t\t? \"execution_cancelled\"\n\t\t\t\t\t\t: result.state === \"FAILED\" || result.state === \"CRASHED\" || result.state === \"TIMED_OUT\"\n\t\t\t\t\t\t\t? \"execution_failed\"\n\t\t\t\t\t\t\t: \"execution_completed\",\n\t\t\t\teventId: `${record.missionId}:${attemptId}:${executionId}:terminal`,\n\t\t\t\tatMs: result.finishedAtMs,\n\t\t\t\tcorrelation: { ...correlation, executionId },\n\t\t\t\tstate: result.state,\n\t\t\t\texecutionOutcome: result.executionOutcome,\n\t\t\t});\n\t\t\tcompleted = true;\n\t\t\treturn terminal;\n\t\t} finally {\n\t\t\theartbeat.stop();\n\t\t\tthis._activeExecutions.delete(missionId);\n\t\t\tif (!completed && handle) {\n\t\t\t\t// Abnormal exit (authority lost, commit failure, or executor\n\t\t\t\t// error): do not leave the child process running.\n\t\t\t\ttry {\n\t\t\t\t\tawait handle.cancel(\"execution aborted\");\n\t\t\t\t} catch {\n\t\t\t\t\t// Best-effort: never mask the original failure.\n\t\t\t\t}\n\t\t\t}\n\t\t}\n\t}\n\n\t/**\n\t * Build a terminal FAILED record for an executor that rejected `launch`.\n\t * No execution id was ever established, so `resultExecutionId` stays unset;\n\t * the attempt remains auditable with `endReason: CRASHED`. The lease is\n\t * cleared atomically with the terminal transition.\n\t */\n\tprivate _failLaunch(\n\t\trecord: DurableMissionRecord,\n\t\tattemptId: string,\n\t\tmessage: string,\n\t\tatMs: number,\n\t): DurableMissionRecord {\n\t\tlet next = this._withTransition(record, \"FAILED\", {\n\t\t\treason: \"executor rejected launch\",\n\t\t\tattemptId,\n\t\t\tatMs,\n\t\t});\n\t\tconst index = next.attempts.length - 1;\n\t\tconst failedAttempt: DurableExecutionAttempt = {\n\t\t\t...next.attempts[index],\n\t\t\tfinishedAtMs: atMs,\n\t\t\tendReason: \"CRASHED\",\n\t\t};\n\t\tnext = {\n\t\t\t...next,\n\t\t\tresult: createMissionResult({\n\t\t\t\tmissionId: next.missionId,\n\t\t\t\tparentMissionId: next.parentMissionId,\n\t\t\t\tdepth: next.depth,\n\t\t\t\tstate: \"FAILED\",\n\t\t\t\texecutionOutcome: \"CRASHED\",\n\t\t\t\tverification: { status: \"unverified\" },\n\t\t\t\tcompletionDecision: \"unavailable\",\n\t\t\t\tfailures: [{ category: \"LAUNCH\", message }],\n\t\t\t\texecutorDiagnostics: { executorId: this._executor.executorId, launchError: message },\n\t\t\t\tstartedAtMs: next.startedAtMs ?? atMs,\n\t\t\t\tfinishedAtMs: atMs,\n\t\t\t}),\n\t\t\tfinishedAtMs: atMs,\n\t\t\tcurrentAttemptId: undefined,\n\t\t\tcurrentExecutionId: undefined,\n\t\t\tlease: undefined,\n\t\t\tattempts: [...next.attempts.slice(0, index), failedAttempt],\n\t\t};\n\t\treturn next;\n\t}\n\n\t// =========================================================================\n\t// Internals\n\t// =========================================================================\n\n\tprivate _emitExecutionEvent(event: MissionExecutionEvent): void {\n\t\ttry {\n\t\t\tthis._executionObserver?.onEvent(event);\n\t\t} catch {\n\t\t\t// Observers are telemetry only and never alter execution authority.\n\t\t}\n\t}\n\n\tprivate _buildHeartbeat(\n\t\tmissionId: string,\n\t\tproof: ExecutionLeaseProof,\n\t\tonAuthorityLost: (info: HeartbeatAuthorityLossInfo) => void,\n\t): ExecutionHeartbeat {\n\t\treturn new ExecutionHeartbeat({\n\t\t\tmissionId,\n\t\t\tleaseId: proof.leaseId,\n\t\t\tfencingToken: proof.fencingToken,\n\t\t\ttiming: this._heartbeatTiming,\n\t\t\tnow: this._now,\n\t\t\tschedule: this._heartbeatScheduler,\n\t\t\trenew: async (now) => {\n\t\t\t\tconst renewed = await this.renewOwnership(missionId, proof, { now });\n\t\t\t\treturn { expiresAtMs: renewed.lease.expiresAtMs };\n\t\t\t},\n\t\t\tonAuthorityLost,\n\t\t});\n\t}\n\n\tprivate _withTransition(\n\t\trecord: DurableMissionRecord,\n\t\tto: MissionState,\n\t\tmeta: { reason?: string; executionId?: string; attemptId?: string; atMs?: number },\n\t): DurableMissionRecord {\n\t\tconst assertion = assertMissionTransition(record.state, to);\n\t\tif (!assertion.ok) {\n\t\t\tthrow new Error(`Illegal mission transition for ${record.missionId}: ${assertion.error}`);\n\t\t}\n\t\tconst from = record.state;\n\t\tconst atMs = meta.atMs ?? this._now();\n\t\tconst seq = record.transitions.length === 0 ? 0 : record.transitions[record.transitions.length - 1].seq + 1;\n\t\treturn {\n\t\t\t...record,\n\t\t\tstate: to,\n\t\t\ttransitions: [\n\t\t\t\t...record.transitions,\n\t\t\t\t{ seq, from, to, atMs, reason: meta.reason, executionId: meta.executionId, attemptId: meta.attemptId },\n\t\t\t],\n\t\t\tupdatedAtMs: atMs,\n\t\t\trevision: record.revision + 1,\n\t\t};\n\t}\n\n\t/**\n\t * Persist an execution-authoritative transition under the current fence.\n\t *\n\t * This is an atomic store mutation, NOT an optimistic `save` against a\n\t * coordinator-held snapshot. Heartbeat renewals legitimately advance the\n\t * ordinary `revision` while the execution is running, so a snapshot-based\n\t * `expectedRevision` would reject a valid terminal commit. The fencing token\n\t * (leaseId + fencingToken) is the authority here; `revision` is rebased onto\n\t * the current record so it remains a monotonic history counter.\n\t */\n\tprivate async _commitFenced(\n\t\tprevious: DurableMissionRecord,\n\t\tnext: DurableMissionRecord,\n\t\tproof: ExecutionLeaseProof,\n\t): Promise<DurableMissionRecord> {\n\t\ttype CommitOutcome =\n\t\t\t| { status: \"committed\"; record: DurableMissionRecord }\n\t\t\t| { status: \"lease_not_found\" }\n\t\t\t| { status: \"stale_owner\"; leaseId: string; fencingToken: number };\n\n\t\tconst result = await this._store.mutate<CommitOutcome>(previous.missionId, (current) => {\n\t\t\tconst lease = current.lease;\n\t\t\tif (!lease) return { kind: \"noop\", value: { status: \"lease_not_found\" as const } };\n\t\t\tif (lease.leaseId !== proof.leaseId || lease.fencingToken !== proof.fencingToken) {\n\t\t\t\treturn {\n\t\t\t\t\tkind: \"noop\",\n\t\t\t\t\tvalue: { status: \"stale_owner\" as const, leaseId: lease.leaseId, fencingToken: lease.fencingToken },\n\t\t\t\t};\n\t\t\t}\n\t\t\tconst committed: DurableMissionRecord = {\n\t\t\t\t...next,\n\t\t\t\trevision: current.revision + 1,\n\t\t\t};\n\t\t\treturn { kind: \"write\", next: committed, value: { status: \"committed\" as const, record: committed } };\n\t\t});\n\n\t\tif (result.status === \"missing\") {\n\t\t\tthrow new ExecutionOwnershipError(\"LEASE_NOT_FOUND\", `Mission ${next.missionId} not found`);\n\t\t}\n\t\tif (result.status === \"corrupt\") {\n\t\t\tthrow new Error(`Mission ${next.missionId} is corrupt: ${result.diagnostic}`);\n\t\t}\n\t\tconst value = result.value;\n\t\tif (value.status === \"committed\") return value.record;\n\t\tif (value.status === \"lease_not_found\") {\n\t\t\tthrow new ExecutionOwnershipError(\"LEASE_NOT_FOUND\", `Mission ${next.missionId} has no execution lease`);\n\t\t}\n\t\tthrow new ExecutionOwnershipError(\n\t\t\t\"STALE_EXECUTION_OWNER\",\n\t\t\t`Stale execution owner for mission ${next.missionId}: lease no longer authoritative`,\n\t\t\t{ missionId: next.missionId, leaseId: value.leaseId, fencingToken: value.fencingToken },\n\t\t);\n\t}\n\n\tprivate _mapRenewResult(\n\t\tmissionId: string,\n\t\tresult: DurableMissionMutateResult<RenewOutcome>,\n\t): { lease: ExecutionLease; record: DurableMissionRecord } {\n\t\tif (result.status === \"missing\") {\n\t\t\tthrow new ExecutionOwnershipError(\"LEASE_NOT_FOUND\", `Mission ${missionId} not found`);\n\t\t}\n\t\tif (result.status === \"corrupt\") {\n\t\t\tthrow new Error(`Mission ${missionId} is corrupt: ${result.diagnostic}`);\n\t\t}\n\t\tconst value = result.value;\n\t\tif (value.status === \"renewed\") return { lease: value.lease, record: value.record };\n\t\tif (value.status === \"lease_not_found\") {\n\t\t\tthrow new ExecutionOwnershipError(\"LEASE_NOT_FOUND\", `Mission ${missionId} has no execution lease`);\n\t\t}\n\t\tif (value.status === \"stale_owner\") {\n\t\t\tthrow new ExecutionOwnershipError(\n\t\t\t\t\"STALE_EXECUTION_OWNER\",\n\t\t\t\t`Mission ${missionId} lease no longer belongs to this owner`,\n\t\t\t\t{ missionId, leaseId: value.lease.leaseId, fencingToken: value.lease.fencingToken },\n\t\t\t);\n\t\t}\n\t\tthrow new ExecutionOwnershipError(\"LEASE_EXPIRED\", `Execution lease for mission ${missionId} has expired`, {\n\t\t\tmissionId,\n\t\t\tleaseId: value.lease.leaseId,\n\t\t\tfencingToken: value.lease.fencingToken,\n\t\t\texpiresAtMs: value.lease.expiresAtMs,\n\t\t});\n\t}\n\n\tprivate async _revokeInterrupted(\n\t\tmissionId: string,\n\t\tnow: number,\n\t\treason: string,\n\t): Promise<\n\t\t| { status: \"reconciled\"; previousState: MissionState }\n\t\t| { status: \"unchanged\" }\n\t\t| { status: \"corrupt\"; diagnostic: string }\n\t> {\n\t\tconst result = await this._store.mutate<RevokeOutcome>(missionId, (current) => {\n\t\t\tif (isTerminalMissionState(current.state) || current.state === \"CREATED\" || current.state === \"INTERRUPTED\") {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"unchanged\" as const } };\n\t\t\t}\n\t\t\tif (!ACTIVE_NONTERMINAL_STATES.has(current.state)) {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"unchanged\" as const } };\n\t\t\t}\n\t\t\tif (current.lease && isExecutionLeaseActive(current.lease, now)) {\n\t\t\t\treturn { kind: \"noop\", value: { status: \"unchanged\" as const } };\n\t\t\t}\n\n\t\t\tconst previousState = current.state;\n\t\t\t// Revoking an expired lease fenced the dead owner: bump the epoch so a\n\t\t\t// late wake-up with the old fence can never write again.\n\t\t\tconst fencingToken = current.lease ? current.fencingToken + 1 : current.fencingToken;\n\t\t\tlet next = this._withTransition(current, \"INTERRUPTED\", { reason, atMs: now });\n\n\t\t\tconst attempts = [...current.attempts];\n\t\t\tif (current.currentAttemptId) {\n\t\t\t\tconst index = attempts.findIndex((a) => a.attemptId === current.currentAttemptId);\n\t\t\t\tif (index >= 0) {\n\t\t\t\t\tattempts[index] = {\n\t\t\t\t\t\t...attempts[index],\n\t\t\t\t\t\tendReason: \"INTERRUPTED\",\n\t\t\t\t\t\trecovery: { reason, recoveredAtMs: now },\n\t\t\t\t\t};\n\t\t\t\t}\n\t\t\t}\n\t\t\tnext = {\n\t\t\t\t...next,\n\t\t\t\tfencingToken,\n\t\t\t\tlease: undefined,\n\t\t\t\tcurrentAttemptId: undefined,\n\t\t\t\tcurrentExecutionId: undefined,\n\t\t\t\tattempts,\n\t\t\t};\n\t\t\treturn { kind: \"write\", next, value: { status: \"reconciled\" as const, previousState } };\n\t\t});\n\n\t\tif (result.status === \"missing\") return { status: \"unchanged\" };\n\t\tif (result.status === \"corrupt\") return { status: \"corrupt\", diagnostic: result.diagnostic };\n\t\treturn result.value;\n\t}\n}\n"]}