diff --git a/package.json b/package.json index bf6737d..17e65e9 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,7 @@ "check": "tsc --noEmit", "self-test": "bun run src/selftest/run.ts", "self-test:postgres-kafka-order": "bun run src/selftest/integration/postgres-kafka-order.ts", + "self-test:postgres-queue-retry": "bun run src/selftest/integration/postgres-queue-retry.ts", "self-test:postgres-runner-retention-fence": "bun run src/selftest/integration/postgres-runner-retention-fence.ts", "test": "bun run src/selftest/run.ts", "cli": "bun scripts/agentrun-cli.ts" diff --git a/scripts/src/cli.ts b/scripts/src/cli.ts index ebdf2a5..713d61f 100644 --- a/scripts/src/cli.ts +++ b/scripts/src/cli.ts @@ -115,6 +115,8 @@ async function dispatch(args: ParsedArgs): Promise { if (group === "queue" && command === "read" && id) return readQueueTask(args, id); if (group === "queue" && command === "cancel" && id) return cancelQueueTask(args, id); if (group === "queue" && command === "dispatch" && id) return dispatchQueueTask(args, id); + if (group === "queue" && command === "retry" && id) return retryQueueTaskCli(args, id); + if (group === "queue" && command === "attempts" && id) return listQueueAttemptsCli(args, id); if (group === "queue" && command === "refresh" && id) return refreshQueueTask(args, id); if (group === "runs" && command === "create") return client(args).post("/api/v1/runs", await jsonFile(args)); if (group === "runs" && command === "show" && id) return showRun(args, id); @@ -573,6 +575,61 @@ export function summarizeQueueDispatchResult(result: JsonValue, taskId: string): }; } +export function summarizeQueueRetryResult(result: JsonValue, taskId: string): JsonRecord { + const record = jsonRecordValue(result); + if (!record) throw new AgentRunError("schema-invalid", "queue retry response must be an object", { httpStatus: 2 }); + const task = jsonRecordValue(record.task); + const attempt = jsonRecordValue(record.attempt); + const run = jsonRecordValue(record.run); + const command = jsonRecordValue(record.command); + const runnerJob = jsonRecordValue(record.runnerJob); + const runId = stringValue(run?.id) ?? stringValue(attempt?.runId); + const commandId = stringValue(command?.id) ?? stringValue(attempt?.commandId); + const sessionId = stringValue(attempt?.sessionId) ?? stringValue(jsonRecordValue(task?.sessionRef)?.sessionId); + return { + action: "queue-retry-summary", + taskId: stringValue(task?.id) ?? taskId, + mutation: record.mutation === true, + idempotentReplay: record.idempotentReplay === true, + task: summarizeQueueTaskRecord(task, taskId), + attempt: summarizeAttemptRecord(attempt), + previousAttempt: summarizeAttemptRecord(jsonRecordValue(record.previousAttempt)), + allowedTransition: jsonRecordValue(record.allowedTransition), + run: summarizeRunRecord(run), + command: summarizeCommandRecord(command), + runnerJob: summarizeRunnerJobRecord(runnerJob), + fullResponseBytes: jsonByteLength(result), + valuesPrinted: false, + pollCommands: { + attempts: `./scripts/agentrun queue attempts ${taskId}`, + queue: `describe task/${taskId}`, + ...(runId ? { run: `describe run/${runId}`, result: `result run/${runId}${commandId ? ` --command ${commandId}` : ""}`, events: `events run/${runId} --after-seq 0 --limit 100 --tail-summary` } : {}), + ...(runId && commandId ? { command: `describe command/${commandId} --run ${runId}` } : {}), + ...(sessionId ? { logs: `logs session/${sessionId} --tail 100` } : {}), + }, + }; +} + +export function summarizeQueueAttemptListResult(result: JsonValue, taskId: string, limit: number): JsonRecord { + const record = jsonRecordValue(result); + if (!record) throw new AgentRunError("schema-invalid", "queue attempts response must be an object", { httpStatus: 2 }); + const items = queueItems(record.items); + return { + action: "queue-attempt-list-summary", + taskId: stringValue(record.taskId) ?? taskId, + count: numberValue(record.count) ?? items.length, + cursor: stringValue(record.cursor), + limit, + items: items.slice(0, limit).map((attempt) => withoutFullRecordBytes(compactRecord(attempt, { keys: ["attemptId", "taskId", "retryIndex", "state", "idempotencyKey", "reason", "payloadHash", "previousAttemptId", "runId", "commandId", "runnerJobId", "sessionId", "sessionPath", "failureKind", "failureMessage", "createdAt", "updatedAt", "activatedAt", "terminalAt"] }))), + fullResponseBytes: jsonByteLength(result), + valuesPrinted: false, + drillDownCommands: { + full: `./scripts/agentrun queue attempts ${taskId} --full`, + next: stringValue(record.cursor) ? `./scripts/agentrun queue attempts ${taskId} --cursor ${String(record.cursor)} --limit ${limit}` : null, + }, + }; +} + function summarizeRunEvent(event: JsonRecord, summaryChars: number): JsonRecord { const payload = jsonRecordValue(event.payload) ?? {}; const type = stringValue(event.type) ?? "unknown"; @@ -1604,6 +1661,28 @@ async function dispatchQueueTask(args: ParsedArgs, taskId: string): Promise { + const idempotencyKey = optionalFlag(args, "idempotency-key"); + const reason = optionalFlag(args, "reason"); + if (!idempotencyKey) throw new AgentRunError("schema-invalid", "queue retry requires --idempotency-key", { httpStatus: 2 }); + if (!reason) throw new AgentRunError("schema-invalid", "queue retry requires --reason", { httpStatus: 2 }); + const body: JsonRecord = { idempotencyKey, reason, dryRun: args.flags.get("dry-run") === true }; + const result = await client(args).post(`/api/v1/queue/tasks/${encodeURIComponent(taskId)}/retry`, body); + if (wantsExpandedOutput(args)) return result; + return summarizeQueueRetryResult(result, taskId); +} + +async function listQueueAttemptsCli(args: ParsedArgs, taskId: string): Promise { + const params = new URLSearchParams(); + const cursor = optionalFlag(args, "cursor"); + const limit = integerFlag(args, "limit", 20, { min: 1, max: 100 }); + if (cursor) params.set("cursor", cursor); + params.set("limit", String(limit)); + const result = await client(args).get(`/api/v1/queue/tasks/${encodeURIComponent(taskId)}/attempts?${params.toString()}`); + if (wantsExpandedOutput(args)) return result; + return summarizeQueueAttemptListResult(result, taskId, limit); +} + async function queueDispatchBody(args: ParsedArgs): Promise { const body = await optionalJsonFile(args); const copy = (flagName: string, key = flagName.replace(/-([a-z])/gu, (_, letter: string) => letter.toUpperCase())): void => { @@ -2247,6 +2326,8 @@ function help(args: ParsedArgs, group?: string): JsonRecord { "queue read [--reader-id ] [--dry-run] [--full|--raw]", "queue cancel [--reason ] [--dry-run] [--full|--raw]", "queue dispatch [--json-stdin|--json-file ] [--idempotency-key ] [--image ] [--namespace ] [--dry-run] [--full|--raw]", + "queue retry --idempotency-key --reason [--dry-run] [--full|--raw]", + "queue attempts [--cursor ] [--limit ] [--full|--raw]", "queue refresh [--dry-run] [--full|--raw]", "aipod-specs list", "aipod-specs show ", diff --git a/src/common/types.ts b/src/common/types.ts index 5ad5386..038d275 100644 --- a/src/common/types.ts +++ b/src/common/types.ts @@ -38,6 +38,7 @@ export type CancelTargetKind = "task" | "session" | "run" | "command"; export type CancelStage = "accepted" | "persisted" | "delivered" | "aborting" | "terminalized" | "fenced" | "late-write-rejected"; export type BackendProfile = string; export type QueueTaskState = "pending" | "running" | "completed" | "failed" | "blocked" | "cancelled"; +export type QueueAttemptState = "reserved" | QueueTaskState; export type QueueTaskAttentionState = "active" | "unread" | "read"; export type SessionExecutionState = "idle" | "running" | "terminal"; export type SessionAttentionState = "active" | "unread" | "read"; @@ -480,6 +481,70 @@ export interface QueueAttemptRef extends JsonRecord { runnerJobId: string | null; sessionId: string | null; sessionPath: string | null; + failureKind?: FailureKind | null; + failureMessage?: string | null; +} + +export interface QueueAttemptRecord extends JsonRecord { + attemptId: string; + taskId: string; + retryIndex: number; + state: QueueAttemptState; + idempotencyKey: string | null; + reason: string | null; + payloadHash: string; + previousAttemptId: string | null; + runId: string | null; + commandId: string | null; + runnerJobId: string | null; + sessionId: string | null; + sessionPath: string | null; + failureKind: FailureKind | null; + failureMessage: string | null; + createdAt: string; + updatedAt: string; + activatedAt: string | null; + terminalAt: string | null; +} + +export interface QueueAttemptListResult extends JsonRecord { + taskId: string; + items: QueueAttemptRecord[]; + count: number; + cursor: string | null; +} + +export interface QueueRetryReservation extends JsonRecord { + action: "queue-retry-reservation"; + mutation: boolean; + idempotentReplay: boolean; + task: QueueTaskRecord; + attempt: QueueAttemptRecord | null; + previousAttempt: QueueAttemptRecord | null; + retryIndex: number; + allowedTransition: JsonRecord; +} + +export interface QueueRetryActivation extends JsonRecord { + task: QueueTaskRecord; + attempt: QueueAttemptRecord; + activated: boolean; +} + +export interface QueueRetryResult extends JsonRecord { + action: "queue-retry"; + mutation: boolean; + idempotentReplay: boolean; + task: QueueTaskRecord; + attempt: QueueAttemptRecord | null; + previousAttempt: QueueAttemptRecord | null; + run: RunRecord | null; + command: CommandRecord | null; + runnerJob: JsonRecord | null; + envImage: JsonRecord | null; + workReady: JsonRecord | null; + allowedTransition: JsonRecord; + pollActions: JsonRecord[]; } export interface CreateQueueTaskInput extends JsonRecord { diff --git a/src/mgr/kubernetes-runner-job.ts b/src/mgr/kubernetes-runner-job.ts index ebdf63e..a31476a 100644 --- a/src/mgr/kubernetes-runner-job.ts +++ b/src/mgr/kubernetes-runner-job.ts @@ -79,9 +79,23 @@ export interface CreateRunnerJobInput extends JsonRecord { } export async function createKubernetesRunnerJob(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults; signal?: AbortSignal }): Promise { - if (!options.defaults.retention) return await createKubernetesRunnerJobUnderFence(options); const namespace = optionalString(options.input.namespace) ?? options.defaults.namespace; - return await options.store.withRunnerCreateFence(namespace, async () => await createKubernetesRunnerJobUnderFence(options)); + const fenceKey = runnerCreateFenceKey({ runId: options.runId, input: options.input, namespace, retentionEnabled: options.defaults.retention !== undefined }); + return await options.store.withRunnerCreateFence(fenceKey, async () => await createKubernetesRunnerJobUnderFence(options)); +} + +export function runnerCreateFenceKey(options: { runId: string; input: CreateRunnerJobInput; namespace: string; retentionEnabled: boolean }): string { + if (options.retentionEnabled) return options.namespace; + const idempotencyKey = optionalString(options.input.idempotencyKey); + const identity = idempotencyKey + ? { runId: options.runId, idempotencyKey } + : { + runId: options.runId, + commandId: stringField(options.input, "commandId"), + attemptId: optionalString(options.input.attemptId) ?? null, + runnerJobId: optionalString(options.input.runnerJobId) ?? null, + }; + return `request:${stableHash(identity)}`; } async function createKubernetesRunnerJobUnderFence(options: { store: AgentRunStore; runId: string; input: CreateRunnerJobInput; defaults: RunnerJobDefaults; signal?: AbortSignal }): Promise { @@ -141,7 +155,7 @@ async function createKubernetesRunnerJobUnderFence(options: { store: AgentRunSto if (idempotencyKey) { const existing = await options.store.getRunnerJobByIdempotencyKey(run.id, idempotencyKey, payloadHash); throwIfAborted(options.signal, "runner dispatch attempt"); - if (existing) return { ...existing.result, idempotentReplay: true }; + if (existing) return { ...existing.result, mutation: false, idempotentReplay: true }; } if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${run.id} is already terminal: ${run.status}`, { httpStatus: 409 }); if (isTerminalCommandState(command.state) || command.state !== "pending") throw new AgentRunError(command.state === "cancelled" ? "cancelled" : "schema-invalid", `command ${commandId} is not pending: ${command.state}`, { httpStatus: 409 }); diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 5d05a86..200abff 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -3,10 +3,10 @@ import { Pool } from "pg"; import type { PoolClient, QueryResultRow } from "pg"; import { AgentRunError } from "../common/errors.js"; import { redactJson } from "../common/redaction.js"; -import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js"; +import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueAttemptListResult, QueueAttemptRecord, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueRetryActivation, QueueRetryReservation, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js"; import { newId, nowIso, stableHash } from "../common/validation.js"; -import type { AgentRunStore, DurableQueueClaim, EventOutboxConfig, ListQueueTasksInput, ListSessionsInput, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js"; -import { assertSessionBoundary, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, lateWriteRejectedPayload, parseQueueCursor, parseSessionCursor, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, runCreatedEventPayload, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeSessionRef, titleFromMetadata } from "./store.js"; +import type { ActivateQueueRetryAttemptInput, AgentRunStore, CreateRunIdentity, DurableQueueClaim, EventOutboxConfig, ListQueueAttemptsInput, ListQueueTasksInput, ListSessionsInput, RecordQueueRetryAttemptFailureInput, ReserveQueueRetryAttemptInput, RunnerRetentionFenceFacts, RunnerRetentionFenceInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js"; +import { assertQueueAttemptActivationReplay, assertQueueRetryable, assertQueueTaskPayloadHash, assertRunCreateReplay, assertSessionBoundary, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, cancelStagePayload, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, fenceLateEventForCancelledRun, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueAttemptState, isTerminalQueueTaskState, isTerminalRunStatus, latestQueueAttempt, lateWriteRejectedPayload, nextQueueRetryIndex, parseQueueCursor, parseSessionCursor, queueAttemptRef, queueRetryReservation, queueRetryStateError, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, runCreatedEventPayload, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeSessionRef, titleFromMetadata } from "./store.js"; import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js"; import { normalizeRunEventPayload, requireEventType, requireExternallyAppendableEventType, userMessagePayloadForCommand } from "../common/events.js"; import { buildKafkaEventOutboxRecord, canonicalAgentRunEventPartitionKey } from "./event-outbox.js"; @@ -373,6 +373,59 @@ CREATE INDEX IF NOT EXISTS agentrun_cancel_requests_run_idx ON agentrun_cancel_r CREATE INDEX IF NOT EXISTS agentrun_cancel_requests_command_idx ON agentrun_cancel_requests (command_id, created_at DESC); `; +const queueAttemptsMigrationSql = ` +CREATE TABLE IF NOT EXISTS agentrun_queue_attempts ( + attempt_id text NOT NULL, + task_id text NOT NULL REFERENCES agentrun_queue_tasks(id) ON DELETE CASCADE, + retry_index integer NOT NULL CHECK (retry_index >= 0), + state text NOT NULL CHECK (state IN ('reserved', 'pending', 'running', 'completed', 'failed', 'blocked', 'cancelled')), + idempotency_key text, + reason text, + payload_hash text NOT NULL, + previous_attempt_id text, + run_id text, + command_id text, + runner_job_id text, + session_id text, + session_path text, + failure_kind text, + failure_message text, + created_at timestamptz NOT NULL, + updated_at timestamptz NOT NULL, + activated_at timestamptz, + terminal_at timestamptz, + PRIMARY KEY (task_id, attempt_id), + UNIQUE (task_id, retry_index) +); + +CREATE UNIQUE INDEX IF NOT EXISTS agentrun_queue_attempts_retry_key_idx + ON agentrun_queue_attempts (task_id, idempotency_key) + WHERE idempotency_key IS NOT NULL; + +CREATE UNIQUE INDEX IF NOT EXISTS agentrun_queue_attempts_active_idx + ON agentrun_queue_attempts (task_id) + WHERE state IN ('reserved', 'running'); + +CREATE INDEX IF NOT EXISTS agentrun_queue_attempts_history_idx + ON agentrun_queue_attempts (task_id, retry_index DESC, created_at DESC); + +INSERT INTO agentrun_queue_attempts ( + attempt_id, task_id, retry_index, state, idempotency_key, reason, payload_hash, + previous_attempt_id, run_id, command_id, runner_job_id, session_id, session_path, + failure_kind, failure_message, created_at, updated_at, activated_at, terminal_at +) +SELECT + latest_attempt->>'attemptId', id, 0, latest_attempt->>'state', NULL, NULL, payload_hash, + NULL, latest_attempt->>'runId', latest_attempt->>'commandId', latest_attempt->>'runnerJobId', + latest_attempt->>'sessionId', latest_attempt->>'sessionPath', latest_attempt->>'failureKind', latest_attempt->>'failureMessage', + created_at, updated_at, updated_at, + CASE WHEN latest_attempt->>'state' IN ('completed', 'failed', 'blocked', 'cancelled') THEN updated_at ELSE NULL END +FROM agentrun_queue_tasks +WHERE latest_attempt IS NOT NULL + AND NULLIF(latest_attempt->>'attemptId', '') IS NOT NULL +ON CONFLICT (task_id, attempt_id) DO NOTHING; +`; + const postgresMigrations: MigrationDefinition[] = [ { id: "001_v01_initial_durable_store", @@ -439,13 +492,28 @@ const postgresMigrations: MigrationDefinition[] = [ checksum: checksumSql(runnerDispatchOutcomeMigrationSql), sql: runnerDispatchOutcomeMigrationSql, }, + { + id: "014_v02_queue_attempt_history_retry", + checksum: checksumSql(queueAttemptsMigrationSql), + sql: queueAttemptsMigrationSql, + }, ]; export function postgresMigrationContract(): JsonRecord { return { migrationIds: postgresMigrations.map((migration) => migration.id), latestMigrationId: latestMigrationId(), - requiredTables: ["agentrun_schema_migrations", "agentrun_runs", "agentrun_commands", "agentrun_events", "agentrun_runners", "agentrun_backends", "agentrun_leases", "agentrun_sessions", "agentrun_session_read_cursors", "agentrun_runner_jobs", "agentrun_runner_dispatch_intents", "agentrun_kafka_event_outbox", "agentrun_queue_tasks", "agentrun_queue_read_cursors", "agentrun_cancel_requests"], + requiredTables: ["agentrun_schema_migrations", "agentrun_runs", "agentrun_commands", "agentrun_events", "agentrun_runners", "agentrun_backends", "agentrun_leases", "agentrun_sessions", "agentrun_session_read_cursors", "agentrun_runner_jobs", "agentrun_runner_dispatch_intents", "agentrun_kafka_event_outbox", "agentrun_queue_tasks", "agentrun_queue_attempts", "agentrun_queue_read_cursors", "agentrun_cancel_requests"], + queueRetryConcurrency: { + attemptIdentity: ["task_id", "attempt_id"], + retryIdentity: ["task_id", "idempotency_key"], + activeAttemptFence: ["task_id", "state IN (reserved,running)"], + retryIndexIdentity: ["task_id", "retry_index"], + taskReservationLock: "SELECT ... FOR UPDATE", + legacyBackfill: "agentrun_queue_tasks.latest_attempt", + failureVisibility: ["failure_kind", "failure_message"], + valuesPrinted: false, + }, checksums: Object.fromEntries(postgresMigrations.map((migration) => [migration.id, migration.checksum])), }; } @@ -465,7 +533,7 @@ export class PostgresAgentRunStore implements AgentRunStore { constructor(options: PostgresStoreOptions) { this.pool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-v01", ...(options.poolMax === undefined ? {} : { max: options.poolMax }) }); - this.runnerCreateFencePool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-runner-capacity-fence", max: 1 }); + this.runnerCreateFencePool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-runner-capacity-fence", max: Math.max(2, options.poolMax ?? 10) }); this.eventOutboxConfig = options.eventOutbox ?? { enabled: false, topic: null, source: null }; } @@ -505,11 +573,21 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( } } - async createRun(input: CreateRunInput): Promise { + async createRun(input: CreateRunInput, identity?: CreateRunIdentity): Promise { const at = nowIso(); return this.withTransaction(async (client) => { + if (identity) { + if (!identity.id.trim()) throw new AgentRunError("schema-invalid", "reserved run identity must be non-empty", { httpStatus: 400 }); + await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-create-run", identity.id]); + const existing = await client.query("SELECT * FROM agentrun_runs WHERE id = $1", [identity.id]); + if (existing.rows[0]) { + const run = runFromRow(existing.rows[0]); + assertRunCreateReplay(run, input); + return run; + } + } const sessionRef = await this.resolveSessionForRun(client, input, at); - const run: RunRecord = { ...input, sessionRef, resourceBundleRef: input.resourceBundleRef ?? null, id: newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null }; + const run: RunRecord = { ...input, sessionRef, resourceBundleRef: input.resourceBundleRef ?? null, id: identity?.id ?? newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null }; await client.query( `INSERT INTO agentrun_runs (id, tenant_id, project_id, workspace_ref, session_ref, resource_bundle_ref, provider_id, backend_profile, execution_policy, trace_sink, status, terminal_status, failure_kind, failure_message, created_at, updated_at, claimed_by, lease_expires_at) VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6::jsonb, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14, $15, $16, $17, $18)`, @@ -554,7 +632,6 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( const payloadHash = stableHash(input.payload); return this.withTransaction(async (client) => { const run = await this.requireRunForUpdate(client, runId); - if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 }); if (input.idempotencyKey) { const existing = await client.query("SELECT * FROM agentrun_commands WHERE run_id = $1 AND idempotency_key = $2", [runId, input.idempotencyKey]); if (existing.rows[0]) { @@ -566,6 +643,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( return attachRunnerDispatchIntent(command, intent); } } + if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 }); const runCreatedResult = await client.query<{ exists: boolean }>( `SELECT EXISTS ( SELECT 1 FROM agentrun_events @@ -893,11 +971,11 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( }); } - async withRunnerCreateFence(namespace: string, operation: () => Promise): Promise { + async withRunnerCreateFence(fenceKey: string, operation: () => Promise): Promise { const client = await this.runnerCreateFencePool.connect(); try { await client.query("BEGIN"); - await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-runner-create", namespace]); + await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-runner-create", fenceKey]); const result = await operation(); await client.query("COMMIT"); return result; @@ -1374,6 +1452,129 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( return queueTaskFromRow(row); } + async listQueueAttempts(taskId: string, input: ListQueueAttemptsInput): Promise { + await this.getQueueTask(taskId); + const limit = clampQueueLimit(input.limit); + const params: unknown[] = [taskId]; + let cursorClause = ""; + if (input.cursor !== undefined) { + params.push(input.cursor); + cursorClause = ` AND retry_index < $${params.length}`; + } + params.push(limit); + const result = await this.pool.query( + `SELECT * FROM agentrun_queue_attempts + WHERE task_id = $1${cursorClause} + ORDER BY retry_index DESC, created_at DESC + LIMIT $${params.length}`, + params, + ); + const items = result.rows.map(queueAttemptFromRow); + return { taskId, items, count: items.length, cursor: items.length === limit ? String(items.at(-1)?.retryIndex ?? "") : null }; + } + + async reserveQueueRetryAttempt(taskId: string, input: ReserveQueueRetryAttemptInput): Promise { + return this.withTransaction(async (client) => { + const taskResult = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]); + if (!taskResult.rows[0]) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 }); + const task = queueTaskFromRow(taskResult.rows[0]); + assertQueueTaskPayloadHash(task); + const attemptsResult = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 ORDER BY retry_index ASC, created_at ASC FOR UPDATE", [taskId]); + const attempts = attemptsResult.rows.map(queueAttemptFromRow); + const existing = attempts.find((attempt) => attempt.idempotencyKey === input.idempotencyKey) ?? null; + if (existing) { + if (existing.reason !== input.reason) throw new AgentRunError("schema-invalid", "queue retry idempotency key reused with different reason", { httpStatus: 409 }); + return queueRetryReservation(task, existing, attempts, false, true); + } + assertQueueRetryable(task, attempts); + const retryIndex = nextQueueRetryIndex(attempts); + if (input.dryRun) return queueRetryReservation(task, null, attempts, false, false, retryIndex); + const at = nowIso(); + const attempt: QueueAttemptRecord = { + attemptId: newId("qa"), + taskId, + retryIndex, + state: "reserved", + idempotencyKey: input.idempotencyKey, + reason: input.reason, + payloadHash: task.payloadHash, + previousAttemptId: latestQueueAttempt(attempts)?.attemptId ?? task.latestAttempt?.attemptId ?? null, + runId: newId("run"), + commandId: null, + runnerJobId: newId("rjob"), + sessionId: task.sessionRef?.sessionId ?? null, + sessionPath: task.sessionPath, + failureKind: null, + failureMessage: null, + createdAt: at, + updatedAt: at, + activatedAt: null, + terminalAt: null, + }; + await client.query( + `INSERT INTO agentrun_queue_attempts ( + attempt_id, task_id, retry_index, state, idempotency_key, reason, payload_hash, + previous_attempt_id, run_id, command_id, runner_job_id, session_id, session_path, + failure_kind, failure_message, created_at, updated_at, activated_at, terminal_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19)`, + [attempt.attemptId, attempt.taskId, attempt.retryIndex, attempt.state, attempt.idempotencyKey, attempt.reason, attempt.payloadHash, attempt.previousAttemptId, attempt.runId, attempt.commandId, attempt.runnerJobId, attempt.sessionId, attempt.sessionPath, attempt.failureKind, attempt.failureMessage, attempt.createdAt, attempt.updatedAt, attempt.activatedAt, attempt.terminalAt], + ); + return queueRetryReservation(task, attempt, attempts, true, false); + }); + } + + async recordQueueRetryAttemptFailure(taskId: string, attemptId: string, input: RecordQueueRetryAttemptFailureInput): Promise { + return this.withTransaction(async (client) => { + const attemptResult = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 AND attempt_id = $2 FOR UPDATE", [taskId, attemptId]); + if (!attemptResult.rows[0]) throw new AgentRunError("schema-invalid", `queue attempt ${attemptId} was not found for task ${taskId}`, { httpStatus: 404 }); + const attempt = queueAttemptFromRow(attemptResult.rows[0]); + if (attempt.state !== "reserved") return attempt; + const updated = await client.query( + `UPDATE agentrun_queue_attempts + SET failure_kind = $3, failure_message = $4, updated_at = $5 + WHERE task_id = $1 AND attempt_id = $2 + RETURNING *`, + [taskId, attemptId, input.failureKind, input.failureMessage, nowIso()], + ); + return queueAttemptFromRow(updated.rows[0]); + }); + } + + async activateQueueRetryAttempt(taskId: string, attemptId: string, input: ActivateQueueRetryAttemptInput): Promise { + return this.withTransaction(async (client) => { + const taskResult = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]); + if (!taskResult.rows[0]) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 }); + const task = queueTaskFromRow(taskResult.rows[0]); + const attemptResult = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 AND attempt_id = $2 FOR UPDATE", [taskId, attemptId]); + if (!attemptResult.rows[0]) throw new AgentRunError("schema-invalid", `queue attempt ${attemptId} was not found for task ${taskId}`, { httpStatus: 404 }); + const attempt = queueAttemptFromRow(attemptResult.rows[0]); + if (attempt.state !== "reserved") { + assertQueueAttemptActivationReplay(attempt, input); + return { task, attempt, activated: false }; + } + if (task.state !== "failed" && task.state !== "blocked") throw queueRetryStateError(task); + const at = nowIso(); + const terminalAt = isTerminalQueueTaskState(input.state) ? at : null; + const updatedAttempt = await client.query( + `UPDATE agentrun_queue_attempts + SET state = $3, command_id = $4, session_id = $5, session_path = $6, + failure_kind = $7, failure_message = $8, updated_at = $9, activated_at = $9, terminal_at = $10 + WHERE task_id = $1 AND attempt_id = $2 + RETURNING *`, + [taskId, attemptId, input.state, input.commandId, input.sessionId, input.sessionPath, input.failureKind, input.failureMessage, at, terminalAt], + ); + const nextAttempt = queueAttemptFromRow(updatedAttempt.rows[0]); + const version = await this.nextQueueVersion(client); + const updatedTask = await client.query( + `UPDATE agentrun_queue_tasks + SET state = $2, latest_attempt = $3::jsonb, session_path = $4, version = $5, updated_at = $6 + WHERE id = $1 RETURNING *`, + [taskId, input.state, JSON.stringify(queueAttemptRef(nextAttempt)), input.sessionPath, version, at], + ); + return { task: queueTaskFromRow(updatedTask.rows[0]), attempt: nextAttempt, activated: true }; + }); + } + async updateQueueTaskAttempt(taskId: string, input: UpdateQueueTaskAttemptInput): Promise { return this.withTransaction(async (client) => { const existing = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]); @@ -1383,6 +1584,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( if (isTerminalQueueTaskState(task.state)) throw new AgentRunError(task.state === "cancelled" ? "cancelled" : "schema-invalid", `queue task ${taskId} is already terminal: ${task.state}`, { httpStatus: 409 }); const version = await this.nextQueueVersion(client); const at = nowIso(); + await this.upsertQueueAttemptProjection(client, task, input.latestAttempt, at); const updated = await client.query( "UPDATE agentrun_queue_tasks SET state = $2, latest_attempt = $3::jsonb, session_path = $4, version = $5, updated_at = $6 WHERE id = $1 RETURNING *", [taskId, input.state, JSON.stringify(input.latestAttempt), input.sessionPath, version, at], @@ -1404,7 +1606,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( if (isTerminalQueueTaskState(lockedTask.state)) return lockedTask; const at = nowIso(); const version = await this.nextQueueVersion(client); - const latestAttempt = lockedTask.latestAttempt ? { ...lockedTask.latestAttempt, state: "cancelled" as const } : null; + const latestAttempt = lockedTask.latestAttempt ? { ...lockedTask.latestAttempt, state: "cancelled" as const, failureKind: "cancelled" as const, failureMessage: reason } : null; + if (latestAttempt) await this.upsertQueueAttemptProjection(client, lockedTask, latestAttempt, at); const updated = await client.query("UPDATE agentrun_queue_tasks SET state = 'cancelled', latest_attempt = $2::jsonb, version = $3, updated_at = $4, cancelled_at = $4, cancel_reason = $5 WHERE id = $1 RETURNING *", [taskId, JSON.stringify(latestAttempt), version, at, reason]); return queueTaskFromRow(updated.rows[0]); }); @@ -1455,6 +1658,38 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( await Promise.all([this.pool.end(), this.runnerCreateFencePool.end()]); } + private async upsertQueueAttemptProjection(client: PoolClient, task: QueueTaskRecord, ref: QueueAttemptRef, at: string): Promise { + const existingResult = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 AND attempt_id = $2 FOR UPDATE", [task.id, ref.attemptId]); + if (existingResult.rows[0]) { + const existing = queueAttemptFromRow(existingResult.rows[0]); + if (isTerminalQueueAttemptState(existing.state)) { + if (stableHash(queueAttemptRef(existing)) !== stableHash(ref)) throw new AgentRunError("schema-invalid", `queue attempt ${ref.attemptId} is terminal and immutable`, { httpStatus: 409 }); + return; + } + await client.query( + `UPDATE agentrun_queue_attempts + SET state = $3, run_id = $4, command_id = $5, runner_job_id = $6, + session_id = $7, session_path = $8, failure_kind = $9, failure_message = $10, updated_at = $11, + activated_at = COALESCE(activated_at, $11), + terminal_at = CASE WHEN $3 IN ('completed', 'failed', 'blocked', 'cancelled') THEN $11 ELSE NULL END + WHERE task_id = $1 AND attempt_id = $2`, + [task.id, ref.attemptId, ref.state, ref.runId, ref.commandId, ref.runnerJobId, ref.sessionId, ref.sessionPath, ref.failureKind ?? null, ref.failureMessage ?? null, at], + ); + return; + } + const history = await client.query("SELECT * FROM agentrun_queue_attempts WHERE task_id = $1 ORDER BY retry_index ASC FOR UPDATE", [task.id]); + const attempts = history.rows.map(queueAttemptFromRow); + const retryIndex = nextQueueRetryIndex(attempts); + await client.query( + `INSERT INTO agentrun_queue_attempts ( + attempt_id, task_id, retry_index, state, idempotency_key, reason, payload_hash, + previous_attempt_id, run_id, command_id, runner_job_id, session_id, session_path, + failure_kind, failure_message, created_at, updated_at, activated_at, terminal_at + ) VALUES ($1, $2, $3, $4, NULL, NULL, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $14, $14, $15)`, + [ref.attemptId, task.id, retryIndex, ref.state, task.payloadHash, latestQueueAttempt(attempts)?.attemptId ?? null, ref.runId, ref.commandId, ref.runnerJobId, ref.sessionId, ref.sessionPath, ref.failureKind ?? null, ref.failureMessage ?? null, at, isTerminalQueueTaskState(ref.state) ? at : null], + ); + } + private async createCancelRequest(client: PoolClient, input: { targetKind: CancelTargetKind; targetId: string; run: RunRecord; command: CommandRecord | null; reason: string; at: string; stage: CancelStage }): Promise { const epoch = input.command ? input.command.cancelEpoch + 1 : input.run.cancelEpoch + 1; const record: CancelRequestRecord = { @@ -1928,6 +2163,30 @@ function queueTaskFromRow(row: QueryResultRow): QueueTaskRecord { }; } +function queueAttemptFromRow(row: QueryResultRow): QueueAttemptRecord { + return { + attemptId: stringValue(row.attempt_id), + taskId: stringValue(row.task_id), + retryIndex: Number(row.retry_index), + state: stringValue(row.state) as QueueAttemptRecord["state"], + idempotencyKey: nullableString(row.idempotency_key), + reason: nullableString(row.reason), + payloadHash: stringValue(row.payload_hash), + previousAttemptId: nullableString(row.previous_attempt_id), + runId: nullableString(row.run_id), + commandId: nullableString(row.command_id), + runnerJobId: nullableString(row.runner_job_id), + sessionId: nullableString(row.session_id), + sessionPath: nullableString(row.session_path), + failureKind: nullableString(row.failure_kind) as FailureKind | null, + failureMessage: nullableString(row.failure_message), + createdAt: iso(row.created_at), + updatedAt: iso(row.updated_at), + activatedAt: nullableIso(row.activated_at), + terminalAt: nullableIso(row.terminal_at), + }; +} + function metadataForRunner(runner: RunnerRecord): JsonRecord { const { id: _id, runId: _runId, attemptId: _attemptId, backendProfile: _backendProfile, placement: _placement, sourceCommit: _sourceCommit, registeredAt: _registeredAt, heartbeatAt: _heartbeatAt, ...metadata } = runner; return redactJson(metadata); diff --git a/src/mgr/queue-dispatch.ts b/src/mgr/queue-dispatch.ts index baefd1b..0bab7ee 100644 --- a/src/mgr/queue-dispatch.ts +++ b/src/mgr/queue-dispatch.ts @@ -16,12 +16,12 @@ export async function dispatchQueueTask(options: DispatchQueueTaskOptions): Prom const task = await options.store.getQueueTask(options.taskId); assertDispatchable(task); - const run = await options.store.createRun(buildRunInput(task, options.input)); + const run = await options.store.createRun(buildQueueRunInput(task, options.input)); const command = await options.store.createCommand(run.id, { type: "turn", payload: task.payload, idempotencyKey: `queue:${task.id}:command` }); const runnerJob = await createKubernetesRunnerJob({ store: options.store, runId: run.id, - input: buildRunnerJobInput(task, command.id, options.input), + input: buildQueueRunnerJobInput(task, command.id, options.input), defaults: options.defaults, }); @@ -85,7 +85,7 @@ export async function refreshQueueTaskFromCore(store: AgentRunStore, taskId: str const state = queueStateFromCore(run, command); const sessionId = run.sessionRef?.sessionId ?? latestAttempt.sessionId ?? null; const sessionPath = sessionId ? `/api/v1/sessions/${encodeURIComponent(sessionId)}` : latestAttempt.sessionPath ?? null; - const nextAttempt: QueueAttemptRef = { ...latestAttempt, state, sessionId, sessionPath }; + const nextAttempt: QueueAttemptRef = { ...latestAttempt, state, sessionId, sessionPath, failureKind: run.failureKind, failureMessage: run.failureMessage }; return store.updateQueueTaskAttempt(task.id, { state, latestAttempt: nextAttempt, sessionPath }); } @@ -97,7 +97,7 @@ function assertDispatchable(task: QueueTaskRecord): void { if (!task.executionPolicy) throw new AgentRunError("schema-invalid", "queue dispatch requires executionPolicy", { httpStatus: 400 }); } -function queueStateFromCore(run: RunRecord, command: CommandRecord): QueueTaskState { +export function queueStateFromCore(run: RunRecord, command: CommandRecord): QueueTaskState { const runState = queueTerminalStateFromRun(run); if (runState) return runState; if (command.state === "completed") return "completed"; @@ -114,7 +114,7 @@ function queueTerminalStateFromRun(run: RunRecord): QueueTaskState | null { return null; } -function buildRunInput(task: QueueTaskRecord, input: JsonRecord): CreateRunInput { +export function buildQueueRunInput(task: QueueTaskRecord, input: JsonRecord): CreateRunInput { if (!task.workspaceRef || !task.providerId || !task.executionPolicy) throw new AgentRunError("schema-invalid", "queue dispatch task is missing runtime fields", { httpStatus: 400 }); const traceSink = input.traceSink === undefined ? { kind: "queue", taskId: task.id, queue: task.queue, lane: task.lane } : input.traceSink; return { @@ -130,7 +130,7 @@ function buildRunInput(task: QueueTaskRecord, input: JsonRecord): CreateRunInput }; } -function buildRunnerJobInput(task: QueueTaskRecord, commandId: string, input: JsonRecord): CreateRunnerJobInput { +export function buildQueueRunnerJobInput(task: QueueTaskRecord, commandId: string, input: JsonRecord): CreateRunnerJobInput { const jobInput: CreateRunnerJobInput = { commandId, idempotencyKey: optionalString(input.idempotencyKey) ?? `queue:${task.id}:runner-job` }; const copyString = (inputKey: string, outputKey: keyof CreateRunnerJobInput): void => { const value = optionalString(input[inputKey]); @@ -140,6 +140,7 @@ function buildRunnerJobInput(task: QueueTaskRecord, commandId: string, input: Js copyString("image", "image"); copyString("namespace", "namespace"); copyString("attemptId", "attemptId"); + copyString("runnerJobId", "runnerJobId"); copyString("runnerId", "runnerId"); copyString("sourceCommit", "sourceCommit"); copyString("serviceAccountName", "serviceAccountName"); diff --git a/src/mgr/queue-retry.ts b/src/mgr/queue-retry.ts new file mode 100644 index 0000000..394ff81 --- /dev/null +++ b/src/mgr/queue-retry.ts @@ -0,0 +1,172 @@ +import { AgentRunError } from "../common/errors.js"; +import { redactText } from "../common/redaction.js"; +import type { CommandRecord, FailureKind, JsonRecord, JsonValue, QueueAttemptRecord, QueueRetryResult, RunRecord } from "../common/types.js"; +import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js"; +import { buildQueueRunInput, buildQueueRunnerJobInput, queueStateFromCore } from "./queue-dispatch.js"; +import type { AgentRunStore } from "./store.js"; + +interface QueueRetryRequest { + idempotencyKey: string; + reason: string; + dryRun: boolean; +} + +export interface RetryQueueTaskOptions { + store: AgentRunStore; + taskId: string; + input: JsonRecord; + defaults: RunnerJobDefaults; +} + +export async function retryQueueTask(options: RetryQueueTaskOptions): Promise { + const request = validateQueueRetryRequest(options.input); + const reservation = await options.store.reserveQueueRetryAttempt(options.taskId, request); + if (request.dryRun) return queueRetryResult({ reservation, mutation: false }); + const attempt = reservation.attempt; + if (!attempt) throw new AgentRunError("infra-failed", "queue retry reservation did not return an attempt", { httpStatus: 500 }); + if (attempt.state !== "reserved") return queueRetryReplayResult(options.store, reservation); + let run: RunRecord; + let command: CommandRecord; + let runnerJob: JsonRecord; + try { + if (!attempt.runId || !attempt.runnerJobId) throw new AgentRunError("infra-failed", `queue retry attempt ${attempt.attemptId} is missing reserved execution identities`, { httpStatus: 500 }); + run = await options.store.createRun(buildQueueRunInput(reservation.task, {}), { id: attempt.runId }); + command = await options.store.createCommand(run.id, { + type: "turn", + payload: reservation.task.payload, + idempotencyKey: `queue:${reservation.task.id}:attempt:${attempt.attemptId}:command`, + }); + runnerJob = await createKubernetesRunnerJob({ + store: options.store, + runId: run.id, + input: buildQueueRunnerJobInput(reservation.task, command.id, { + attemptId: attempt.attemptId, + runnerJobId: attempt.runnerJobId, + idempotencyKey: `queue:${reservation.task.id}:attempt:${attempt.attemptId}:runner-job`, + }), + defaults: options.defaults, + }); + } catch (error) { + await options.store.recordQueueRetryAttemptFailure(reservation.task.id, attempt.attemptId, materializationFailure(error)); + throw error; + } + const state = queueStateFromCore(run, command); + const sessionId = run.sessionRef?.sessionId ?? attempt.sessionId; + const sessionPath = sessionId ? `/api/v1/sessions/${encodeURIComponent(sessionId)}` : attempt.sessionPath; + const activation = await options.store.activateQueueRetryAttempt(reservation.task.id, attempt.attemptId, { + commandId: command.id, + sessionId, + sessionPath, + state: state === "pending" ? "running" : state, + failureKind: run.failureKind, + failureMessage: run.failureMessage, + }); + if (activation.activated) { + await options.store.appendEvent(run.id, "backend_status", { + phase: "queue-retried", + taskId: reservation.task.id, + queue: reservation.task.queue, + lane: reservation.task.lane, + attemptId: attempt.attemptId, + retryIndex: attempt.retryIndex, + previousAttemptId: attempt.previousAttemptId, + runnerJobId: attempt.runnerJobId, + sessionPath, + reason: attempt.reason, + }); + } + return queueRetryResult({ + reservation, + mutation: activation.activated || runnerJob.mutation === true, + task: activation.task, + attempt: activation.attempt, + run, + command, + runnerJob, + }); +} + +export function validateQueueRetryRequest(input: JsonRecord): QueueRetryRequest { + const allowed = new Set(["idempotencyKey", "reason", "dryRun"]); + const unexpected = Object.keys(input).filter((key) => !allowed.has(key)); + if (unexpected.length > 0) throw new AgentRunError("schema-invalid", `queue retry does not accept task or dispatch patches: ${unexpected.join(", ")}`, { httpStatus: 400, details: { unexpected, allowed: [...allowed], valuesPrinted: false } }); + const idempotencyKey = requiredBoundedString(input.idempotencyKey, "idempotencyKey", 200); + const reason = requiredBoundedString(input.reason, "reason", 1_000); + if (input.dryRun !== undefined && typeof input.dryRun !== "boolean") throw new AgentRunError("schema-invalid", "dryRun must be a boolean", { httpStatus: 400 }); + return { idempotencyKey, reason, dryRun: input.dryRun === true }; +} + +async function queueRetryReplayResult(store: AgentRunStore, reservation: Awaited>): Promise { + const attempt = reservation.attempt; + if (!attempt) return queueRetryResult({ reservation, mutation: false }); + const run = attempt.runId ? await store.getRun(attempt.runId) : null; + const command = attempt.commandId ? await store.getCommand(attempt.commandId) : null; + const runnerJob = run && command + ? (await store.listRunnerJobs(run.id, command.id)).find((item) => item.id === attempt.runnerJobId || item.attemptId === attempt.attemptId)?.result ?? null + : null; + return queueRetryResult({ reservation, mutation: false, run, command, runnerJob }); +} + +function queueRetryResult(input: { + reservation: Awaited>; + mutation: boolean; + task?: QueueRetryResult["task"]; + attempt?: QueueAttemptRecord | null; + run?: RunRecord | null; + command?: CommandRecord | null; + runnerJob?: JsonRecord | null; +}): QueueRetryResult { + const task = input.task ?? input.reservation.task; + const attempt = input.attempt === undefined ? input.reservation.attempt : input.attempt; + const run = input.run ?? null; + const command = input.command ?? null; + const runnerJob = input.runnerJob ?? null; + return { + action: "queue-retry", + mutation: input.mutation, + idempotentReplay: input.reservation.idempotentReplay, + task, + attempt, + previousAttempt: input.reservation.previousAttempt, + run, + command, + runnerJob, + envImage: jsonRecordOrNull(runnerJob?.envImage), + workReady: jsonRecordOrNull(runnerJob?.workReady), + allowedTransition: input.reservation.allowedTransition, + pollActions: attempt ? retryPollActions(task.id, attempt, run, command) : [], + }; +} + +function retryPollActions(taskId: string, attempt: QueueAttemptRecord, run: RunRecord | null, command: CommandRecord | null): JsonRecord[] { + return [ + actionDescriptor("inspect-task", "describe", "task", taskId), + actionDescriptor("list-attempts", "get", "attempt", attempt.attemptId), + ...(run ? [actionDescriptor("inspect-run", "describe", "run", run.id, run.id)] : []), + ...(run && command ? [ + actionDescriptor("inspect-command", "describe", "command", command.id, run.id, command.id), + { ...actionDescriptor("poll-events", "events", "run", run.id, run.id, command.id), afterSeq: 0, limit: 100 }, + ] : []), + ]; +} + +function actionDescriptor(action: string, operation: string, resourceKind: string, resourceName: string, runId: string | null = null, commandId: string | null = null): JsonRecord { + return { action, operation, resourceKind, resourceName, runId, commandId, valuesPrinted: false }; +} + +function requiredBoundedString(value: JsonValue | undefined, field: string, maxLength: number): string { + if (typeof value !== "string" || value.trim().length === 0) throw new AgentRunError("schema-invalid", `${field} is required`, { httpStatus: 400 }); + const normalized = value.trim(); + if (normalized.length > maxLength) throw new AgentRunError("schema-invalid", `${field} must be at most ${maxLength} characters`, { httpStatus: 400 }); + return normalized; +} + +function materializationFailure(error: unknown): { failureKind: FailureKind; failureMessage: string } { + const failureKind = error instanceof AgentRunError ? error.failureKind : "infra-failed"; + const rawMessage = error instanceof Error ? error.message : String(error); + return { failureKind, failureMessage: redactText(rawMessage).trim().slice(0, 500) || "queue retry materialization failed" }; +} + +function jsonRecordOrNull(value: JsonValue | undefined): JsonRecord | null { + return value && typeof value === "object" && !Array.isArray(value) ? value as JsonRecord : null; +} diff --git a/src/mgr/server.ts b/src/mgr/server.ts index a478d46..58b1a64 100644 --- a/src/mgr/server.ts +++ b/src/mgr/server.ts @@ -11,6 +11,7 @@ import type { ApiErrorBody, ApiOkBody, CommandRecord, JsonRecord, JsonValue, Que import { createKubernetesRunnerJob, type RunnerJobDefaults } from "./kubernetes-runner-job.js"; import type { RunnerRetentionOptions } from "./runner-retention.js"; import { dispatchQueueTask, refreshQueueTaskFromCore } from "./queue-dispatch.js"; +import { retryQueueTask } from "./queue-retry.js"; import { buildRunResult } from "./result.js"; import { runnerJobStatusSummary } from "./runner-job-status.js"; import { reconcileRunnerJobsOnce, startRunnerJobReconciler } from "./runner-reconciler.js"; @@ -859,6 +860,24 @@ async function route({ method, url, body, store, sourceCommit, authSummary, diag } const queueTaskMatch = path.match(/^\/api\/v1\/queue\/tasks\/([^/]+)$/u); if (method === "GET" && queueTaskMatch) return await refreshQueueTaskForRead(store, queueTaskMatch[1] ?? "") as unknown as JsonValue; + const queueTaskAttemptsMatch = path.match(/^\/api\/v1\/queue\/tasks\/([^/]+)\/attempts$/u); + if (method === "GET" && queueTaskAttemptsMatch) { + const taskId = queueTaskAttemptsMatch[1] ?? ""; + const cursorValue = url.searchParams.get("cursor"); + const cursor = cursorValue === null ? undefined : Number(cursorValue); + if (cursor !== undefined && (!Number.isInteger(cursor) || cursor < 0)) throw new AgentRunError("schema-invalid", "queue attempt cursor must be a non-negative integer", { httpStatus: 400 }); + await refreshQueueTaskRecordForRead(store, await store.getQueueTask(taskId)); + return await store.listQueueAttempts(taskId, { ...(cursor === undefined ? {} : { cursor }), limit: integerQuery(url, "limit", 50) }) as unknown as JsonValue; + } + const queueTaskRetryMatch = path.match(/^\/api\/v1\/queue\/tasks\/([^/]+)\/retry$/u); + if (method === "POST" && queueTaskRetryMatch) { + return await retryQueueTask({ + store, + taskId: queueTaskRetryMatch[1] ?? "", + input: asRecord(body ?? {}, "queueRetry"), + defaults: resolvedRunnerJobDefaults, + }) as unknown as JsonValue; + } const queueTaskDispatchMatch = path.match(/^\/api\/v1\/queue\/tasks\/([^/]+)\/dispatch$/u); if (method === "POST" && queueTaskDispatchMatch) { return await dispatchQueueTask({ diff --git a/src/mgr/store.ts b/src/mgr/store.ts index a594d98..e09649c 100644 --- a/src/mgr/store.ts +++ b/src/mgr/store.ts @@ -1,4 +1,4 @@ -import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, FailureKind, JsonRecord, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, QueueTaskSummary, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, SessionEventPage, SessionListResult, SessionListState, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js"; +import type { BackendProfile, BackendTurnResult, CancelRequestRecord, CancelStage, CancelTargetKind, CommandRecord, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, FailureKind, JsonRecord, KafkaEventOutboxRecord, ListGcExpiredSessionsInput, QueueAttemptListResult, QueueAttemptRecord, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueRetryActivation, QueueRetryReservation, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, QueueTaskSummary, RunEvent, RunnerDispatchCompletion, RunnerDispatchIntentRecord, RunnerJobRecord, RunnerRecord, RunRecord, SessionEventPage, SessionListResult, SessionListState, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js"; import { AgentRunError } from "../common/errors.js"; import { newId, nowIso, stableHash } from "../common/validation.js"; import { redactJson } from "../common/redaction.js"; @@ -48,7 +48,7 @@ export interface RunnerRetentionFenceInput { export interface AgentRunStore { health(): MaybePromise; - createRun(input: CreateRunInput): MaybePromise; + createRun(input: CreateRunInput, identity?: CreateRunIdentity): MaybePromise; getRun(runId: string): MaybePromise; listEvents(runId: string, afterSeq: number, limit: number): MaybePromise; listEventsForCommand(runId: string, commandId: string, limit: number): MaybePromise; @@ -71,7 +71,7 @@ export interface AgentRunStore { getRunnerJobByIdempotencyKey(runId: string, idempotencyKey: string, payloadHash: string): MaybePromise; saveRunnerJob(input: SaveRunnerJobInput): MaybePromise; updateRunnerJobResult(runnerJobId: string, patch: JsonRecord): MaybePromise; - withRunnerCreateFence(namespace: string, operation: () => Promise): MaybePromise; + withRunnerCreateFence(fenceKey: string, operation: () => Promise): MaybePromise; withRunnerRetentionFence(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise): MaybePromise; claimRun(runId: string, runnerId: string, leaseMs: number): MaybePromise; heartbeat(runId: string, runnerId: string, leaseMs: number): MaybePromise; @@ -94,6 +94,10 @@ export interface AgentRunStore { createQueueTask(input: CreateQueueTaskInput): MaybePromise; listQueueTasks(input: ListQueueTasksInput): MaybePromise; getQueueTask(taskId: string): MaybePromise; + listQueueAttempts(taskId: string, input: ListQueueAttemptsInput): MaybePromise; + reserveQueueRetryAttempt(taskId: string, input: ReserveQueueRetryAttemptInput): MaybePromise; + recordQueueRetryAttemptFailure(taskId: string, attemptId: string, input: RecordQueueRetryAttemptFailureInput): MaybePromise; + activateQueueRetryAttempt(taskId: string, attemptId: string, input: ActivateQueueRetryAttemptInput): MaybePromise; updateQueueTaskAttempt(taskId: string, input: UpdateQueueTaskAttemptInput): MaybePromise; cancelQueueTask(taskId: string, reason?: string): MaybePromise; markQueueTaskRead(taskId: string, readerId: string): MaybePromise; @@ -103,6 +107,10 @@ export interface AgentRunStore { close?(): MaybePromise; } +export interface CreateRunIdentity { + id: string; +} + export interface ListQueueTasksInput { queue?: string; state?: QueueTaskState; @@ -111,6 +119,31 @@ export interface ListQueueTasksInput { updatedAfter?: number; } +export interface ListQueueAttemptsInput { + cursor?: number; + limit: number; +} + +export interface ReserveQueueRetryAttemptInput { + idempotencyKey: string; + reason: string; + dryRun: boolean; +} + +export interface ActivateQueueRetryAttemptInput { + commandId: string; + sessionId: string | null; + sessionPath: string | null; + state: Exclude; + failureKind: FailureKind | null; + failureMessage: string | null; +} + +export interface RecordQueueRetryAttemptFailureInput { + failureKind: FailureKind; + failureMessage: string; +} + export interface ListSessionsInput { state?: SessionListState; backendProfile?: BackendProfile; @@ -194,6 +227,7 @@ export class MemoryAgentRunStore implements AgentRunStore { private readonly runnerDispatchIntents = new Map(); private readonly kafkaEventOutbox = new Map(); private readonly queueTasks = new Map(); + private readonly queueAttempts = new Map(); private readonly queueReadCursors = new Map(); private readonly runnerCreateFenceTails = new Map>(); private readonly runnerRetentionFences = new Set(); @@ -209,10 +243,17 @@ export class MemoryAgentRunStore implements AgentRunStore { return { adapter: "memory-self-test", ready: true, reachable: true, migrationReady: true, migrationId: "memory-self-test", failureKind: null, message: null, credentialValuesPrinted: false }; } - createRun(input: CreateRunInput): RunRecord { + createRun(input: CreateRunInput, identity?: CreateRunIdentity): RunRecord { + if (identity) { + const existing = this.runs.get(identity.id); + if (existing) { + assertRunCreateReplay(existing, input); + return existing; + } + } const at = nowIso(); const sessionRef = this.resolveSessionForRun(input, at); - const run: RunRecord = { ...input, sessionRef, resourceBundleRef: input.resourceBundleRef ?? null, id: newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null }; + const run: RunRecord = { ...input, sessionRef, resourceBundleRef: input.resourceBundleRef ?? null, id: identity?.id ?? newId("run"), status: "pending", terminalStatus: null, failureKind: null, failureMessage: null, cancelEpoch: 0, cancelRequestId: null, cancelRequestedAt: null, cancelReason: null, createdAt: at, updatedAt: at, claimedBy: null, leaseExpiresAt: null }; this.runs.set(run.id, run); this.eventsByRun.set(run.id, []); this.touchSessionForRun(run, { lastRunId: run.id, lastActivityAt: at }, { bumpVersion: false, at }); @@ -240,7 +281,6 @@ export class MemoryAgentRunStore implements AgentRunStore { createCommand(runId: string, input: CreateCommandInput): CommandRecord { const run = this.getRun(runId); - if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 }); const payloadHash = stableHash(input.payload); if (input.idempotencyKey) { const existing = Array.from(this.commands.values()).find((command) => command.runId === runId && command.idempotencyKey === input.idempotencyKey); @@ -251,6 +291,7 @@ export class MemoryAgentRunStore implements AgentRunStore { return attachRunnerDispatchIntent(existing, intent); } } + if (isTerminalRunStatus(run.status)) throw new AgentRunError(run.failureKind ?? (run.status === "cancelled" ? "cancelled" : "schema-invalid"), `run ${runId} is already terminal: ${run.status}`, { httpStatus: 409 }); const at = nowIso(); const seq = Array.from(this.commands.values()).filter((command) => command.runId === runId).length + 1; const { dispatch: _dispatch, ...commandInput } = input; @@ -448,18 +489,18 @@ export class MemoryAgentRunStore implements AgentRunStore { return next; } - async withRunnerCreateFence(namespace: string, operation: () => Promise): Promise { - const predecessor = this.runnerCreateFenceTails.get(namespace) ?? Promise.resolve(); + async withRunnerCreateFence(fenceKey: string, operation: () => Promise): Promise { + const predecessor = this.runnerCreateFenceTails.get(fenceKey) ?? Promise.resolve(); let release = (): void => {}; const current = new Promise((resolve) => { release = resolve; }); const tail = predecessor.then(async () => current); - this.runnerCreateFenceTails.set(namespace, tail); + this.runnerCreateFenceTails.set(fenceKey, tail); await predecessor; try { return await operation(); } finally { release(); - if (this.runnerCreateFenceTails.get(namespace) === tail) this.runnerCreateFenceTails.delete(namespace); + if (this.runnerCreateFenceTails.get(fenceKey) === tail) this.runnerCreateFenceTails.delete(fenceKey); } } @@ -812,10 +853,99 @@ export class MemoryAgentRunStore implements AgentRunStore { return task; } + listQueueAttempts(taskId: string, input: ListQueueAttemptsInput): QueueAttemptListResult { + this.getQueueTask(taskId); + const cursor = input.cursor ?? Number.POSITIVE_INFINITY; + const items = Array.from(this.queueAttempts.values()) + .filter((attempt) => attempt.taskId === taskId && attempt.retryIndex < cursor) + .sort((a, b) => b.retryIndex - a.retryIndex || b.createdAt.localeCompare(a.createdAt)) + .slice(0, clampQueueLimit(input.limit)); + return { taskId, items, count: items.length, cursor: items.length === clampQueueLimit(input.limit) ? String(items.at(-1)?.retryIndex ?? "") : null }; + } + + reserveQueueRetryAttempt(taskId: string, input: ReserveQueueRetryAttemptInput): QueueRetryReservation { + const task = this.getQueueTask(taskId); + assertQueueTaskPayloadHash(task); + const attempts = this.queueAttemptsForTask(taskId); + const existing = attempts.find((attempt) => attempt.idempotencyKey === input.idempotencyKey) ?? null; + if (existing) { + if (existing.reason !== input.reason) throw new AgentRunError("schema-invalid", "queue retry idempotency key reused with different reason", { httpStatus: 409 }); + return queueRetryReservation(task, existing, attempts, false, true); + } + assertQueueRetryable(task, attempts); + const retryIndex = nextQueueRetryIndex(attempts); + if (input.dryRun) return queueRetryReservation(task, null, attempts, false, false, retryIndex); + const at = nowIso(); + const attempt: QueueAttemptRecord = { + attemptId: newId("qa"), + taskId, + retryIndex, + state: "reserved", + idempotencyKey: input.idempotencyKey, + reason: input.reason, + payloadHash: task.payloadHash, + previousAttemptId: latestQueueAttempt(attempts)?.attemptId ?? task.latestAttempt?.attemptId ?? null, + runId: newId("run"), + commandId: null, + runnerJobId: newId("rjob"), + sessionId: task.sessionRef?.sessionId ?? null, + sessionPath: task.sessionPath, + failureKind: null, + failureMessage: null, + createdAt: at, + updatedAt: at, + activatedAt: null, + terminalAt: null, + }; + this.queueAttempts.set(queueAttemptKey(taskId, attempt.attemptId), attempt); + return queueRetryReservation(task, attempt, attempts, true, false); + } + + recordQueueRetryAttemptFailure(taskId: string, attemptId: string, input: RecordQueueRetryAttemptFailureInput): QueueAttemptRecord { + this.getQueueTask(taskId); + const attemptKey = queueAttemptKey(taskId, attemptId); + const attempt = this.queueAttempts.get(attemptKey); + if (!attempt) throw new AgentRunError("schema-invalid", `queue attempt ${attemptId} was not found for task ${taskId}`, { httpStatus: 404 }); + if (attempt.state !== "reserved") return attempt; + const next: QueueAttemptRecord = { ...attempt, failureKind: input.failureKind, failureMessage: input.failureMessage, updatedAt: nowIso() }; + this.queueAttempts.set(attemptKey, next); + return next; + } + + activateQueueRetryAttempt(taskId: string, attemptId: string, input: ActivateQueueRetryAttemptInput): QueueRetryActivation { + const task = this.getQueueTask(taskId); + const attempt = this.queueAttempts.get(queueAttemptKey(taskId, attemptId)); + if (!attempt) throw new AgentRunError("schema-invalid", `queue attempt ${attemptId} was not found for task ${taskId}`, { httpStatus: 404 }); + if (attempt.state !== "reserved") { + assertQueueAttemptActivationReplay(attempt, input); + return { task, attempt, activated: false }; + } + if (task.state !== "failed" && task.state !== "blocked") throw queueRetryStateError(task); + const at = nowIso(); + const nextAttempt: QueueAttemptRecord = { + ...attempt, + state: input.state, + commandId: input.commandId, + sessionId: input.sessionId, + sessionPath: input.sessionPath, + failureKind: input.failureKind, + failureMessage: input.failureMessage, + updatedAt: at, + activatedAt: at, + terminalAt: isTerminalQueueTaskState(input.state) ? at : null, + }; + const latestAttempt = queueAttemptRef(nextAttempt); + const nextTask: QueueTaskRecord = { ...task, state: input.state, latestAttempt, sessionPath: input.sessionPath, version: this.nextQueueVersion(), updatedAt: at }; + this.queueAttempts.set(queueAttemptKey(taskId, attemptId), nextAttempt); + this.queueTasks.set(taskId, nextTask); + return { task: nextTask, attempt: nextAttempt, activated: true }; + } + updateQueueTaskAttempt(taskId: string, input: UpdateQueueTaskAttemptInput): QueueTaskRecord { const task = this.getQueueTask(taskId); if (isTerminalQueueTaskState(task.state)) throw new AgentRunError(task.state === "cancelled" ? "cancelled" : "schema-invalid", `queue task ${taskId} is already terminal: ${task.state}`, { httpStatus: 409 }); const next: QueueTaskRecord = { ...task, state: input.state, latestAttempt: input.latestAttempt, sessionPath: input.sessionPath, version: this.nextQueueVersion(), updatedAt: nowIso() }; + this.upsertQueueAttemptProjection(task, input.latestAttempt, next.updatedAt); this.queueTasks.set(taskId, next); return next; } @@ -826,8 +956,9 @@ export class MemoryAgentRunStore implements AgentRunStore { if (task.latestAttempt?.runId) this.cancelRun(task.latestAttempt.runId, reason); else if (task.latestAttempt?.commandId) this.cancelCommand(task.latestAttempt.commandId, reason); const at = nowIso(); - const latestAttempt = task.latestAttempt ? { ...task.latestAttempt, state: "cancelled" as const } : null; + const latestAttempt = task.latestAttempt ? { ...task.latestAttempt, state: "cancelled" as const, failureKind: "cancelled" as const, failureMessage: reason } : null; const next: QueueTaskRecord = { ...task, state: "cancelled", latestAttempt, version: this.nextQueueVersion(), updatedAt: at, cancelledAt: at, cancelReason: reason }; + if (latestAttempt) this.upsertQueueAttemptProjection(task, latestAttempt, at); this.queueTasks.set(taskId, next); return next; } @@ -854,6 +985,56 @@ export class MemoryAgentRunStore implements AgentRunStore { return { queue: queue ?? null, readerId, stats: buildQueueStats(tasks, queue ?? null, generatedAt), items, generatedAt }; } + private queueAttemptsForTask(taskId: string): QueueAttemptRecord[] { + return Array.from(this.queueAttempts.values()).filter((attempt) => attempt.taskId === taskId).sort((a, b) => a.retryIndex - b.retryIndex || a.createdAt.localeCompare(b.createdAt)); + } + + private upsertQueueAttemptProjection(task: QueueTaskRecord, ref: QueueAttemptRef, at: string): void { + const attemptKey = queueAttemptKey(task.id, ref.attemptId); + const existing = this.queueAttempts.get(attemptKey); + if (existing) { + if (isTerminalQueueAttemptState(existing.state)) { + if (stableHash(queueAttemptRef(existing)) !== stableHash(ref)) throw new AgentRunError("schema-invalid", `queue attempt ${ref.attemptId} is terminal and immutable`, { httpStatus: 409 }); + return; + } + const next: QueueAttemptRecord = { + ...existing, + ...ref, + state: ref.state, + failureKind: ref.failureKind ?? null, + failureMessage: ref.failureMessage ?? null, + updatedAt: at, + activatedAt: existing.activatedAt ?? at, + terminalAt: isTerminalQueueTaskState(ref.state) ? at : null, + }; + this.queueAttempts.set(attemptKey, next); + return; + } + const attempts = this.queueAttemptsForTask(task.id); + const record: QueueAttemptRecord = { + attemptId: ref.attemptId, + taskId: task.id, + retryIndex: nextQueueRetryIndex(attempts), + state: ref.state, + idempotencyKey: null, + reason: null, + payloadHash: task.payloadHash, + previousAttemptId: latestQueueAttempt(attempts)?.attemptId ?? null, + runId: ref.runId, + commandId: ref.commandId, + runnerJobId: ref.runnerJobId, + sessionId: ref.sessionId, + sessionPath: ref.sessionPath, + failureKind: ref.failureKind ?? null, + failureMessage: ref.failureMessage ?? null, + createdAt: at, + updatedAt: at, + activatedAt: at, + terminalAt: isTerminalQueueTaskState(ref.state) ? at : null, + }; + this.queueAttempts.set(attemptKey, record); + } + backends(): JsonRecord[] { return backendCapabilities(); } @@ -1260,6 +1441,123 @@ export function queueTaskPayloadHash(input: CreateQueueTaskInput): string { }); } +export function assertRunCreateReplay(existing: RunRecord, input: CreateRunInput): void { + const existingContract = { + tenantId: existing.tenantId, + projectId: existing.projectId, + workspaceRef: existing.workspaceRef, + sessionId: existing.sessionRef?.sessionId ?? null, + resourceBundleRef: existing.resourceBundleRef, + providerId: existing.providerId, + backendProfile: existing.backendProfile, + executionPolicy: existing.executionPolicy, + traceSink: existing.traceSink, + }; + const inputContract = { + tenantId: input.tenantId, + projectId: input.projectId, + workspaceRef: input.workspaceRef, + sessionId: input.sessionRef?.sessionId ?? null, + resourceBundleRef: input.resourceBundleRef ?? null, + providerId: input.providerId, + backendProfile: input.backendProfile, + executionPolicy: input.executionPolicy, + traceSink: input.traceSink, + }; + if (stableHash(existingContract) !== stableHash(inputContract)) { + throw new AgentRunError("schema-invalid", `run identity ${existing.id} reused with a different creation contract`, { httpStatus: 409 }); + } +} + +export function assertQueueTaskPayloadHash(task: QueueTaskRecord): void { + if (queueTaskPayloadHash(task) !== task.payloadHash) throw new AgentRunError("schema-invalid", `queue task ${task.id} payloadHash does not match its immutable contract`, { httpStatus: 409 }); +} + +export function assertQueueRetryable(task: QueueTaskRecord, attempts: QueueAttemptRecord[]): void { + if (task.state !== "failed" && task.state !== "blocked") throw queueRetryStateError(task); + if (!task.workspaceRef) throw new AgentRunError("schema-invalid", "queue retry requires the original workspaceRef", { httpStatus: 400 }); + if (!task.providerId) throw new AgentRunError("schema-invalid", "queue retry requires the original providerId", { httpStatus: 400 }); + if (!task.executionPolicy) throw new AgentRunError("schema-invalid", "queue retry requires the original executionPolicy", { httpStatus: 400 }); + const active = attempts.find((attempt) => attempt.state === "reserved" || attempt.state === "running"); + if (active) throw new AgentRunError("schema-invalid", `queue task ${task.id} already has an active retry attempt: ${active.attemptId}`, { httpStatus: 409, details: { attemptId: active.attemptId, state: active.state, valuesPrinted: false } }); +} + +export function queueRetryStateError(task: QueueTaskRecord): AgentRunError { + const failureKind = task.state === "cancelled" ? "cancelled" : "schema-invalid"; + return new AgentRunError(failureKind, `queue task ${task.id} cannot be retried from state ${task.state}`, { httpStatus: 409, details: { allowedStates: ["failed", "blocked"], state: task.state, valuesPrinted: false } }); +} + +export function queueRetryReservation(task: QueueTaskRecord, attempt: QueueAttemptRecord | null, attempts: QueueAttemptRecord[], mutation: boolean, idempotentReplay: boolean, plannedRetryIndex?: number): QueueRetryReservation { + const retryIndex = attempt?.retryIndex ?? plannedRetryIndex ?? nextQueueRetryIndex(attempts); + const previousAttempt = attempt?.previousAttemptId + ? attempts.find((item) => item.attemptId === attempt.previousAttemptId) ?? null + : latestQueueAttempt(attempts); + return { + action: "queue-retry-reservation", + mutation, + idempotentReplay, + task, + attempt, + previousAttempt, + retryIndex, + allowedTransition: { + from: task.state, + to: attempt && attempt.state !== "reserved" ? attempt.state : "running", + retryIndex, + previousAttemptId: attempt?.previousAttemptId ?? previousAttempt?.attemptId ?? task.latestAttempt?.attemptId ?? null, + payloadHash: task.payloadHash, + valuesPrinted: false, + }, + }; +} + +export function latestQueueAttempt(attempts: QueueAttemptRecord[]): QueueAttemptRecord | null { + return attempts.reduce((latest, attempt) => !latest || attempt.retryIndex > latest.retryIndex ? attempt : latest, null); +} + +export function nextQueueRetryIndex(attempts: QueueAttemptRecord[]): number { + return (latestQueueAttempt(attempts)?.retryIndex ?? -1) + 1; +} + +export function queueAttemptRef(attempt: QueueAttemptRecord): QueueAttemptRef { + if (attempt.state === "reserved" || attempt.state === "pending") throw new AgentRunError("schema-invalid", `queue attempt ${attempt.attemptId} is not active`, { httpStatus: 409 }); + return { + attemptId: attempt.attemptId, + state: attempt.state, + runId: attempt.runId, + commandId: attempt.commandId, + runnerJobId: attempt.runnerJobId, + sessionId: attempt.sessionId, + sessionPath: attempt.sessionPath, + failureKind: attempt.failureKind, + failureMessage: attempt.failureMessage, + }; +} + +export function assertQueueAttemptActivationReplay(attempt: QueueAttemptRecord, input: ActivateQueueRetryAttemptInput): void { + const expected = { + state: input.state, + commandId: input.commandId, + sessionId: input.sessionId, + sessionPath: input.sessionPath, + failureKind: input.failureKind, + failureMessage: input.failureMessage, + }; + const actual = { + state: attempt.state, + commandId: attempt.commandId, + sessionId: attempt.sessionId, + sessionPath: attempt.sessionPath, + failureKind: attempt.failureKind, + failureMessage: attempt.failureMessage, + }; + if (stableHash(actual) !== stableHash(expected)) throw new AgentRunError("schema-invalid", `queue attempt ${attempt.attemptId} activation replay does not match its immutable execution identity`, { httpStatus: 409 }); +} + +export function isTerminalQueueAttemptState(state: QueueAttemptRecord["state"]): boolean { + return state === "completed" || state === "failed" || state === "blocked" || state === "cancelled"; +} + export function buildQueueStats(tasks: QueueTaskRecord[], queue: string | null, generatedAt = nowIso()): QueueStats { const byState: Record = {}; const byLane: Record = {}; @@ -1286,6 +1584,10 @@ function queueReadKey(taskId: string, readerId: string): string { return `${taskId}:${readerId}`; } +function queueAttemptKey(taskId: string, attemptId: string): string { + return `${taskId}:${attemptId}`; +} + export function sessionReadKey(sessionId: string, readerId: string): string { return `${sessionId}:${readerId}`; } diff --git a/src/selftest/cases/00-redaction-postgres.ts b/src/selftest/cases/00-redaction-postgres.ts index 15f4fc6..9d1b0b2 100644 --- a/src/selftest/cases/00-redaction-postgres.ts +++ b/src/selftest/cases/00-redaction-postgres.ts @@ -13,19 +13,21 @@ const selfTest: SelfTestCase = async () => { (error) => error instanceof AgentRunError && error.failureKind === "infra-failed" && error.message.includes("DATABASE_URL is required"), ); const postgresContract = postgresMigrationContract(); - assert.equal(postgresContract.latestMigrationId, "013_v02_runner_dispatch_outcome"); + assert.equal(postgresContract.latestMigrationId, "014_v02_queue_attempt_history_retry"); assert.equal((postgresContract.migrationIds as string[]).includes("008_v01_dsflash_go_backend_profile"), true); assert.equal((postgresContract.migrationIds as string[]).includes("009_v01_dsflash_go_model_catalog"), true); assert.equal((postgresContract.migrationIds as string[]).includes("010_v01_queue_session_ref"), true); assert.equal((postgresContract.migrationIds as string[]).includes("011_v01_cancel_lifecycle"), true); assert.equal((postgresContract.migrationIds as string[]).includes("012_v02_durable_dispatch_kafka_outbox"), true); assert.equal((postgresContract.migrationIds as string[]).includes("013_v02_runner_dispatch_outcome"), true); + assert.equal((postgresContract.migrationIds as string[]).includes("014_v02_queue_attempt_history_retry"), true); assert.ok(typeof (postgresContract.checksums as Record)["008_v01_dsflash_go_backend_profile"] === "string" && (postgresContract.checksums as Record)["008_v01_dsflash_go_backend_profile"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["009_v01_dsflash_go_model_catalog"] === "string" && (postgresContract.checksums as Record)["009_v01_dsflash_go_model_catalog"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["010_v01_queue_session_ref"] === "string" && (postgresContract.checksums as Record)["010_v01_queue_session_ref"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["011_v01_cancel_lifecycle"] === "string" && (postgresContract.checksums as Record)["011_v01_cancel_lifecycle"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["012_v02_durable_dispatch_kafka_outbox"] === "string" && (postgresContract.checksums as Record)["012_v02_durable_dispatch_kafka_outbox"].length > 0); assert.ok(typeof (postgresContract.checksums as Record)["013_v02_runner_dispatch_outcome"] === "string" && (postgresContract.checksums as Record)["013_v02_runner_dispatch_outcome"].length > 0); + assert.ok(typeof (postgresContract.checksums as Record)["014_v02_queue_attempt_history_retry"] === "string" && (postgresContract.checksums as Record)["014_v02_queue_attempt_history_retry"].length > 0); assert.equal((postgresContract.checksums as Record)["002_v01_backend_profiles"], "928b5c490cc4539cb64ecef34784557601b2724fa2870570f16a53576804e49c"); assert.ok(Array.isArray(postgresContract.requiredTables)); assert.ok(postgresContract.requiredTables.includes("agentrun_schema_migrations")); @@ -34,6 +36,7 @@ const selfTest: SelfTestCase = async () => { assert.ok(postgresContract.requiredTables.includes("agentrun_sessions")); assert.ok(postgresContract.requiredTables.includes("agentrun_runner_jobs")); assert.ok(postgresContract.requiredTables.includes("agentrun_queue_tasks")); + assert.ok(postgresContract.requiredTables.includes("agentrun_queue_attempts")); assert.ok(postgresContract.requiredTables.includes("agentrun_queue_read_cursors")); assert.ok(postgresContract.requiredTables.includes("agentrun_cancel_requests")); assert.ok(postgresContract.requiredTables.includes("agentrun_runner_dispatch_intents")); diff --git a/src/selftest/cases/78-queue-q5-retry.ts b/src/selftest/cases/78-queue-q5-retry.ts new file mode 100644 index 0000000..5dd0b62 --- /dev/null +++ b/src/selftest/cases/78-queue-q5-retry.ts @@ -0,0 +1,325 @@ +import assert from "node:assert/strict"; +import { spawn } from "node:child_process"; +import { chmod, rm, writeFile } from "node:fs/promises"; +import path from "node:path"; +import type { CreateRunInput, JsonRecord, QueueAttemptListResult, QueueRetryResult, QueueTaskRecord } from "../../common/types.js"; +import { ManagerClient } from "../../mgr/client.js"; +import { createKubernetesRunnerJob, type RunnerJobDefaults } from "../../mgr/kubernetes-runner-job.js"; +import { startManagerServer } from "../../mgr/server.js"; +import { MemoryAgentRunStore } from "../../mgr/store.js"; +import { postgresMigrationContract } from "../../mgr/postgres-store.js"; +import type { SelfTestCase } from "../harness.js"; + +class FenceObservingMemoryStore extends MemoryAgentRunStore { + readonly observedFenceKeys: string[] = []; + maxConcurrentFenceOperations = 0; + private concurrentFenceOperations = 0; + + resetFenceObservation(): void { + this.observedFenceKeys.length = 0; + this.maxConcurrentFenceOperations = 0; + this.concurrentFenceOperations = 0; + } + + override async withRunnerCreateFence(fenceKey: string, operation: () => Promise): Promise { + this.observedFenceKeys.push(fenceKey); + return await super.withRunnerCreateFence(fenceKey, async () => { + this.concurrentFenceOperations += 1; + this.maxConcurrentFenceOperations = Math.max(this.maxConcurrentFenceOperations, this.concurrentFenceOperations); + try { + return await operation(); + } finally { + this.concurrentFenceOperations -= 1; + } + }); + } +} + +const selfTest: SelfTestCase = async (context) => { + const fakeKubectl = path.join(context.tmp, "fake-kubectl-queue-q5.js"); + const failMarker = path.join(context.tmp, "fake-kubectl-queue-q5.fail"); + await writeFile(fakeKubectl, `#!/usr/bin/env bun +const args = Bun.argv.slice(2); +if (args[0] !== "create") { + console.error("unsupported fake kubectl args: " + args.join(" ")); + process.exit(1); +} +const chunks = []; +for await (const chunk of Bun.stdin.stream()) chunks.push(Buffer.from(chunk)); +const manifest = JSON.parse(Buffer.concat(chunks).toString("utf8")); +if (await Bun.file(${JSON.stringify(failMarker)}).exists()) { + console.error("simulated queue q5 job create failure"); + process.exit(1); +} +await Bun.sleep(20); +console.log(JSON.stringify({ apiVersion: manifest.apiVersion, kind: manifest.kind, metadata: { uid: "queue-q5-" + manifest.metadata.name, resourceVersion: "1", name: manifest.metadata.name, namespace: manifest.metadata.namespace } })); +`); + await chmod(fakeKubectl, 0o755); + const runnerJobDefaults = { + namespace: "agentrun-v01", + managerUrl: "http://agentrun-mgr.agentrun-v01.svc.cluster.local:8080", + runnerApiKeySecretRef: { name: "agentrun-selftest-api-key", key: "HWLAB_API_KEY" }, + image: "127.0.0.1:5000/agentrun/agentrun-mgr@sha256:1111111111111111111111111111111111111111111111111111111111111111", + serviceAccountName: "agentrun-v01-runner", + jobNamePrefix: "agentrun-v01-runner", + lane: "v0.1", + kubectlCommand: fakeKubectl, + }; + const resolvedRunnerJobDefaults: RunnerJobDefaults = { ...runnerJobDefaults, sourceCommit: "self-test" }; + const store = new FenceObservingMemoryStore(); + const server = await startManagerServer({ + port: 0, + host: "127.0.0.1", + sourceCommit: "self-test", + store, + runnerJobDefaults, + }); + try { + const client = new ManagerClient(server.baseUrl); + const failed = await createTerminalTask(client, context.workspace, "failed", "queue-q5-failed"); + const before = await attempts(client, failed.task.id); + assert.equal(before.count, 1); + assert.equal(before.items[0]?.retryIndex, 0); + assert.equal(before.items[0]?.state, "failed"); + assert.equal(before.items[0]?.runId, failed.runId); + assert.equal(before.items[0]?.commandId, failed.commandId); + assert.ok(before.items[0]?.runnerJobId); + const oldAttempt = structuredClone(before.items[0]); + const versionBeforeDryRun = (await client.get(`/api/v1/queue/tasks/${failed.task.id}`) as QueueTaskRecord).version; + + const dryRun = await client.post(`/api/v1/queue/tasks/${failed.task.id}/retry`, { idempotencyKey: "retry-failed-1", reason: "dependency repaired", dryRun: true }) as QueueRetryResult; + assert.equal(dryRun.mutation, false); + assert.equal(dryRun.attempt, null); + assert.equal((dryRun.allowedTransition as JsonRecord).from, "failed"); + assert.equal((dryRun.allowedTransition as JsonRecord).retryIndex, 1); + assert.equal((await attempts(client, failed.task.id)).count, 1); + assert.equal((await client.get(`/api/v1/queue/tasks/${failed.task.id}`) as QueueTaskRecord).version, versionBeforeDryRun); + + await assert.rejects( + () => client.post(`/api/v1/queue/tasks/${failed.task.id}/retry`, { idempotencyKey: "retry-invalid-patch", reason: "must reject patch", dryRun: true, payload: { prompt: "changed" } }), + /does not accept task or dispatch patches/u, + ); + assert.equal((await attempts(client, failed.task.id)).count, 1); + + await writeFile(failMarker, "fail once\n"); + await assert.rejects( + () => client.post(`/api/v1/queue/tasks/${failed.task.id}/retry`, { idempotencyKey: "retry-failed-1", reason: "dependency repaired" }), + /kubectl create runner job failed/u, + ); + const failedAfterJobError = await client.get(`/api/v1/queue/tasks/${failed.task.id}`) as QueueTaskRecord; + assert.equal(failedAfterJobError.state, "failed"); + assert.equal(failedAfterJobError.latestAttempt?.attemptId, oldAttempt?.attemptId); + const reserved = await attempts(client, failed.task.id); + assert.equal(reserved.count, 2); + assert.equal(reserved.items[0]?.state, "reserved"); + assert.equal(reserved.items[0]?.failureKind, "infra-failed"); + assert.match(reserved.items[0]?.failureMessage ?? "", /kubectl create runner job failed/u); + const failedAttemptsCli = await runCliJson(context.root, server.baseUrl, ["queue", "attempts", failed.task.id]); + const failedAttemptsItems = (failedAttemptsCli.data as JsonRecord).items as JsonRecord[]; + assert.equal(failedAttemptsItems[0]?.failureKind, "infra-failed"); + assert.match(String(failedAttemptsItems[0]?.failureMessage ?? ""), /kubectl create runner job failed/u); + await rm(failMarker); + + const [first, duplicate] = await Promise.all([ + client.post(`/api/v1/queue/tasks/${failed.task.id}/retry`, { idempotencyKey: "retry-failed-1", reason: "dependency repaired" }) as Promise, + client.post(`/api/v1/queue/tasks/${failed.task.id}/retry`, { idempotencyKey: "retry-failed-1", reason: "dependency repaired" }) as Promise, + ]); + assert.ok(first.attempt); + assert.ok(duplicate.attempt); + assert.equal(first.attempt?.attemptId, duplicate.attempt?.attemptId); + assert.equal(first.attempt?.runId, duplicate.attempt?.runId); + assert.equal(first.attempt?.commandId, duplicate.attempt?.commandId); + assert.equal(first.attempt?.runnerJobId, duplicate.attempt?.runnerJobId); + assert.equal([first.mutation, duplicate.mutation].filter(Boolean).length, 1); + const retryRunId = String(first.attempt?.runId); + const retryCommandId = String(first.attempt?.commandId); + const jobs = await client.get(`/api/v1/runs/${retryRunId}/runner-jobs?commandId=${retryCommandId}`) as { count?: number; items?: JsonRecord[] }; + assert.equal(jobs.count, 1); + assert.equal(jobs.items?.[0]?.id, first.attempt?.runnerJobId); + assert.equal(jobs.items?.[0]?.attemptId, first.attempt?.attemptId); + const after = await attempts(client, failed.task.id); + assert.equal(after.count, 2); + assert.deepEqual(after.items.find((item) => item.attemptId === oldAttempt?.attemptId), oldAttempt); + assert.equal(after.items[0]?.retryIndex, 1); + assert.equal(after.items[0]?.state, "running"); + assert.equal(after.items[0]?.failureKind, null); + assert.equal(after.items[0]?.failureMessage, null); + assert.equal(after.items[0]?.previousAttemptId, oldAttempt?.attemptId); + assert.equal(after.items[0]?.payloadHash, failed.task.payloadHash); + + const replay = await client.post(`/api/v1/queue/tasks/${failed.task.id}/retry`, { idempotencyKey: "retry-failed-1", reason: "dependency repaired" }) as QueueRetryResult; + assert.equal(replay.mutation, false); + assert.equal(replay.idempotentReplay, true); + assert.equal(replay.attempt?.attemptId, first.attempt?.attemptId); + await assert.rejects( + () => client.post(`/api/v1/queue/tasks/${failed.task.id}/retry`, { idempotencyKey: "retry-failed-1", reason: "different reason" }), + /different reason/u, + ); + await assert.rejects( + () => client.post(`/api/v1/queue/tasks/${failed.task.id}/retry`, { idempotencyKey: "retry-failed-2", reason: "different key while running" }), + /cannot be retried from state running/u, + ); + + const cliDryRun = await runCliJson(context.root, server.baseUrl, ["queue", "retry", failed.task.id, "--idempotency-key", "retry-failed-1", "--reason", "dependency repaired", "--dry-run"]); + assert.equal(cliDryRun.ok, true); + assert.equal((cliDryRun.data as JsonRecord).action, "queue-retry-summary"); + assert.equal((cliDryRun.data as JsonRecord).mutation, false); + const cliAttempts = await runCliJson(context.root, server.baseUrl, ["queue", "attempts", failed.task.id]); + assert.equal(cliAttempts.ok, true); + assert.equal((cliAttempts.data as JsonRecord).action, "queue-attempt-list-summary"); + assert.equal((cliAttempts.data as JsonRecord).count, 2); + + await client.patch(`/api/v1/commands/${retryCommandId}/status`, { terminalStatus: "completed", failureKind: null, failureMessage: null }); + await client.patch(`/api/v1/runs/${retryRunId}/status`, { terminalStatus: "completed", failureKind: null, failureMessage: null }); + const terminalAttempts = await attempts(client, failed.task.id); + assert.equal(terminalAttempts.items[0]?.state, "completed"); + assert.deepEqual(terminalAttempts.items.find((item) => item.attemptId === oldAttempt?.attemptId), oldAttempt); + + const completed = await createTerminalTask(client, context.workspace, "completed", "queue-q5-completed"); + await assert.rejects( + () => client.post(`/api/v1/queue/tasks/${completed.task.id}/retry`, { idempotencyKey: "retry-completed", reason: "must fail closed" }), + /cannot be retried from state completed/u, + ); + const cancelled = await createCancelledTask(client, context.workspace, "queue-q5-cancelled"); + await assert.rejects( + () => client.post(`/api/v1/queue/tasks/${cancelled.id}/retry`, { idempotencyKey: "retry-cancelled", reason: "must fail closed" }), + /cannot be retried from state cancelled/u, + ); + const blocked = await createTerminalTask(client, context.workspace, "blocked", "queue-q5-blocked"); + const blockedPlan = await client.post(`/api/v1/queue/tasks/${blocked.task.id}/retry`, { idempotencyKey: "retry-blocked", reason: "operator unblocked dependency", dryRun: true }) as QueueRetryResult; + assert.equal((blockedPlan.allowedTransition as JsonRecord).from, "blocked"); + assert.equal(blockedPlan.mutation, false); + + store.resetFenceObservation(); + const fenceRunA = store.createRun(directRunInput(context.workspace, "fence-a")); + const fenceRunB = store.createRun(directRunInput(context.workspace, "fence-b")); + const fenceCommandA = store.createCommand(fenceRunA.id, { type: "turn", payload: { prompt: "fence-a" } }); + const fenceCommandB = store.createCommand(fenceRunB.id, { type: "turn", payload: { prompt: "fence-b" } }); + await Promise.all([ + createKubernetesRunnerJob({ store, runId: fenceRunA.id, input: { commandId: fenceCommandA.id, attemptId: "attempt-fence-a", idempotencyKey: "fence-a" }, defaults: resolvedRunnerJobDefaults }), + createKubernetesRunnerJob({ store, runId: fenceRunB.id, input: { commandId: fenceCommandB.id, attemptId: "attempt-fence-b", idempotencyKey: "fence-b" }, defaults: resolvedRunnerJobDefaults }), + ]); + assert.equal(new Set(store.observedFenceKeys).size, 2); + assert.equal(store.maxConcurrentFenceOperations, 2); + + const migration = postgresMigrationContract(); + assert.equal(migration.latestMigrationId, "014_v02_queue_attempt_history_retry"); + assert.ok((migration.requiredTables as string[]).includes("agentrun_queue_attempts")); + assert.deepEqual((migration.queueRetryConcurrency as JsonRecord).attemptIdentity, ["task_id", "attempt_id"]); + assert.deepEqual((migration.queueRetryConcurrency as JsonRecord).retryIdentity, ["task_id", "idempotency_key"]); + assert.deepEqual((migration.queueRetryConcurrency as JsonRecord).activeAttemptFence, ["task_id", "state IN (reserved,running)"]); + assert.equal((migration.queueRetryConcurrency as JsonRecord).legacyBackfill, "agentrun_queue_tasks.latest_attempt"); + assert.deepEqual((migration.queueRetryConcurrency as JsonRecord).failureVisibility, ["failure_kind", "failure_message"]); + return { + name: "queue-q5-retry", + tests: [ + "retry-server-dry-run-zero-write", + "initial-dispatch-attempt-history", + "retry-rejects-task-patch", + "retry-same-key-concurrency", + "retry-job-failure-keeps-task-terminal", + "retry-job-failure-visible-on-reserved-attempt", + "retry-reserved-attempt-same-key-resume", + "retry-stable-run-command-job-identity", + "retry-old-attempt-immutable", + "retry-same-key-before-state", + "retry-different-key-running-rejected", + "retry-completed-cancelled-rejected", + "retry-failed-blocked-allowed", + "retry-native-cli", + "attempts-read-refreshes-latest-from-core", + "runner-create-fence-does-not-serialize-different-runs", + "retry-postgres-migration-concurrency-contract", + ], + }; + } finally { + await new Promise((resolve) => server.server.close(() => resolve())); + } +}; + +export default selfTest; + +async function createTerminalTask(client: ManagerClient, workspace: string, terminalStatus: "completed" | "failed" | "blocked", key: string): Promise<{ task: QueueTaskRecord; runId: string; commandId: string }> { + const task = await client.post("/api/v1/queue/tasks", taskInput(workspace, key)) as QueueTaskRecord; + const dispatched = await client.post(`/api/v1/queue/tasks/${task.id}/dispatch`, { attemptId: `attempt_${key}` }) as JsonRecord; + const runId = String((dispatched.run as JsonRecord).id); + const commandId = String((dispatched.command as JsonRecord).id); + const failureKind = terminalStatus === "blocked" ? "required-skill-unavailable" : terminalStatus === "failed" ? "infra-failed" : null; + await client.patch(`/api/v1/commands/${commandId}/status`, { terminalStatus, failureKind, failureMessage: failureKind ? `${key} terminal` : null }); + await client.patch(`/api/v1/runs/${runId}/status`, { terminalStatus, failureKind, failureMessage: failureKind ? `${key} terminal` : null }); + return { task: await client.get(`/api/v1/queue/tasks/${task.id}`) as QueueTaskRecord, runId, commandId }; +} + +async function createCancelledTask(client: ManagerClient, workspace: string, key: string): Promise { + const task = await client.post("/api/v1/queue/tasks", taskInput(workspace, key)) as QueueTaskRecord; + await client.post(`/api/v1/queue/tasks/${task.id}/dispatch`, { attemptId: `attempt_${key}` }); + return await client.post(`/api/v1/queue/tasks/${task.id}/cancel`, { reason: "self-test cancelled" }) as QueueTaskRecord; +} + +function taskInput(workspace: string, key: string): JsonRecord { + return { + tenantId: "unidesk", + projectId: "pikasTech/agentrun", + queue: "dev", + lane: "q5", + title: key, + priority: 10, + backendProfile: "codex", + providerId: "G14", + workspaceRef: { kind: "host-path", path: workspace }, + sessionRef: null, + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 15_000, + network: "default", + secretScope: { allowCredentialEcho: false, providerCredentials: [{ profile: "codex", secretRef: { name: "agentrun-v01-provider-codex", keys: ["auth.json", "config.toml"] } }] }, + }, + resourceBundleRef: null, + payload: { prompt: `execute ${key}` }, + references: [{ kind: "issue", url: "https://github.com/pikasTech/agentrun/issues/307" }], + metadata: { source: "queue-q5-self-test" }, + idempotencyKey: key, + }; +} + +function directRunInput(workspace: string, key: string): CreateRunInput { + return { + tenantId: "unidesk", + projectId: "pikasTech/agentrun", + workspaceRef: { kind: "host-path", path: workspace }, + sessionRef: null, + resourceBundleRef: null, + providerId: "G14", + backendProfile: "codex", + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 15_000, + network: "default", + secretScope: { allowCredentialEcho: false, providerCredentials: [{ profile: "codex", secretRef: { name: "agentrun-v01-provider-codex", keys: ["auth.json", "config.toml"] } }] }, + }, + traceSink: { kind: "self-test", key }, + }; +} + +async function attempts(client: ManagerClient, taskId: string): Promise { + return await client.get(`/api/v1/queue/tasks/${taskId}/attempts?limit=100`) as QueueAttemptListResult; +} + +async function runCliJson(root: string, managerUrl: string, args: string[]): Promise { + const proc = spawn(process.execPath, [`${root}/scripts/agentrun-cli.ts`, "--manager-url", managerUrl, ...args], { stdio: ["ignore", "pipe", "pipe"] }); + const [stdout, stderr, code] = await Promise.all([readStream(proc.stdout), readStream(proc.stderr), new Promise((resolve) => proc.on("close", resolve))]); + assert.equal(code, 0, stderr || stdout); + return JSON.parse(stdout) as JsonRecord; +} + +async function readStream(stream: NodeJS.ReadableStream): Promise { + const chunks: Buffer[] = []; + stream.on("data", (chunk: Buffer | string) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))); + await new Promise((resolve, reject) => { + stream.on("end", resolve); + stream.on("error", reject); + }); + return Buffer.concat(chunks).toString("utf8"); +} diff --git a/src/selftest/integration/postgres-queue-retry.ts b/src/selftest/integration/postgres-queue-retry.ts new file mode 100644 index 0000000..65ab157 --- /dev/null +++ b/src/selftest/integration/postgres-queue-retry.ts @@ -0,0 +1,105 @@ +import assert from "node:assert/strict"; +import { Pool } from "pg"; +import { AgentRunError } from "../../common/errors.js"; +import type { CreateQueueTaskInput, QueueAttemptRef } from "../../common/types.js"; +import { createPostgresAgentRunStore } from "../../mgr/postgres-store.js"; + +const connectionString = process.env.AGENTRUN_SELFTEST_DATABASE_URL?.trim(); +if (!connectionString) throw new Error("AGENTRUN_SELFTEST_DATABASE_URL is required; this integration test never falls back to memory"); + +const suffix = `${Date.now()}-${process.pid}`; +const control = new Pool({ connectionString, application_name: "agentrun-pg-queue-retry-control", max: 2 }); +const primary = await createPostgresAgentRunStore({ connectionString, poolMax: 4 }); +const concurrent = await createPostgresAgentRunStore({ connectionString, poolMax: 4 }); +let taskId: string | null = null; + +try { + const task = await primary.createQueueTask(queueTaskInput(suffix)); + taskId = task.id; + const initialAttempt: QueueAttemptRef = { + attemptId: `attempt_initial_${suffix}`, + state: "failed", + runId: null, + commandId: null, + runnerJobId: null, + sessionId: null, + sessionPath: null, + failureKind: "infra-failed", + failureMessage: "postgres queue retry initial failure", + }; + const failed = await primary.updateQueueTaskAttempt(task.id, { state: "failed", latestAttempt: initialAttempt, sessionPath: null }); + assert.equal(failed.state, "failed"); + assert.equal((await primary.listQueueAttempts(task.id, { limit: 100 })).count, 1); + + const versionBeforeDryRun = (await primary.getQueueTask(task.id)).version; + const dryRun = await primary.reserveQueueRetryAttempt(task.id, { idempotencyKey: `retry-dry-${suffix}`, reason: "postgres dry run", dryRun: true }); + assert.equal(dryRun.mutation, false); + assert.equal(dryRun.attempt, null); + assert.equal((await primary.getQueueTask(task.id)).version, versionBeforeDryRun); + assert.equal((await primary.listQueueAttempts(task.id, { limit: 100 })).count, 1); + + const retryInput = { idempotencyKey: `retry-shared-${suffix}`, reason: "postgres concurrent retry", dryRun: false } as const; + const [first, second] = await Promise.all([ + primary.reserveQueueRetryAttempt(task.id, retryInput), + concurrent.reserveQueueRetryAttempt(task.id, retryInput), + ]); + assert.ok(first.attempt); + assert.ok(second.attempt); + assert.equal(first.attempt?.attemptId, second.attempt?.attemptId); + assert.equal(first.attempt?.runId, second.attempt?.runId); + assert.equal(first.attempt?.runnerJobId, second.attempt?.runnerJobId); + assert.equal([first.mutation, second.mutation].filter(Boolean).length, 1); + assert.equal([first.idempotentReplay, second.idempotentReplay].filter(Boolean).length, 1); + assert.equal((await primary.listQueueAttempts(task.id, { limit: 100 })).count, 2); + assert.equal((await primary.getQueueTask(task.id)).state, "failed"); + assert.equal((await primary.getQueueTask(task.id)).latestAttempt?.attemptId, initialAttempt.attemptId); + + await assert.rejects( + () => concurrent.reserveQueueRetryAttempt(task.id, { idempotencyKey: `retry-different-${suffix}`, reason: "must hit active attempt fence", dryRun: false }), + (error) => error instanceof AgentRunError && error.httpStatus === 409 && /active retry attempt/u.test(error.message), + ); + + console.log(JSON.stringify({ + ok: true, + tests: [ + "postgres-queue-retry-dry-run-zero-write", + "postgres-queue-retry-same-key-concurrent-single-reservation", + "postgres-queue-retry-stable-identities", + "postgres-queue-retry-different-key-active-fence", + ], + taskId: task.id, + attemptId: first.attempt?.attemptId ?? null, + valuesPrinted: false, + })); +} finally { + if (taskId) await control.query("DELETE FROM agentrun_queue_tasks WHERE id = $1", [taskId]).catch(() => undefined); + await Promise.all([primary.close(), concurrent.close()]); + await control.end(); +} + +function queueTaskInput(key: string): CreateQueueTaskInput { + return { + tenantId: "selftest", + projectId: "pikasTech/agentrun", + queue: "integration", + lane: "postgres-q5", + title: "PostgreSQL Queue retry concurrency", + priority: 10, + backendProfile: "codex", + providerId: "NC01", + workspaceRef: { kind: "host-path", path: "/tmp/agentrun-postgres-queue-retry" }, + sessionRef: null, + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 60_000, + network: "default", + secretScope: { allowCredentialEcho: false, providerCredentials: [] }, + }, + resourceBundleRef: null, + payload: { prompt: "postgres queue retry concurrency self-test" }, + references: [{ kind: "issue", url: "https://github.com/pikasTech/agentrun/issues/307" }], + metadata: { source: "postgres-queue-retry-self-test" }, + idempotencyKey: `postgres-queue-retry-${key}`, + }; +} diff --git a/src/selftest/integration/postgres-runner-retention-fence.ts b/src/selftest/integration/postgres-runner-retention-fence.ts index 88915ca..cb70959 100644 --- a/src/selftest/integration/postgres-runner-retention-fence.ts +++ b/src/selftest/integration/postgres-runner-retention-fence.ts @@ -10,10 +10,11 @@ const concurrent = await createPostgresAgentRunStore({ connectionString }); try { await assertNamespaceCreateFence(); + await assertIndependentCreateFencesDoNotSerialize(); await assertRetentionFenceBlocksClaim(); await assertRetentionTerminalizationIsIdempotent(); await assertRetentionTerminalizationRollsBackOnCasFailure(); - console.log(JSON.stringify({ ok: true, tests: ["postgres-namespace-create-fence", "postgres-run-retention-fence-blocks-claim", "postgres-run-retention-terminalization-idempotent", "postgres-run-retention-cas-failure-rolls-back-terminalization"] })); + console.log(JSON.stringify({ ok: true, tests: ["postgres-namespace-create-fence", "postgres-independent-create-fences-do-not-serialize", "postgres-run-retention-fence-blocks-claim", "postgres-run-retention-terminalization-idempotent", "postgres-run-retention-cas-failure-rolls-back-terminalization"] })); } finally { await Promise.all([primary.close(), concurrent.close()]); } @@ -40,6 +41,30 @@ async function assertNamespaceCreateFence(): Promise { assert.deepEqual(await Promise.all([first, second]), ["first", "second"]); } +async function assertIndependentCreateFencesDoNotSerialize(): Promise { + let firstEntered = (): void => {}; + let secondEntered = (): void => {}; + let releaseFirst = (): void => {}; + const enteredFirst = new Promise((resolve) => { firstEntered = resolve; }); + const enteredSecond = new Promise((resolve) => { secondEntered = resolve; }); + const release = new Promise((resolve) => { releaseFirst = resolve; }); + const first = primary.withRunnerCreateFence("request:postgres-fine-fence-a", async () => { + firstEntered(); + await release; + return "first"; + }); + await enteredFirst; + const second = primary.withRunnerCreateFence("request:postgres-fine-fence-b", async () => { + secondEntered(); + return "second"; + }); + const enteredWithoutSerialization = await Promise.race([enteredSecond.then(() => true), delay(250).then(() => false)]); + releaseFirst(); + const results = await Promise.all([first, second]); + assert.equal(enteredWithoutSerialization, true, "independent PostgreSQL runner create fence was serialized"); + assert.deepEqual(results, ["first", "second"]); +} + async function assertRetentionFenceBlocksClaim(): Promise { const run = await primary.createRun({ tenantId: "selftest",