import type { ExecutionOutcome, ExecutionOutputSink, } from '../domain/execution'; import type { Executor } from '../ports/executor'; import type { RunRepository } from '../ports/runRepository'; import { buildLegacyCronExecutionSpec } from '../adapters/legacy/legacyCronExecutionSpec'; import { createLegacyTaskRevision } from '../compatibility/legacyTaskRevision'; import { createLegacyLogOutputRef } from '../compatibility/legacyLogOutputRef'; import type { ManualPrimaryCompletion, ManualPrimaryExecutionRouter, ManualPrimaryStartInput, ManualPrimaryStartedExecution, ManualPrimaryStopResult, } from '../compatibility/manualPrimaryExecutionBridge'; import type { RuntimeRolloutPolicy } from '../domain/runtimeRollout'; import { PrimaryRunOrchestrator, type ActivePrimaryRun, type PrimaryRunClock, type PrimaryRunOrchestratorOptions, } from './primaryRunOrchestrator'; export interface PreparedManualPrimaryLog { logPath: string; output: ExecutionOutputSink; completionCommitted?(attemptId: string): Promise; close(): Promise; } export interface ManualPrimaryLogFiles { prepare(input: ManualPrimaryStartInput): Promise; } export interface ManualPrimaryRuntimeOptions { clock?: PrimaryRunClock; orchestrator?: PrimaryRunOrchestratorOptions; } interface ActiveManualExecution { cronId: number; attemptId: string; execution: ActivePrimaryRun; } interface PendingManualExecution { cronId: number; controller: AbortController; } export class ManualPrimaryOwnershipError extends Error { constructor() { super('Manual execution is not Runtime-owned'); this.name = 'ManualPrimaryOwnershipError'; } } /** * Process-local manual Primary owner. The active maps are bounded by actual * concurrent executions; durable restart/cross-worker ownership remains the * startup Reconciler and future supervisor's responsibility. */ export class ManualPrimaryRuntime implements ManualPrimaryExecutionRouter { private readonly orchestrator: PrimaryRunOrchestrator; private readonly clock: PrimaryRunClock; private readonly pending = new Map(); private readonly active = new Map(); constructor( repository: RunRepository, executor: Executor, private readonly rollout: RuntimeRolloutPolicy, private readonly logs: ManualPrimaryLogFiles, options: ManualPrimaryRuntimeOptions = {}, ) { this.clock = options.clock ?? { now: Date.now }; this.orchestrator = new PrimaryRunOrchestrator(repository, executor, { ...options.orchestrator, clock: this.clock, }); } ownsNewRuns(): boolean { return this.rollout.decide('manual').owner === 'runtime'; } async start( input: ManualPrimaryStartInput, ): Promise { if (!this.ownsNewRuns()) throw new ManualPrimaryOwnershipError(); const pendingId = Symbol('manual-primary-pending'); const controller = new AbortController(); this.pending.set(pendingId, { cronId: input.cron.id, controller }); let prepared: PreparedManualPrimaryLog | undefined; try { prepared = await this.logs.prepare(input); const taskRevision = createLegacyTaskRevision({ command: input.cron.command, ...(input.cron.schedule === undefined ? {} : { schedule: input.cron.schedule }), extraSchedules: input.cron.extraSchedules, ...(input.cron.taskBefore === undefined ? {} : { taskBefore: input.cron.taskBefore }), ...(input.cron.taskAfter === undefined ? {} : { taskAfter: input.cron.taskAfter }), ...(input.cron.workDirectory === undefined ? {} : { workDirectory: input.cron.workDirectory }), ...(input.cron.logName === undefined ? {} : { logName: input.cron.logName }), }); const outputRef = createLegacyLogOutputRef(prepared.logPath); const active = await this.orchestrator.start({ definition: { projectId: 'default', taskId: 'legacy-cron:' + input.cron.id, taskRevision, ...(input.cron.name === undefined ? {} : { taskName: input.cron.name }), legacyCronId: input.cron.id, triggerType: 'manual', executionOrigin: 'manual', triggeredBy: 'legacy:manual-api', outputRef, acceptedAtMs: input.acceptedAtMs, actor: { type: 'compatibility', id: 'legacy:manual-api' }, }, createSpec: (reference) => buildLegacyCronExecutionSpec({ runId: reference.run.id, attemptId: reference.attempt.id, projectId: reference.run.projectId, taskRevision: reference.run.taskRevision, cron: { id: input.cron.id, command: input.cron.command, ...(input.cron.taskBefore === undefined ? {} : { taskBefore: input.cron.taskBefore }), ...(input.cron.taskAfter === undefined ? {} : { taskAfter: input.cron.taskAfter }), ...(input.cron.workDirectory === undefined ? {} : { workDirectory: input.cron.workDirectory }), ...(input.cron.logName === undefined ? {} : { logName: input.cron.logName }), }, realTime: true, realLogPath: prepared!.logPath, noDelay: true, }), context: { environment: {}, signal: controller.signal, output: prepared.output, }, }); const entry: ActiveManualExecution = { cronId: input.cron.id, attemptId: active.attempt.id, execution: active, }; this.active.set(active.run.id, entry); this.pending.delete(pendingId); const completion = this.completion(active, prepared).finally(() => { if (this.active.get(active.run.id) === entry) { this.active.delete(active.run.id); } }); void completion.catch(() => undefined); return { runId: active.run.id, attemptId: active.attempt.id, ...(active.handle.pid === undefined ? {} : { pid: active.handle.pid }), logPath: prepared.logPath, completion, }; } catch (error) { if (prepared) await prepared.close().catch(() => undefined); throw error; } finally { this.pending.delete(pendingId); } } async stopCron( cronId: number, requestedAtMs: number, ): Promise { let matched = 0; for (const pending of this.pending.values()) { if (pending.cronId !== cronId) continue; matched += 1; pending.controller.abort(); } const executions = [...this.active.values()].filter( (entry) => entry.cronId === cronId, ); matched += executions.length; const failed = await this.stopActive(executions, requestedAtMs); return { matched, failed }; } async stopAttempt( attemptId: string, requestedAtMs: number, ): Promise { const execution = [...this.active.values()].find( (entry) => entry.attemptId === attemptId, ); if (!execution) return { matched: 0, failed: 0 }; const failed = await this.stopActive([execution], requestedAtMs); return { matched: 1, failed }; } private async stopActive( executions: readonly ActiveManualExecution[], requestedAtMs: number, ): Promise { const results = await Promise.allSettled( executions.map((entry) => entry.execution.cancel({ kind: 'user', requestedAtMs }), ), ); return results.filter((result) => result.status === 'rejected').length; } private async completion( active: ActivePrimaryRun, prepared: PreparedManualPrimaryLog, ): Promise { try { const completed = await active.completion; await prepared .completionCommitted?.(completed.attempt.id) .catch(() => undefined); const result = completed.result; return { runId: completed.run.id, attemptId: completed.attempt.id, outcome: result.outcome as ExecutionOutcome, ...(result.exitCode === undefined ? {} : { exitCode: result.exitCode }), }; } finally { await prepared.close().catch(() => undefined); } } }