AgentRun Codex
889 lines
51 KiB
TypeScript
889 lines
51 KiB
TypeScript
import { appendFile, mkdir, writeFile } from "node:fs/promises";
|
|
import path from "node:path";
|
|
import { RunnerManagerApi, failureKindFromError, terminalStatusForFailure, errorMessage } from "./manager-api.js";
|
|
import { createBackendSession, runBackendTurn, type BackendActiveTurnControl, type BackendAdapterOptions, type BackendSession } from "../backend/adapter.js";
|
|
import { materializeResourceBundle } from "./resource-bundle.js";
|
|
import type { BackendEvent, BackendProfile, BackendTurnResult, CommandRecord, FailureKind, InitialPromptAssembly, JsonRecord, RunRecord, RunnerRecord, TerminalStatus } from "../common/types.js";
|
|
import { AgentRunError } from "../common/errors.js";
|
|
import { smokeBundledWorkReadyCapabilities, smokeImageWorkReadyCapabilities } from "../common/work-ready.js";
|
|
import { stableHash } from "../common/validation.js";
|
|
import { prepareGithubSshCredentialEnvironment } from "./github-ssh-credential.js";
|
|
|
|
export interface RunnerOnceOptions extends BackendAdapterOptions {
|
|
managerUrl: string;
|
|
runId: string;
|
|
commandId?: string;
|
|
runnerJobId?: string;
|
|
runnerId?: string;
|
|
attemptId?: string;
|
|
leaseMs?: number;
|
|
backendProfile?: BackendProfile;
|
|
placement?: "host-process" | "kubernetes-job";
|
|
sourceCommit?: string;
|
|
jobName?: string;
|
|
podName?: string;
|
|
logPath?: string;
|
|
idleTimeoutMs?: number;
|
|
pollIntervalMs?: number;
|
|
claimRetryTimeoutMs?: number;
|
|
claimRetryIntervalMs?: number;
|
|
backendRetryMaxAttempts?: number;
|
|
backendRetryInitialBackoffMs?: number;
|
|
backendRetryMaxBackoffMs?: number;
|
|
oneShot?: boolean;
|
|
}
|
|
|
|
interface CommandExecutionResult extends JsonRecord {
|
|
commandId: string;
|
|
terminalStatus: TerminalStatus;
|
|
failureKind: FailureKind | null;
|
|
}
|
|
|
|
interface RunnerFailure {
|
|
terminalStatus: TerminalStatus;
|
|
failureKind: FailureKind;
|
|
message: string;
|
|
details?: JsonRecord | null;
|
|
}
|
|
|
|
interface CommandTerminalReport {
|
|
terminalStatus: TerminalStatus;
|
|
failureKind: FailureKind | null;
|
|
failureMessage: string | null;
|
|
threadId?: string;
|
|
turnId?: string;
|
|
}
|
|
|
|
interface BackendRetryPolicy {
|
|
maxAttempts: number;
|
|
initialBackoffMs: number;
|
|
maxBackoffMs: number;
|
|
}
|
|
|
|
interface RunnerLogSink {
|
|
write(label: string, payload: JsonRecord): Promise<void>;
|
|
}
|
|
|
|
class TerminalOutboxReportError extends Error {
|
|
constructor(readonly cause: unknown) {
|
|
super(`terminal report failed after durable outbox write: ${errorMessage(cause)}`);
|
|
this.name = "TerminalOutboxReportError";
|
|
}
|
|
}
|
|
|
|
export async function runOnce(options: RunnerOnceOptions): Promise<JsonRecord> {
|
|
const runnerLog = await createRunnerLogSink(options.logPath);
|
|
await runnerLog.write("runner.starting", { runId: options.runId, commandId: options.commandId ?? null, runnerJobId: options.runnerJobId ?? null, valuesPrinted: false });
|
|
const api = new RunnerManagerApi(options.managerUrl);
|
|
const targetRun = await api.getRun(options.runId);
|
|
if (isTerminalRun(targetRun)) return { terminalStatus: targetRun.terminalStatus, failureKind: targetRun.failureKind, run: targetRun, skipped: "run-terminal" } as JsonRecord;
|
|
if (options.backendProfile && options.backendProfile !== targetRun.backendProfile) {
|
|
throw new AgentRunError("schema-invalid", `runner backendProfile ${options.backendProfile} does not match run backendProfile ${targetRun.backendProfile}`, { httpStatus: 400 });
|
|
}
|
|
const leaseMs = options.leaseMs ?? 60_000;
|
|
const attemptId = options.attemptId ?? `attempt_${Date.now().toString(36)}`;
|
|
const runner = await api.register({
|
|
runId: options.runId,
|
|
attemptId,
|
|
backendProfile: targetRun.backendProfile,
|
|
placement: options.placement ?? "host-process",
|
|
sourceCommit: options.sourceCommit ?? process.env.AGENTRUN_SOURCE_COMMIT ?? "unknown",
|
|
...(options.runnerId ? { runnerId: options.runnerId } : {}),
|
|
...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}),
|
|
...(options.jobName ? { jobName: options.jobName } : {}),
|
|
...(options.podName ? { podName: options.podName } : {}),
|
|
...(options.logPath ? { logPath: options.logPath } : {}),
|
|
}) as RunnerRecord;
|
|
await runnerLog.write("runner.registered", { runId: options.runId, runnerId: runner.id, runnerJobId: options.runnerJobId ?? null, attemptId, valuesPrinted: false });
|
|
try {
|
|
const imageWorkReady = await smokeImageWorkReadyCapabilities(options.env ?? process.env);
|
|
await api.appendEvent(options.runId, { type: "backend_status", payload: { phase: "runner-image-work-ready-smoke", attemptId, runnerId: runner.id, ...imageWorkReady } });
|
|
} catch (error) {
|
|
const failureKind = failureKindFromError(error);
|
|
const message = errorMessage(error);
|
|
const details = failureDetailsFromError(error);
|
|
await api.appendEvent(options.runId, { type: "error", payload: { failureKind, message, phase: "runner-image-work-ready-smoke", attemptId, runnerId: runner.id, ...(details ? { details } : {}) } });
|
|
const finalRun = await api.reportStatus(options.runId, { terminalStatus: terminalStatusForFailure(failureKind), failureKind, failureMessage: message }) as RunRecord;
|
|
return { runner, terminalStatus: finalRun.terminalStatus, failureKind, run: finalRun, commandsProcessed: 0, commandResults: [], stopped: "image-work-ready-smoke-failed" } as JsonRecord;
|
|
}
|
|
let claimed: RunRecord;
|
|
try {
|
|
claimed = await claimRunWithLeaseRecovery(api, options, runner, attemptId, leaseMs);
|
|
await api.heartbeat(options.runId, runner.id, leaseMs);
|
|
} catch (error) {
|
|
const failureKind = failureKindFromError(error);
|
|
if (failureKind !== "runner-lease-conflict") {
|
|
await api.reportFailure(options.runId, { terminalStatus: terminalStatusForFailure(failureKind), failureKind, failureMessage: errorMessage(error) });
|
|
}
|
|
throw error;
|
|
}
|
|
|
|
const stopHeartbeat = startHeartbeat(api, options.runId, runner.id, leaseMs);
|
|
const idleTimeoutMs = options.idleTimeoutMs ?? 120_000;
|
|
const pollIntervalMs = normalizePollIntervalMs(options.pollIntervalMs);
|
|
const commandResults: CommandExecutionResult[] = [];
|
|
let workspacePath: string | undefined;
|
|
let resourceEnv: NodeJS.ProcessEnv | undefined;
|
|
let initialPrompt: InitialPromptAssembly | undefined;
|
|
let materializationAttempted = false;
|
|
let materializationFailure: RunnerFailure | null = null;
|
|
let backendSession: BackendSession | null = null;
|
|
|
|
try {
|
|
let idleSince = Date.now();
|
|
let firstPoll = true;
|
|
let commandPollAfterSeq = 0;
|
|
while (true) {
|
|
const currentRun = await api.getRun(options.runId);
|
|
if (isTerminalRun(currentRun)) return { runner, claimed, terminalStatus: currentRun.terminalStatus, failureKind: currentRun.failureKind, run: currentRun, commandsProcessed: commandResults.length, commandResults, stopped: "run-terminal" } as JsonRecord;
|
|
|
|
const commandsResponse = await api.pollCommands(options.runId, { afterSeq: commandPollAfterSeq, limit: 50, ...(firstPoll && options.commandId ? { commandId: options.commandId } : {}) });
|
|
firstPoll = false;
|
|
const command = commandsResponse.selected;
|
|
if (!command) {
|
|
await failPendingSteerWithoutActiveTurn(api, options.runId, commandsResponse.items, runner.id, attemptId, runnerLog, options.runnerJobId);
|
|
commandPollAfterSeq = Math.max(commandPollAfterSeq, lastCommandSeq(commandsResponse.items));
|
|
if (options.oneShot === true) return noPendingResult(runner, claimed, commandResults, commandsResponse.items.length, "one-shot-no-pending");
|
|
if (Date.now() - idleSince >= idleTimeoutMs) return noPendingResult(runner, claimed, commandResults, commandsResponse.items.length, "idle-timeout");
|
|
await sleep(pollIntervalMs);
|
|
continue;
|
|
}
|
|
|
|
idleSince = Date.now();
|
|
if (!materializationAttempted) {
|
|
materializationAttempted = true;
|
|
try {
|
|
const materialized = await materializeResourceBundle(claimed.resourceBundleRef ?? null, claimed.workspaceRef, resourceMaterializationEnv(options.env ?? process.env, options.runId, attemptId));
|
|
if (materialized) {
|
|
workspacePath = materialized.workspacePath;
|
|
resourceEnv = resourceEnvForMaterialized(options.env ?? process.env, materialized);
|
|
initialPrompt = materialized.initialPrompt;
|
|
await api.appendEvent(options.runId, { type: "backend_status", payload: { ...materialized.event, commandId: command.id, attemptId, runnerId: runner.id } });
|
|
if (requiresBundledWorkReadyTools(claimed)) {
|
|
const bundleWorkReady = await smokeBundledWorkReadyCapabilities(resourceEnv ?? options.env ?? process.env);
|
|
await api.appendEvent(options.runId, { type: "backend_status", payload: { phase: "runner-bundle-work-ready-smoke", commandId: command.id, attemptId, runnerId: runner.id, ...bundleWorkReady } });
|
|
}
|
|
}
|
|
} catch (error) {
|
|
const failureKind = failureKindFromError(error);
|
|
materializationFailure = { failureKind, terminalStatus: terminalStatusForFailure(failureKind), message: errorMessage(error), details: failureDetailsFromError(error) };
|
|
}
|
|
}
|
|
|
|
const result = materializationFailure
|
|
? await reportCommandFailure(api, options.runId, command.id, runner, attemptId, materializationFailure, "runner:resource-bundle", runnerLog, { terminalRun: true, ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}) })
|
|
: await executeCommand(api, withResourceAssembly(options, resourceEnv, initialPrompt), command, runner, attemptId, workspacePath, backendSession ?? (backendSession = createBackendSession(currentRun, withResourceAssembly(options, resourceEnv, initialPrompt))), runnerLog);
|
|
commandResults.push(result);
|
|
commandPollAfterSeq = Math.max(commandPollAfterSeq, command.seq);
|
|
idleSince = Date.now();
|
|
if (options.oneShot === true) {
|
|
const run = await api.reportStatus(options.runId, { terminalStatus: result.terminalStatus, failureKind: result.failureKind, failureMessage: null });
|
|
return { runner, commandId: command.id, terminalStatus: result.terminalStatus, failureKind: result.failureKind, run, commandsProcessed: commandResults.length, commandResults, stopped: "one-shot" } as JsonRecord;
|
|
}
|
|
}
|
|
} catch (error) {
|
|
const failureKind = failureKindFromError(error);
|
|
const terminalStatus = terminalStatusForFailure(failureKind);
|
|
const message = errorMessage(error);
|
|
await api.appendEvent(options.runId, { type: "error", payload: { failureKind, message, phase: "runner:loop", attemptId, runnerId: runner.id } });
|
|
const finalRun = await api.reportStatus(options.runId, { terminalStatus, failureKind, failureMessage: message }) as RunRecord;
|
|
return { runner, terminalStatus, failureKind, run: finalRun, commandsProcessed: commandResults.length, commandResults } as JsonRecord;
|
|
} finally {
|
|
if (backendSession) {
|
|
try {
|
|
const closeEvents = await backendSession.close();
|
|
for (const event of closeEvents) await api.appendEvent(options.runId, annotateCommandEvent(event, "runner-shutdown", attemptId, runner.id));
|
|
} catch {
|
|
// Runner shutdown must not hide the terminal result already reported for the command.
|
|
}
|
|
}
|
|
stopHeartbeat();
|
|
}
|
|
}
|
|
|
|
function requiresBundledWorkReadyTools(run: RunRecord): boolean {
|
|
const toolCredentials = run.executionPolicy.secretScope.toolCredentials ?? [];
|
|
if (toolCredentials.some((item) => item.tool === "unidesk-ssh" || item.tool === "github")) return true;
|
|
const requiredSkills = run.resourceBundleRef?.requiredSkills ?? [];
|
|
return requiredSkills.some((item) => item.name === "unidesk-trans" || item.name === "unidesk-gh" || item.name === "unidesk-agentrun" || item.name === "unidesk-cicd" || item.name === "dad-dev");
|
|
}
|
|
|
|
function withResourceAssembly(options: RunnerOnceOptions, resourceEnv: NodeJS.ProcessEnv | undefined, initialPrompt: InitialPromptAssembly | undefined): RunnerOnceOptions {
|
|
return {
|
|
...options,
|
|
...(resourceEnv ? { env: resourceEnv } : {}),
|
|
...(initialPrompt ? { initialPrompt } : {}),
|
|
};
|
|
}
|
|
|
|
function resourceMaterializationEnv(env: NodeJS.ProcessEnv, runId: string, attemptId: string): NodeJS.ProcessEnv {
|
|
return {
|
|
...env,
|
|
AGENTRUN_RUN_ID: env.AGENTRUN_RUN_ID ?? runId,
|
|
AGENTRUN_ATTEMPT_ID: env.AGENTRUN_ATTEMPT_ID ?? attemptId,
|
|
};
|
|
}
|
|
|
|
function resourceEnvForMaterialized(env: NodeJS.ProcessEnv, materialized: Awaited<ReturnType<typeof materializeResourceBundle>>): NodeJS.ProcessEnv | undefined {
|
|
if (!materialized) return undefined;
|
|
const sanitized = { ...env, AGENTRUN_WORKSPACE_PATH: materialized.workspacePath };
|
|
delete sanitized.AGENTRUN_RESOURCE_CREDENTIAL_PATH;
|
|
let next: NodeJS.ProcessEnv | undefined = sanitized;
|
|
if (materialized.binPath) {
|
|
const withPath = prependPath(sanitized, materialized.binPath);
|
|
next = { ...withPath, AGENTRUN_RESOURCE_BIN_PATH: withPath.AGENTRUN_RESOURCE_BIN_PATH ?? materialized.binPath };
|
|
}
|
|
if (materialized.skillsDir) {
|
|
const base = next ?? sanitized;
|
|
const previous = base.AGENTRUN_SKILLS_DIRS;
|
|
next = {
|
|
...base,
|
|
AGENTRUN_SKILLS_DIRS: previous && previous.trim().length > 0 ? `${materialized.skillsDir}${pathDelimiter()}${previous}` : materialized.skillsDir,
|
|
HWLAB_CODE_AGENT_SKILLS_DIRS: materialized.skillsDir,
|
|
};
|
|
}
|
|
return next;
|
|
}
|
|
|
|
function prependPath(env: NodeJS.ProcessEnv, binPath: string): NodeJS.ProcessEnv {
|
|
return { ...env, PATH: `${binPath}${pathDelimiter()}${env.PATH ?? process.env.PATH ?? ""}` };
|
|
}
|
|
|
|
function pathDelimiter(): string {
|
|
return process.platform === "win32" ? ";" : ":";
|
|
}
|
|
|
|
async function executeCommand(api: RunnerManagerApi, options: RunnerOnceOptions, command: CommandRecord, runner: RunnerRecord, attemptId: string, workspacePath: string | undefined, backendSession: BackendSession | null, runnerLog: RunnerLogSink): Promise<CommandExecutionResult> {
|
|
const resumeDurableTurn = command.state === "acknowledged";
|
|
await api.ackCommand(command.id);
|
|
const acked = await api.getCommand(options.runId, command.id);
|
|
if (acked.state === "cancelled") return await reportCancelled(api, options.runId, command.id, runner, attemptId, "command cancelled before backend start", runnerLog, options.runnerJobId);
|
|
await assertNotCancelled(api, options.runId, command.id);
|
|
const abortController = new AbortController();
|
|
let stopCancelWatch: () => void = () => undefined;
|
|
const backendProgress = startBackendProgress();
|
|
let backendStarted = false;
|
|
let stopSteerWatch: (() => void) | undefined;
|
|
const terminalOutboxEvents: BackendEvent[] = [];
|
|
try {
|
|
const latestRun = await api.getRun(options.runId);
|
|
const preparedCredential = await prepareGithubSshCredentialEnvironment(latestRun.executionPolicy, options.env ?? process.env);
|
|
if (preparedCredential.summary) {
|
|
await api.appendEvent(options.runId, { type: "backend_status", payload: { phase: "github-ssh-tool-credential-ready", commandId: command.id, attemptId, runnerId: runner.id, githubSsh: preparedCredential.summary, valuesPrinted: false } });
|
|
}
|
|
await api.appendEvent(options.runId, { type: "backend_status", payload: { phase: "backend-turn-started", commandId: command.id, attemptId, runnerId: runner.id, backendProfile: latestRun.backendProfile, workspaceReady: Boolean(workspacePath) } });
|
|
backendStarted = true;
|
|
await runnerLog.write("command.started", { runId: options.runId, commandId: command.id, runnerId: runner.id, runnerJobId: options.runnerJobId ?? null, attemptId, backendProfile: latestRun.backendProfile, valuesPrinted: false });
|
|
stopCancelWatch = watchCancellation(api, options.runId, command.id, abortController, { attemptId, runnerId: runner.id, ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}), runnerLog });
|
|
let backendAttempt = 1;
|
|
let durableReplayCommands: CommandRecord[] = [];
|
|
let activeTurnRegistered = false;
|
|
let replayOutcome: Promise<unknown | null> = Promise.resolve(null);
|
|
const backendOptions = {
|
|
...options,
|
|
env: preparedCredential.env,
|
|
...(workspacePath ? { workspacePath } : {}),
|
|
abortSignal: abortController.signal,
|
|
otelContext: {
|
|
run: latestRun,
|
|
command,
|
|
attemptId,
|
|
runnerId: runner.id,
|
|
runnerJobId: options.runnerJobId ?? null,
|
|
jobName: options.jobName ?? null,
|
|
podName: options.podName ?? null,
|
|
sourceCommit: options.sourceCommit ?? null,
|
|
logPath: options.logPath ?? null,
|
|
},
|
|
onEvent: async (event: BackendEvent) => {
|
|
await runnerLog.write("backend.event", runnerLogEventSummary(event, options.runId, command.id, attemptId, runner.id, options.runnerJobId ?? null));
|
|
if (shouldKeepTerminalOutboxEvent(event)) {
|
|
terminalOutboxEvents.push(event);
|
|
return;
|
|
}
|
|
await appendBestEffort(api, options.runId, annotateCommandEvent(event, command.id, attemptId, runner.id));
|
|
},
|
|
onActiveTurn: (control: BackendActiveTurnControl) => {
|
|
activeTurnRegistered = true;
|
|
const watcher = startActiveTurnCommandWatch(api, options.runId, command, attemptId, runner.id, control, options.pollIntervalMs, runnerLog, options.runnerJobId, backendAttempt, resumeDurableTurn, durableReplayCommands);
|
|
stopSteerWatch = watcher.stop;
|
|
replayOutcome = watcher.replayBarrier.then(() => null, (error) => error);
|
|
return () => {
|
|
stopSteerWatch?.();
|
|
stopSteerWatch = undefined;
|
|
};
|
|
},
|
|
};
|
|
const retryPolicy = backendRetryPolicy(latestRun.executionPolicy, preparedCredential.env, options);
|
|
const runBackendAttempt = async (): Promise<BackendTurnResult> => {
|
|
const replayRequired = resumeDurableTurn || backendAttempt > 1;
|
|
durableReplayCommands = replayRequired ? await durableSteersForTurnResume(api, options.runId, command.seq) : [];
|
|
activeTurnRegistered = false;
|
|
replayOutcome = Promise.resolve(null);
|
|
const result = backendSession ? await backendSession.runTurn(latestRun, command, backendOptions) : await runBackendTurn(latestRun, command, backendOptions);
|
|
if (result.terminalStatus !== "completed" || durableReplayCommands.length === 0) return result;
|
|
const replayError = activeTurnRegistered
|
|
? await replayOutcome
|
|
: new AgentRunError("backend-protocol-error", "backend completed before durable steer replay control became available", { details: { backendAttempt, replayCommandCount: durableReplayCommands.length, valuesPrinted: false } });
|
|
if (!replayError) return result;
|
|
const failureKind = failureKindFromError(replayError);
|
|
const message = errorMessage(replayError);
|
|
await appendBestEffort(api, options.runId, { type: "error", payload: { phase: "turn/steer:replay-barrier-failed", commandId: command.id, failureKind, message, backendAttempt, replayCommandCount: durableReplayCommands.length, replayCommandIds: durableReplayCommands.map((item) => item.id), resumeDurableTurn, attemptId, runnerId: runner.id, valuesPrinted: false } });
|
|
return { terminalStatus: "failed", failureKind, failureMessage: message, events: [], ...(result.threadId ? { threadId: result.threadId } : {}), ...(result.turnId ? { turnId: result.turnId } : {}) };
|
|
};
|
|
let result = await runBackendAttempt();
|
|
while (shouldRetryBackendTurn(result, retryPolicy, backendAttempt)) {
|
|
const retryDelayMs = backendRetryDelayMs(retryPolicy, backendAttempt);
|
|
const retryPayload = {
|
|
phase: "runner:backend-retry",
|
|
commandId: command.id,
|
|
attemptId,
|
|
runnerId: runner.id,
|
|
failureKind: result.failureKind,
|
|
message: result.failureMessage ?? `${result.failureKind} retryable backend failure`,
|
|
retryable: true,
|
|
willRetry: true,
|
|
retryAttempt: backendAttempt,
|
|
retryNextAttempt: backendAttempt + 1,
|
|
retryMaxAttempts: retryPolicy.maxAttempts,
|
|
retryDelayMs,
|
|
retryExhausted: false,
|
|
...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}),
|
|
};
|
|
await appendBestEffort(api, options.runId, { type: "error", payload: retryPayload });
|
|
await appendBestEffort(api, options.runId, { type: "backend_status", payload: retryPayload });
|
|
await runnerLog.write("command.retrying", { runId: options.runId, ...retryPayload, valuesPrinted: false });
|
|
if (backendSession) {
|
|
const closeEvents = await backendSession.close();
|
|
for (const event of closeEvents) await appendBestEffort(api, options.runId, annotateCommandEvent(event, command.id, attemptId, runner.id));
|
|
}
|
|
await sleep(retryDelayMs);
|
|
backendAttempt += 1;
|
|
result = await runBackendAttempt();
|
|
}
|
|
for (const event of result.events) {
|
|
if (shouldKeepTerminalOutboxEvent(event)) terminalOutboxEvents.push(event);
|
|
else await appendBestEffort(api, options.runId, annotateCommandEvent(event, command.id, attemptId, runner.id));
|
|
}
|
|
if (isRetryableBackendFailure(result.failureKind) && retryPolicy.maxAttempts > 1 && backendAttempt >= retryPolicy.maxAttempts && result.terminalStatus !== "completed") {
|
|
terminalOutboxEvents.push({ type: "error", payload: { phase: "runner:backend-retry-exhausted", commandId: command.id, attemptId, runnerId: runner.id, failureKind: result.failureKind, message: result.failureMessage ?? `${result.failureKind} retry attempts exhausted`, retryable: true, willRetry: false, retryAttempt: backendAttempt, retryMaxAttempts: retryPolicy.maxAttempts, retryExhausted: true, ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}) } });
|
|
}
|
|
const report: CommandTerminalReport = { terminalStatus: result.terminalStatus, failureKind: result.failureKind, failureMessage: result.failureMessage, ...(result.threadId ? { threadId: result.threadId } : {}), ...(result.turnId ? { turnId: result.turnId } : {}) };
|
|
await reportTerminalCommand(api, {
|
|
runId: options.runId,
|
|
commandId: command.id,
|
|
runnerId: runner.id,
|
|
runnerJobId: options.runnerJobId ?? null,
|
|
attemptId,
|
|
report,
|
|
events: terminalOutboxEvents,
|
|
phase: "runner:execute",
|
|
runnerLog,
|
|
});
|
|
await runnerLog.write("command.terminal", { runId: options.runId, commandId: command.id, runnerId: runner.id, runnerJobId: options.runnerJobId ?? null, attemptId, terminalStatus: result.terminalStatus, failureKind: result.failureKind, valuesPrinted: false });
|
|
return { commandId: command.id, terminalStatus: result.terminalStatus, failureKind: result.failureKind } as CommandExecutionResult;
|
|
} catch (error) {
|
|
if (error instanceof TerminalOutboxReportError) throw error;
|
|
const failureKind = failureKindFromError(error);
|
|
const failure = { failureKind, terminalStatus: terminalStatusForFailure(failureKind), message: errorMessage(error), details: failureDetailsFromError(error) };
|
|
return await reportCommandFailure(api, options.runId, command.id, runner, attemptId, failure, "runner:execute", runnerLog, { terminalRun: failureKind === "secret-unavailable", ...(options.runnerJobId ? { runnerJobId: options.runnerJobId } : {}), events: terminalOutboxEvents });
|
|
} finally {
|
|
stopSteerWatch?.();
|
|
const progressSummary = backendProgress.stop();
|
|
if (backendStarted) await appendBestEffort(api, options.runId, { type: "backend_status", payload: { phase: "backend-turn-finished", commandId: command.id, attemptId, runnerId: runner.id, ...progressSummary } });
|
|
stopCancelWatch();
|
|
}
|
|
}
|
|
|
|
function backendRetryPolicy(policy: RunRecord["executionPolicy"], env: NodeJS.ProcessEnv, options: RunnerOnceOptions): BackendRetryPolicy {
|
|
const configured = jsonRecordValue(policy.backendRetry) ?? jsonRecordValue(policy.retryPolicy);
|
|
const maxAttempts = positiveInteger(configured?.maxAttempts, positiveInteger(env.AGENTRUN_BACKEND_RETRY_MAX_ATTEMPTS, options.backendRetryMaxAttempts ?? 1));
|
|
const initialBackoffMs = positiveInteger(configured?.initialBackoffMs, positiveInteger(env.AGENTRUN_BACKEND_RETRY_INITIAL_BACKOFF_MS, options.backendRetryInitialBackoffMs ?? 1_000));
|
|
const maxBackoffMs = positiveInteger(configured?.maxBackoffMs, positiveInteger(env.AGENTRUN_BACKEND_RETRY_MAX_BACKOFF_MS, options.backendRetryMaxBackoffMs ?? 30_000));
|
|
return { maxAttempts: Math.max(1, Math.min(10, maxAttempts)), initialBackoffMs: Math.max(100, initialBackoffMs), maxBackoffMs: Math.max(100, maxBackoffMs) };
|
|
}
|
|
|
|
function shouldRetryBackendTurn(result: { terminalStatus: TerminalStatus; failureKind: FailureKind | null }, policy: BackendRetryPolicy, backendAttempt: number): boolean {
|
|
return result.terminalStatus !== "completed" && backendAttempt < policy.maxAttempts && isRetryableBackendFailure(result.failureKind);
|
|
}
|
|
|
|
function isRetryableBackendFailure(failureKind: FailureKind | null): boolean {
|
|
return failureKind === "backend-timeout" || failureKind === "provider-stream-disconnected" || failureKind === "provider-unavailable" || failureKind === "provider-rate-limited" || failureKind === "provider-http-error";
|
|
}
|
|
|
|
function backendRetryDelayMs(policy: BackendRetryPolicy, backendAttempt: number): number {
|
|
const exponent = Math.max(0, backendAttempt - 1);
|
|
const delay = policy.initialBackoffMs * (2 ** exponent);
|
|
return Math.min(policy.maxBackoffMs, delay);
|
|
}
|
|
|
|
function jsonRecordValue(value: unknown): JsonRecord | null {
|
|
return value && typeof value === "object" && !Array.isArray(value) ? value as JsonRecord : null;
|
|
}
|
|
|
|
function positiveInteger(value: unknown, fallback: number): number {
|
|
const parsed = typeof value === "number" ? value : typeof value === "string" && value.trim().length > 0 ? Number(value) : NaN;
|
|
if (!Number.isFinite(parsed)) return fallback;
|
|
return Math.max(1, Math.floor(parsed));
|
|
}
|
|
|
|
function startActiveTurnCommandWatch(api: RunnerManagerApi, runId: string, targetCommand: CommandRecord, attemptId: string, runnerId: string, control: BackendActiveTurnControl, pollIntervalMs: number | undefined, runnerLog: RunnerLogSink, runnerJobId: string | undefined, backendAttempt: number, resumeDurableTurn: boolean, durableReplayCommands: CommandRecord[]): { stop: () => void; replayBarrier: Promise<void> } {
|
|
let stopped = false;
|
|
let polling = false;
|
|
let replayFailed = false;
|
|
const seen = new Set<string>();
|
|
const replayBarrier = (async (): Promise<void> => {
|
|
try {
|
|
for (const command of durableReplayCommands) {
|
|
await replayDurableSteerCommand(api, runId, command, targetCommand.id, attemptId, runnerId, control, backendAttempt, resumeDurableTurn, runnerLog, runnerJobId);
|
|
seen.add(command.id);
|
|
}
|
|
} catch (error) {
|
|
replayFailed = true;
|
|
throw error;
|
|
}
|
|
})();
|
|
void replayBarrier.catch(() => undefined);
|
|
const intervalMs = normalizePollIntervalMs(pollIntervalMs);
|
|
const poll = async (): Promise<void> => {
|
|
if (stopped || polling || replayFailed) return;
|
|
polling = true;
|
|
try {
|
|
await replayBarrier;
|
|
if (stopped) return;
|
|
const commands = await listCurrentTurnControlCommands(api, runId, targetCommand.seq);
|
|
const controlCommands = commands.filter((item) => !seen.has(item.id) && (item.type === "steer" || item.type === "interrupt") && item.state === "pending");
|
|
for (const controlCommand of controlCommands) {
|
|
seen.add(controlCommand.id);
|
|
if (controlCommand.type === "interrupt") await handleInterruptCommand(api, runId, controlCommand, targetCommand.id, attemptId, runnerId, control, runnerLog, runnerJobId);
|
|
else await handleSteerCommand(api, runId, controlCommand, targetCommand.id, attemptId, runnerId, control, runnerLog, runnerJobId);
|
|
}
|
|
} catch {
|
|
// The active backend turn remains authoritative; missed control commands stay pending for the next poll.
|
|
} finally {
|
|
polling = false;
|
|
}
|
|
};
|
|
const timer = setInterval(() => { void poll(); }, intervalMs);
|
|
void poll();
|
|
return {
|
|
stop: () => {
|
|
stopped = true;
|
|
clearInterval(timer);
|
|
},
|
|
replayBarrier,
|
|
};
|
|
}
|
|
|
|
async function listCurrentTurnControlCommands(api: RunnerManagerApi, runId: string, targetCommandSeq: number): Promise<CommandRecord[]> {
|
|
const commands: CommandRecord[] = [];
|
|
let afterSeq = targetCommandSeq;
|
|
while (true) {
|
|
const batch = await api.listCommands(runId, { afterSeq, limit: 100 });
|
|
if (batch.length === 0) return commands;
|
|
for (const command of batch) {
|
|
if (command.type === "turn") return commands;
|
|
commands.push(command);
|
|
}
|
|
if (batch.length < 100) return commands;
|
|
afterSeq = batch[batch.length - 1]?.seq ?? afterSeq;
|
|
}
|
|
}
|
|
|
|
async function durableSteersForTurnResume(api: RunnerManagerApi, runId: string, targetCommandSeq: number): Promise<CommandRecord[]> {
|
|
const commands = await listCurrentTurnControlCommands(api, runId, targetCommandSeq);
|
|
return commands.filter((command) => command.type === "steer" && (command.state === "acknowledged" || command.state === "completed"));
|
|
}
|
|
|
|
async function replayDurableSteerCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, backendAttempt: number, resumeDurableTurn: boolean, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise<void> {
|
|
const prompt = steerPrompt(command.payload);
|
|
if (!prompt) return;
|
|
try {
|
|
await control.steer(prompt);
|
|
await appendBestEffort(api, runId, { type: "backend_status", payload: { phase: "turn/steer:replayed", commandId: command.id, commandType: "steer", commandSeq: command.seq, commandState: command.state, targetCommandId, deliveryState: "replayed-to-resumed-backend", backendAccepted: true, backendAttempt, resumeDurableTurn, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId, valuesPrinted: false } });
|
|
if (command.state === "acknowledged") {
|
|
await reportTerminalCommand(api, { runId, commandId: command.id, runnerId, runnerJobId: runnerJobId ?? null, attemptId, report: { terminalStatus: "completed", failureKind: null, failureMessage: null, threadId: control.threadId, turnId: control.turnId }, phase: "runner:steer:replayed", runnerLog });
|
|
}
|
|
} catch (error) {
|
|
await appendBestEffort(api, runId, { type: "error", payload: { phase: "turn/steer:replay-failed", commandId: command.id, commandType: "steer", commandSeq: command.seq, commandState: command.state, targetCommandId, failureKind: failureKindFromError(error), message: errorMessage(error), backendAttempt, resumeDurableTurn, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId, valuesPrinted: false } });
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
async function handleSteerCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise<void> {
|
|
const acked = await api.ackCommand(command.id);
|
|
if (acked.state === "cancelled") {
|
|
await reportTerminalCommand(api, { runId, commandId: command.id, runnerId, runnerJobId: runnerJobId ?? null, attemptId, report: { terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: "steer command cancelled before delivery", threadId: control.threadId, turnId: control.turnId }, phase: "runner:steer:cancelled", runnerLog });
|
|
return;
|
|
}
|
|
const prompt = steerPrompt(command.payload);
|
|
if (!prompt) {
|
|
await reportNonTerminalCommandFailure(api, runId, command.id, attemptId, runnerId, { terminalStatus: "blocked", failureKind: "schema-invalid", message: "steer command payload requires a non-empty prompt, message, or text" }, "runner:steer:payload", runnerLog, runnerJobId, control);
|
|
return;
|
|
}
|
|
await appendBestEffort(api, runId, { type: "backend_status", payload: { phase: "steer-command-acknowledged", commandId: command.id, commandType: "steer", targetCommandId, deliveryState: "acknowledged-by-runner", backendAccepted: false, targetEffect: "not-yet-observed", attemptId, runnerId, threadId: control.threadId, turnId: control.turnId } });
|
|
try {
|
|
await control.steer(prompt);
|
|
await appendBestEffort(api, runId, { type: "backend_status", payload: { phase: "turn/steer:completed", commandId: command.id, commandType: "steer", targetCommandId, deliveryState: "forwarded-to-backend", backendAccepted: true, targetEffect: "not-guaranteed", semantics: "turn/steer RPC returned; target turn liveness must be read from the target command result", attemptId, runnerId, threadId: control.threadId, turnId: control.turnId } });
|
|
await reportTerminalCommand(api, { runId, commandId: command.id, runnerId, runnerJobId: runnerJobId ?? null, attemptId, report: { terminalStatus: "completed", failureKind: null, failureMessage: null, threadId: control.threadId, turnId: control.turnId }, phase: "runner:steer", runnerLog });
|
|
} catch (error) {
|
|
const failureKind = failureKindFromError(error);
|
|
await reportNonTerminalCommandFailure(api, runId, command.id, attemptId, runnerId, { terminalStatus: terminalStatusForFailure(failureKind), failureKind, message: errorMessage(error) }, "runner:steer", runnerLog, runnerJobId, control);
|
|
}
|
|
}
|
|
|
|
async function handleInterruptCommand(api: RunnerManagerApi, runId: string, command: CommandRecord, targetCommandId: string, attemptId: string, runnerId: string, control: BackendActiveTurnControl, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise<void> {
|
|
const acked = await api.ackCommand(command.id);
|
|
if (acked.state === "cancelled") {
|
|
await reportTerminalCommand(api, { runId, commandId: command.id, runnerId, runnerJobId: runnerJobId ?? null, attemptId, report: { terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: "interrupt command cancelled before delivery", threadId: control.threadId, turnId: control.turnId }, phase: "runner:interrupt:cancelled", runnerLog });
|
|
return;
|
|
}
|
|
const reason = interruptReason(command.payload);
|
|
await appendBestEffort(api, runId, { type: "backend_status", payload: { phase: "interrupt-command-acknowledged", commandId: command.id, commandType: "interrupt", targetCommandId, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId, reason } });
|
|
try {
|
|
await control.interrupt();
|
|
await appendBestEffort(api, runId, { type: "backend_status", payload: { phase: "turn/interrupt:completed", commandId: command.id, commandType: "interrupt", targetCommandId, attemptId, runnerId, threadId: control.threadId, turnId: control.turnId, reason } });
|
|
await reportTerminalCommand(api, { runId, commandId: command.id, runnerId, runnerJobId: runnerJobId ?? null, attemptId, report: { terminalStatus: "completed", failureKind: null, failureMessage: null, threadId: control.threadId, turnId: control.turnId }, phase: "runner:interrupt", runnerLog });
|
|
} catch (error) {
|
|
const failureKind = failureKindFromError(error);
|
|
await reportNonTerminalCommandFailure(api, runId, command.id, attemptId, runnerId, { terminalStatus: terminalStatusForFailure(failureKind), failureKind, message: errorMessage(error) }, "runner:interrupt", runnerLog, runnerJobId, control);
|
|
}
|
|
}
|
|
|
|
async function failPendingSteerWithoutActiveTurn(api: RunnerManagerApi, runId: string, commands: CommandRecord[], runnerId: string, attemptId: string, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise<void> {
|
|
for (const command of commands.filter((item) => (item.type === "steer" || item.type === "interrupt") && item.state === "pending")) {
|
|
const acked = await api.ackCommand(command.id);
|
|
if (acked.state === "cancelled") continue;
|
|
await reportNonTerminalCommandFailure(api, runId, command.id, attemptId, runnerId, { terminalStatus: "blocked", failureKind: "schema-invalid", message: `${command.type} command requires an active turn` }, `runner:${command.type}:no-active-turn`, runnerLog, runnerJobId);
|
|
}
|
|
}
|
|
|
|
async function reportNonTerminalCommandFailure(api: RunnerManagerApi, runId: string, commandId: string, attemptId: string, runnerId: string, failure: { terminalStatus: TerminalStatus; failureKind: FailureKind; message: string }, phase: string, runnerLog: RunnerLogSink, runnerJobId: string | undefined, control?: BackendActiveTurnControl): Promise<void> {
|
|
const event: BackendEvent = { type: "error", payload: { failureKind: failure.failureKind, message: failure.message, phase, commandId, attemptId, runnerId, ...(control ? { threadId: control.threadId, turnId: control.turnId } : {}) } };
|
|
await reportTerminalCommand(api, { runId, commandId, runnerId, runnerJobId: runnerJobId ?? null, attemptId, report: { terminalStatus: failure.terminalStatus, failureKind: failure.failureKind, failureMessage: failure.message, ...(control ? { threadId: control.threadId, turnId: control.turnId } : {}) }, events: [event], phase, runnerLog });
|
|
}
|
|
|
|
function failureDetailsFromError(error: unknown): JsonRecord | null {
|
|
return error instanceof AgentRunError ? error.details : null;
|
|
}
|
|
|
|
function steerPrompt(payload: JsonRecord): string | null {
|
|
for (const key of ["prompt", "message", "text"]) {
|
|
const value = payload[key];
|
|
if (typeof value === "string" && value.trim().length > 0) return value.trim();
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function interruptReason(payload: JsonRecord): string | null {
|
|
const value = payload.reason ?? payload.message ?? payload.prompt ?? payload.text;
|
|
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
|
|
}
|
|
|
|
function isTerminalRun(run: RunRecord): boolean {
|
|
return run.status === "completed" || run.status === "failed" || run.status === "blocked" || run.status === "cancelled";
|
|
}
|
|
|
|
async function assertNotCancelled(api: RunnerManagerApi, runId: string, commandId: string): Promise<void> {
|
|
const [run, command] = await Promise.all([api.getRun(runId), api.getCommand(runId, commandId)]);
|
|
if (run.status === "cancelled" || command.state === "cancelled") throw new AgentRunError("cancelled", "run or command was cancelled", { httpStatus: 409 });
|
|
}
|
|
|
|
function watchCancellation(api: RunnerManagerApi, runId: string, commandId: string, controller: AbortController, context: { attemptId: string; runnerId: string; runnerJobId?: string; runnerLog: RunnerLogSink }): () => void {
|
|
let stopped = false;
|
|
let reported = false;
|
|
const check = async (): Promise<void> => {
|
|
if (stopped || controller.signal.aborted) return;
|
|
try {
|
|
const [run, command] = await Promise.all([api.getRun(runId), api.getCommand(runId, commandId)]);
|
|
if (run.status === "cancelled" || command.state === "cancelled") {
|
|
if (!reported) {
|
|
reported = true;
|
|
const cancelRequestId = command.cancelRequestId ?? run.cancelRequestId;
|
|
const cancelEpoch = command.cancelEpoch || run.cancelEpoch;
|
|
const reason = command.cancelReason ?? run.cancelReason ?? "cancel requested";
|
|
const base = { commandId, attemptId: context.attemptId, runnerId: context.runnerId, runnerJobId: context.runnerJobId ?? null, cancelRequestId, cancelEpoch, reason, source: "runner-watch" };
|
|
await appendBestEffort(api, runId, { type: "backend_status", payload: { ...base, phase: "cancel-delivered", cancelStage: "delivered", deliveryState: "observed-by-runner", runStatus: run.status, commandState: command.state } });
|
|
await appendBestEffort(api, runId, { type: "backend_status", payload: { ...base, phase: "cancel-aborting", cancelStage: "aborting", abortSignal: "triggered" } });
|
|
await context.runnerLog.write("cancel.observed", { runId, ...base, valuesPrinted: false });
|
|
}
|
|
controller.abort();
|
|
}
|
|
} catch {
|
|
// Cancellation polling must not hide the backend's own terminal result.
|
|
}
|
|
};
|
|
const timer = setInterval(() => { void check(); }, 2_000);
|
|
void check();
|
|
return () => {
|
|
stopped = true;
|
|
clearInterval(timer);
|
|
};
|
|
}
|
|
|
|
function startHeartbeat(api: RunnerManagerApi, runId: string, runnerId: string, leaseMs: number): () => void {
|
|
let stopped = false;
|
|
const beat = async (): Promise<void> => {
|
|
if (stopped) return;
|
|
try {
|
|
await api.heartbeat(runId, runnerId, leaseMs);
|
|
} catch {
|
|
// The next manager call will surface lease or run-terminal details.
|
|
}
|
|
};
|
|
const timer = setInterval(() => { void beat(); }, Math.max(1_000, Math.floor(leaseMs / 3)));
|
|
return () => {
|
|
stopped = true;
|
|
clearInterval(timer);
|
|
};
|
|
}
|
|
|
|
function startBackendProgress(): { stop: () => JsonRecord } {
|
|
let stopped = false;
|
|
let ticks = 0;
|
|
const startedAt = Date.now();
|
|
const tick = (): void => {
|
|
if (stopped) return;
|
|
ticks += 1;
|
|
};
|
|
const timer = setInterval(tick, 10_000);
|
|
return {
|
|
stop: () => {
|
|
stopped = true;
|
|
clearInterval(timer);
|
|
return { elapsedMs: Date.now() - startedAt, progressTicks: ticks, progressEventsPrinted: false };
|
|
},
|
|
};
|
|
}
|
|
|
|
async function appendBestEffort(api: RunnerManagerApi, runId: string, event: BackendEvent): Promise<void> {
|
|
try {
|
|
await api.appendEvent(runId, event);
|
|
} catch {
|
|
// Visibility events are best effort; terminal command reporting remains authoritative.
|
|
}
|
|
}
|
|
|
|
async function claimRunWithLeaseRecovery(api: RunnerManagerApi, options: RunnerOnceOptions, runner: RunnerRecord, attemptId: string, leaseMs: number): Promise<RunRecord> {
|
|
const startedAt = Date.now();
|
|
const timeoutMs = normalizeClaimRetryTimeoutMs(options.claimRetryTimeoutMs, leaseMs);
|
|
const intervalMs = normalizeClaimRetryIntervalMs(options.claimRetryIntervalMs);
|
|
let waitingEventWritten = false;
|
|
let lastError: unknown = null;
|
|
|
|
while (Date.now() - startedAt <= timeoutMs) {
|
|
try {
|
|
const claimed = await api.claim(options.runId, runner.id, leaseMs);
|
|
if (waitingEventWritten) {
|
|
await appendBestEffort(api, options.runId, {
|
|
type: "backend_status",
|
|
payload: { phase: "runner-claim-recovered", attemptId, runnerId: runner.id, waitedMs: Date.now() - startedAt },
|
|
});
|
|
}
|
|
return claimed;
|
|
} catch (error) {
|
|
if (failureKindFromError(error) !== "runner-lease-conflict") throw error;
|
|
lastError = error;
|
|
const run = await getRunBestEffort(api, options.runId);
|
|
if (!waitingEventWritten) {
|
|
waitingEventWritten = true;
|
|
await appendBestEffort(api, options.runId, {
|
|
type: "backend_status",
|
|
payload: {
|
|
phase: "runner-claim-waiting-for-stale-lease",
|
|
attemptId,
|
|
runnerId: runner.id,
|
|
claimedBy: run?.claimedBy ?? null,
|
|
leaseExpiresAt: run?.leaseExpiresAt ?? null,
|
|
retryTimeoutMs: timeoutMs,
|
|
},
|
|
});
|
|
}
|
|
const remainingMs = timeoutMs - (Date.now() - startedAt);
|
|
if (remainingMs <= 0) break;
|
|
await sleep(Math.min(remainingMs, claimRetryDelayMs(run, intervalMs)));
|
|
}
|
|
}
|
|
|
|
throw lastError ?? new AgentRunError("runner-lease-conflict", `run ${options.runId} could not be claimed before retry timeout`, { httpStatus: 409 });
|
|
}
|
|
|
|
async function getRunBestEffort(api: RunnerManagerApi, runId: string): Promise<RunRecord | null> {
|
|
try {
|
|
return await api.getRun(runId);
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function claimRetryDelayMs(run: RunRecord | null, intervalMs: number): number {
|
|
const leaseExpiresAt = run?.leaseExpiresAt ? Date.parse(run.leaseExpiresAt) : NaN;
|
|
if (Number.isFinite(leaseExpiresAt)) {
|
|
const untilExpiryMs = leaseExpiresAt - Date.now() + 100;
|
|
if (untilExpiryMs > 0) return Math.max(25, Math.min(intervalMs, untilExpiryMs));
|
|
}
|
|
return intervalMs;
|
|
}
|
|
|
|
function normalizeClaimRetryTimeoutMs(value: number | undefined, leaseMs: number): number {
|
|
if (Number.isFinite(value ?? NaN)) return Math.max(0, Math.floor(value!));
|
|
return Math.max(leaseMs + Math.max(5_000, Math.floor(leaseMs / 3)), leaseMs);
|
|
}
|
|
|
|
function normalizeClaimRetryIntervalMs(value: number | undefined): number {
|
|
if (!Number.isFinite(value ?? NaN)) return 5_000;
|
|
return Math.max(25, Math.min(30_000, Math.floor(value!)));
|
|
}
|
|
|
|
function annotateCommandEvent(event: BackendEvent, commandId: string, attemptId: string, runnerId: string): BackendEvent {
|
|
return { ...event, payload: { ...event.payload, commandId, attemptId, runnerId } };
|
|
}
|
|
|
|
async function reportTerminalCommand(api: RunnerManagerApi, input: { runId: string; commandId: string; runnerId: string; runnerJobId: string | null; attemptId: string; report: CommandTerminalReport; events?: BackendEvent[]; phase: string; runnerLog: RunnerLogSink; terminalRun?: boolean }): Promise<void> {
|
|
const outboxKey = terminalOutboxKey(input);
|
|
const events = (input.events ?? []).map((event, index) => annotateTerminalOutboxEvent(event, input.commandId, input.attemptId, input.runnerId, outboxKey, index));
|
|
const outboxPayload: JsonRecord = {
|
|
schema: "agentrun-terminal-outbox-v1",
|
|
outboxKey,
|
|
phase: input.phase,
|
|
runId: input.runId,
|
|
commandId: input.commandId,
|
|
runnerId: input.runnerId,
|
|
runnerJobId: input.runnerJobId,
|
|
attemptId: input.attemptId,
|
|
report: terminalReportJson(input.report),
|
|
terminalRun: input.terminalRun === true,
|
|
events: events.map((event) => ({ type: event.type, payload: event.payload })),
|
|
eventCount: events.length,
|
|
valuesPrinted: false,
|
|
};
|
|
await writeTerminalOutbox(input.runnerLog, outboxPayload);
|
|
for (const event of events) await appendBestEffort(api, input.runId, event);
|
|
try {
|
|
await api.reportCommandStatus(input.commandId, input.report);
|
|
if (input.terminalRun === true) await api.reportStatus(input.runId, input.report);
|
|
} catch (error) {
|
|
throw new TerminalOutboxReportError(error);
|
|
}
|
|
await input.runnerLog.write("terminal.outbox.ack", { runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, runnerJobId: input.runnerJobId, attemptId: input.attemptId, outboxKey, valuesPrinted: false });
|
|
}
|
|
|
|
function terminalOutboxKey(input: { runId: string; commandId: string; runnerId: string; attemptId: string; phase: string; report: CommandTerminalReport; terminalRun?: boolean }): string {
|
|
return stableHash({ schema: "agentrun-terminal-outbox-v1", runId: input.runId, commandId: input.commandId, runnerId: input.runnerId, attemptId: input.attemptId, phase: input.phase, terminalRun: input.terminalRun === true, report: terminalReportJson(input.report) });
|
|
}
|
|
|
|
function terminalReportJson(report: CommandTerminalReport): JsonRecord {
|
|
return { terminalStatus: report.terminalStatus, failureKind: report.failureKind, failureMessage: report.failureMessage, threadId: report.threadId ?? null, turnId: report.turnId ?? null };
|
|
}
|
|
|
|
function annotateTerminalOutboxEvent(event: BackendEvent, commandId: string, attemptId: string, runnerId: string, outboxKey: string, index: number): BackendEvent {
|
|
const annotated = annotateCommandEvent(event, commandId, attemptId, runnerId);
|
|
return { ...annotated, payload: { ...annotated.payload, terminalOutboxKey: outboxKey, terminalOutboxIndex: index } };
|
|
}
|
|
|
|
async function writeTerminalOutbox(runnerLog: RunnerLogSink, payload: JsonRecord): Promise<void> {
|
|
await runnerLog.write("terminal.outbox", payload);
|
|
try {
|
|
process.stdout.write(JSON.stringify({ at: new Date().toISOString(), label: "terminal.outbox", ...payload, valuesPrinted: false }) + "\n");
|
|
} catch {
|
|
// Kubernetes stdout is the recovery path; if it is unavailable, the file log remains diagnostic evidence.
|
|
}
|
|
}
|
|
|
|
function shouldKeepTerminalOutboxEvent(event: BackendEvent): boolean {
|
|
if (event.type === "terminal_status") return true;
|
|
if (event.type === "backend_status") return event.payload.phase === "turn/completed" || event.payload.finalSeal === true;
|
|
if (event.type !== "assistant_message") return false;
|
|
return event.payload.final === true || event.payload.replyAuthority === true;
|
|
}
|
|
|
|
async function reportCommandFailure(api: RunnerManagerApi, runId: string, commandId: string, runner: RunnerRecord, attemptId: string, failure: RunnerFailure, phase: string, runnerLog: RunnerLogSink, options: { terminalRun?: boolean; runnerJobId?: string; events?: BackendEvent[] } = {}): Promise<CommandExecutionResult> {
|
|
const details = failure.details ? { details: failure.details } : {};
|
|
const events: BackendEvent[] = [
|
|
...(options.events ?? []),
|
|
{ type: "error", payload: { failureKind: failure.failureKind, message: failure.message, phase, commandId, attemptId, runnerId: runner.id, ...details } },
|
|
{ type: "terminal_status", payload: { terminalStatus: failure.terminalStatus, failureKind: failure.failureKind, message: failure.message, commandId, attemptId, runnerId: runner.id, ...details } },
|
|
];
|
|
await reportTerminalCommand(api, { runId, commandId, runnerId: runner.id, runnerJobId: options.runnerJobId ?? null, attemptId, report: { terminalStatus: failure.terminalStatus, failureKind: failure.failureKind, failureMessage: failure.message }, events, phase, runnerLog, terminalRun: options.terminalRun === true });
|
|
return { commandId, terminalStatus: failure.terminalStatus, failureKind: failure.failureKind } as CommandExecutionResult;
|
|
}
|
|
|
|
async function reportCancelled(api: RunnerManagerApi, runId: string, commandId: string, runner: RunnerRecord, attemptId: string, message: string, runnerLog: RunnerLogSink, runnerJobId: string | undefined): Promise<CommandExecutionResult> {
|
|
const events: BackendEvent[] = [
|
|
{ type: "backend_status", payload: { phase: "turn-cancelled", commandId, attemptId, runnerId: runner.id, failureKind: "cancelled", message } },
|
|
{ type: "terminal_status", payload: { terminalStatus: "cancelled", failureKind: "cancelled", message, commandId, attemptId, runnerId: runner.id } },
|
|
];
|
|
await reportTerminalCommand(api, { runId, commandId, runnerId: runner.id, runnerJobId: runnerJobId ?? null, attemptId, report: { terminalStatus: "cancelled", failureKind: "cancelled", failureMessage: message }, events, phase: "runner:cancelled", runnerLog, terminalRun: true });
|
|
return { commandId, terminalStatus: "cancelled", failureKind: "cancelled" };
|
|
}
|
|
|
|
function noPendingResult(runner: RunnerRecord, claimed: RunRecord, commandResults: CommandExecutionResult[], polledCommands: number, stopped: string): JsonRecord {
|
|
if (commandResults.length > 0) {
|
|
const last = commandResults.at(-1)!;
|
|
return { runner, claimed, terminalStatus: last.terminalStatus, failureKind: last.failureKind, commandsProcessed: commandResults.length, commandResults, polledCommands, stopped };
|
|
}
|
|
return { runner, claimed, terminalStatus: "blocked", failureKind: "schema-invalid", commandsProcessed: 0, commandResults, polledCommands, stopped };
|
|
}
|
|
|
|
function lastCommandSeq(commands: CommandRecord[]): number {
|
|
let seq = 0;
|
|
for (const command of commands) seq = Math.max(seq, command.seq);
|
|
return seq;
|
|
}
|
|
|
|
function sleep(ms: number): Promise<void> {
|
|
return new Promise((resolve) => setTimeout(resolve, ms));
|
|
}
|
|
|
|
function normalizePollIntervalMs(value: number | undefined): number {
|
|
if (!Number.isFinite(value ?? NaN)) return 250;
|
|
return Math.max(50, Math.min(2_000, Math.floor(value!)));
|
|
}
|
|
|
|
async function createRunnerLogSink(logPath: string | undefined): Promise<RunnerLogSink> {
|
|
const normalized = logPath?.trim();
|
|
if (!normalized) return { write: async () => undefined };
|
|
try {
|
|
await mkdir(path.dirname(normalized), { recursive: true });
|
|
await writeFile(normalized, "", { flag: "a" });
|
|
} catch {
|
|
return { write: async () => undefined };
|
|
}
|
|
return {
|
|
write: async (label: string, payload: JsonRecord) => {
|
|
try {
|
|
const line = JSON.stringify({ at: new Date().toISOString(), label, ...payload, valuesPrinted: false }) + "\n";
|
|
await appendFile(normalized, line, "utf8");
|
|
} catch {
|
|
// Local runner log is diagnostic-only; manager events and terminal state remain authoritative.
|
|
}
|
|
},
|
|
};
|
|
}
|
|
|
|
function runnerLogEventSummary(event: BackendEvent, runId: string, commandId: string, attemptId: string, runnerId: string, runnerJobId: string | null): JsonRecord {
|
|
const payload = event.payload ?? {};
|
|
return {
|
|
runId,
|
|
commandId,
|
|
runnerId,
|
|
runnerJobId,
|
|
attemptId,
|
|
eventType: event.type,
|
|
phase: stringPayload(payload, "phase"),
|
|
terminalStatus: stringPayload(payload, "terminalStatus"),
|
|
failureKind: stringPayload(payload, "failureKind"),
|
|
threadId: stringPayload(payload, "threadId"),
|
|
turnId: stringPayload(payload, "turnId"),
|
|
itemId: stringPayload(payload, "itemId"),
|
|
itemType: stringPayload(payload, "type"),
|
|
method: stringPayload(payload, "method"),
|
|
valuesPrinted: false,
|
|
};
|
|
}
|
|
|
|
function stringPayload(record: JsonRecord, key: string): string | null {
|
|
const value = record[key];
|
|
return typeof value === "string" && value.trim().length > 0 ? value.trim() : null;
|
|
}
|