feat: 支持 Queue 终态幂等重试
This commit is contained in:
root
@@ -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"
|
||||
|
||||
@@ -115,6 +115,8 @@ async function dispatch(args: ParsedArgs): Promise<CliResult> {
|
||||
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<Json
|
||||
return summarizeQueueDispatchResult(result, taskId);
|
||||
}
|
||||
|
||||
async function retryQueueTaskCli(args: ParsedArgs, taskId: string): Promise<JsonValue> {
|
||||
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<JsonValue> {
|
||||
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<JsonRecord> {
|
||||
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 <taskId> [--reader-id <reader>] [--dry-run] [--full|--raw]",
|
||||
"queue cancel <taskId> [--reason <text>] [--dry-run] [--full|--raw]",
|
||||
"queue dispatch <taskId> [--json-stdin|--json-file <dispatch.json>] [--idempotency-key <key>] [--image <image>] [--namespace <namespace>] [--dry-run] [--full|--raw]",
|
||||
"queue retry <taskId> --idempotency-key <key> --reason <text> [--dry-run] [--full|--raw]",
|
||||
"queue attempts <taskId> [--cursor <retry-index>] [--limit <n>] [--full|--raw]",
|
||||
"queue refresh <taskId> [--dry-run] [--full|--raw]",
|
||||
"aipod-specs list",
|
||||
"aipod-specs show <name>",
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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<JsonRecord> {
|
||||
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<JsonRecord> {
|
||||
@@ -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 });
|
||||
|
||||
+270
-11
@@ -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<RunRecord> {
|
||||
async createRun(input: CreateRunInput, identity?: CreateRunIdentity): Promise<RunRecord> {
|
||||
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<T>(namespace: string, operation: () => Promise<T>): Promise<T> {
|
||||
async withRunnerCreateFence<T>(fenceKey: string, operation: () => Promise<T>): Promise<T> {
|
||||
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<QueueAttemptListResult> {
|
||||
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<QueueRetryReservation> {
|
||||
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<QueueAttemptRecord> {
|
||||
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<QueueRetryActivation> {
|
||||
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<QueueTaskRecord> {
|
||||
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<void> {
|
||||
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<CancelRequestRecord> {
|
||||
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);
|
||||
|
||||
@@ -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");
|
||||
|
||||
@@ -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<QueueRetryResult> {
|
||||
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<ReturnType<AgentRunStore["reserveQueueRetryAttempt"]>>): Promise<QueueRetryResult> {
|
||||
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<ReturnType<AgentRunStore["reserveQueueRetryAttempt"]>>;
|
||||
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;
|
||||
}
|
||||
@@ -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({
|
||||
|
||||
+313
-11
@@ -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<StoreHealth>;
|
||||
createRun(input: CreateRunInput): MaybePromise<RunRecord>;
|
||||
createRun(input: CreateRunInput, identity?: CreateRunIdentity): MaybePromise<RunRecord>;
|
||||
getRun(runId: string): MaybePromise<RunRecord>;
|
||||
listEvents(runId: string, afterSeq: number, limit: number): MaybePromise<RunEvent[]>;
|
||||
listEventsForCommand(runId: string, commandId: string, limit: number): MaybePromise<RunEvent[]>;
|
||||
@@ -71,7 +71,7 @@ export interface AgentRunStore {
|
||||
getRunnerJobByIdempotencyKey(runId: string, idempotencyKey: string, payloadHash: string): MaybePromise<RunnerJobRecord | null>;
|
||||
saveRunnerJob(input: SaveRunnerJobInput): MaybePromise<RunnerJobRecord>;
|
||||
updateRunnerJobResult(runnerJobId: string, patch: JsonRecord): MaybePromise<RunnerJobRecord>;
|
||||
withRunnerCreateFence<T>(namespace: string, operation: () => Promise<T>): MaybePromise<T>;
|
||||
withRunnerCreateFence<T>(fenceKey: string, operation: () => Promise<T>): MaybePromise<T>;
|
||||
withRunnerRetentionFence<T>(input: RunnerRetentionFenceInput, operation: (facts: RunnerRetentionFenceFacts) => Promise<T>): MaybePromise<T>;
|
||||
claimRun(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
|
||||
heartbeat(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
|
||||
@@ -94,6 +94,10 @@ export interface AgentRunStore {
|
||||
createQueueTask(input: CreateQueueTaskInput): MaybePromise<QueueTaskRecord>;
|
||||
listQueueTasks(input: ListQueueTasksInput): MaybePromise<QueueTaskListResult>;
|
||||
getQueueTask(taskId: string): MaybePromise<QueueTaskRecord>;
|
||||
listQueueAttempts(taskId: string, input: ListQueueAttemptsInput): MaybePromise<QueueAttemptListResult>;
|
||||
reserveQueueRetryAttempt(taskId: string, input: ReserveQueueRetryAttemptInput): MaybePromise<QueueRetryReservation>;
|
||||
recordQueueRetryAttemptFailure(taskId: string, attemptId: string, input: RecordQueueRetryAttemptFailureInput): MaybePromise<QueueAttemptRecord>;
|
||||
activateQueueRetryAttempt(taskId: string, attemptId: string, input: ActivateQueueRetryAttemptInput): MaybePromise<QueueRetryActivation>;
|
||||
updateQueueTaskAttempt(taskId: string, input: UpdateQueueTaskAttemptInput): MaybePromise<QueueTaskRecord>;
|
||||
cancelQueueTask(taskId: string, reason?: string): MaybePromise<QueueTaskRecord>;
|
||||
markQueueTaskRead(taskId: string, readerId: string): MaybePromise<QueueReadCursorRecord>;
|
||||
@@ -103,6 +107,10 @@ export interface AgentRunStore {
|
||||
close?(): MaybePromise<void>;
|
||||
}
|
||||
|
||||
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<QueueTaskState, "pending">;
|
||||
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<string, RunnerDispatchIntentRecord>();
|
||||
private readonly kafkaEventOutbox = new Map<string, KafkaEventOutboxRecord>();
|
||||
private readonly queueTasks = new Map<string, QueueTaskRecord>();
|
||||
private readonly queueAttempts = new Map<string, QueueAttemptRecord>();
|
||||
private readonly queueReadCursors = new Map<string, QueueReadCursorRecord>();
|
||||
private readonly runnerCreateFenceTails = new Map<string, Promise<void>>();
|
||||
private readonly runnerRetentionFences = new Set<string>();
|
||||
@@ -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<T>(namespace: string, operation: () => Promise<T>): Promise<T> {
|
||||
const predecessor = this.runnerCreateFenceTails.get(namespace) ?? Promise.resolve();
|
||||
async withRunnerCreateFence<T>(fenceKey: string, operation: () => Promise<T>): Promise<T> {
|
||||
const predecessor = this.runnerCreateFenceTails.get(fenceKey) ?? Promise.resolve();
|
||||
let release = (): void => {};
|
||||
const current = new Promise<void>((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<QueueAttemptRecord | null>((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<string, number> = {};
|
||||
const byLane: Record<string, number> = {};
|
||||
@@ -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}`;
|
||||
}
|
||||
|
||||
@@ -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<string, string>)["008_v01_dsflash_go_backend_profile"] === "string" && (postgresContract.checksums as Record<string, string>)["008_v01_dsflash_go_backend_profile"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["009_v01_dsflash_go_model_catalog"] === "string" && (postgresContract.checksums as Record<string, string>)["009_v01_dsflash_go_model_catalog"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["010_v01_queue_session_ref"] === "string" && (postgresContract.checksums as Record<string, string>)["010_v01_queue_session_ref"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["011_v01_cancel_lifecycle"] === "string" && (postgresContract.checksums as Record<string, string>)["011_v01_cancel_lifecycle"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["012_v02_durable_dispatch_kafka_outbox"] === "string" && (postgresContract.checksums as Record<string, string>)["012_v02_durable_dispatch_kafka_outbox"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["013_v02_runner_dispatch_outcome"] === "string" && (postgresContract.checksums as Record<string, string>)["013_v02_runner_dispatch_outcome"].length > 0);
|
||||
assert.ok(typeof (postgresContract.checksums as Record<string, string>)["014_v02_queue_attempt_history_retry"] === "string" && (postgresContract.checksums as Record<string, string>)["014_v02_queue_attempt_history_retry"].length > 0);
|
||||
assert.equal((postgresContract.checksums as Record<string, string>)["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"));
|
||||
|
||||
@@ -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<T>(fenceKey: string, operation: () => Promise<T>): Promise<T> {
|
||||
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<QueueRetryResult>,
|
||||
client.post(`/api/v1/queue/tasks/${failed.task.id}/retry`, { idempotencyKey: "retry-failed-1", reason: "dependency repaired" }) as Promise<QueueRetryResult>,
|
||||
]);
|
||||
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<void>((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<QueueTaskRecord> {
|
||||
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<QueueAttemptListResult> {
|
||||
return await client.get(`/api/v1/queue/tasks/${taskId}/attempts?limit=100`) as QueueAttemptListResult;
|
||||
}
|
||||
|
||||
async function runCliJson(root: string, managerUrl: string, args: string[]): Promise<JsonRecord> {
|
||||
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<number | null>((resolve) => proc.on("close", resolve))]);
|
||||
assert.equal(code, 0, stderr || stdout);
|
||||
return JSON.parse(stdout) as JsonRecord;
|
||||
}
|
||||
|
||||
async function readStream(stream: NodeJS.ReadableStream): Promise<string> {
|
||||
const chunks: Buffer[] = [];
|
||||
stream.on("data", (chunk: Buffer | string) => chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)));
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
stream.on("end", resolve);
|
||||
stream.on("error", reject);
|
||||
});
|
||||
return Buffer.concat(chunks).toString("utf8");
|
||||
}
|
||||
@@ -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}`,
|
||||
};
|
||||
}
|
||||
@@ -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<void> {
|
||||
assert.deepEqual(await Promise.all([first, second]), ["first", "second"]);
|
||||
}
|
||||
|
||||
async function assertIndependentCreateFencesDoNotSerialize(): Promise<void> {
|
||||
let firstEntered = (): void => {};
|
||||
let secondEntered = (): void => {};
|
||||
let releaseFirst = (): void => {};
|
||||
const enteredFirst = new Promise<void>((resolve) => { firstEntered = resolve; });
|
||||
const enteredSecond = new Promise<void>((resolve) => { secondEntered = resolve; });
|
||||
const release = new Promise<void>((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<void> {
|
||||
const run = await primary.createRun({
|
||||
tenantId: "selftest",
|
||||
|
||||
Reference in New Issue
Block a user