feat: 实现 Queue Q1 API 和 CLI 骨架
This commit is contained in:
Codex
+194
-4
@@ -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, CommandRecord, CommandState, CreateCommandInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, RunEvent, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionRecord, SessionRef, TerminalStatus } from "../common/types.js";
|
||||
import type { BackendProfile, BackendTurnResult, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionRecord, SessionRef, TerminalStatus } from "../common/types.js";
|
||||
import { newId, nowIso, stableHash } from "../common/validation.js";
|
||||
import type { AgentRunStore, SaveRunnerJobInput, StoreHealth } from "./store.js";
|
||||
import { assertSessionBoundary, commandStateFromTerminal, isTerminalCommandState, isTerminalRunStatus, sessionRefFromRecord, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef } from "./store.js";
|
||||
import type { AgentRunStore, ListQueueTasksInput, SaveRunnerJobInput, StoreHealth } from "./store.js";
|
||||
import { assertSessionBoundary, buildQueueStats, clampQueueLimit, commandStateFromTerminal, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, parseQueueCursor, queueTaskSort, sessionRefFromRecord, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef } from "./store.js";
|
||||
import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js";
|
||||
import { normalizeRunEventPayload, requireEventType } from "../common/events.js";
|
||||
|
||||
@@ -171,6 +171,53 @@ CREATE INDEX IF NOT EXISTS agentrun_runner_jobs_run_command_idx ON agentrun_runn
|
||||
CREATE INDEX IF NOT EXISTS agentrun_sessions_tenant_project_idx ON agentrun_sessions (tenant_id, project_id, backend_profile, updated_at);
|
||||
`;
|
||||
|
||||
const queueQ1MigrationSql = `
|
||||
CREATE TABLE IF NOT EXISTS agentrun_queue_tasks (
|
||||
id text PRIMARY KEY,
|
||||
tenant_id text NOT NULL,
|
||||
project_id text NOT NULL,
|
||||
queue text NOT NULL,
|
||||
lane text NOT NULL,
|
||||
title text NOT NULL,
|
||||
priority integer NOT NULL,
|
||||
state text NOT NULL,
|
||||
version bigint NOT NULL,
|
||||
backend_profile text NOT NULL,
|
||||
provider_id text,
|
||||
workspace_ref jsonb,
|
||||
execution_policy jsonb,
|
||||
resource_bundle_ref jsonb,
|
||||
payload jsonb NOT NULL,
|
||||
payload_hash text NOT NULL,
|
||||
references_json jsonb NOT NULL DEFAULT '[]'::jsonb,
|
||||
metadata jsonb NOT NULL DEFAULT '{}'::jsonb,
|
||||
idempotency_key text,
|
||||
latest_attempt jsonb,
|
||||
session_path text,
|
||||
created_at timestamptz NOT NULL,
|
||||
updated_at timestamptz NOT NULL,
|
||||
cancelled_at timestamptz,
|
||||
cancel_reason text
|
||||
);
|
||||
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS agentrun_queue_tasks_idempotency_idx
|
||||
ON agentrun_queue_tasks (tenant_id, project_id, idempotency_key)
|
||||
WHERE idempotency_key IS NOT NULL;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS agentrun_queue_tasks_list_idx ON agentrun_queue_tasks (queue, state, version, priority, created_at);
|
||||
CREATE INDEX IF NOT EXISTS agentrun_queue_tasks_updated_idx ON agentrun_queue_tasks (version, updated_at);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS agentrun_queue_read_cursors (
|
||||
task_id text NOT NULL REFERENCES agentrun_queue_tasks(id) ON DELETE CASCADE,
|
||||
reader_id text NOT NULL,
|
||||
task_version bigint NOT NULL,
|
||||
read_at timestamptz NOT NULL,
|
||||
PRIMARY KEY (task_id, reader_id)
|
||||
);
|
||||
|
||||
CREATE SEQUENCE IF NOT EXISTS agentrun_queue_version_seq;
|
||||
`;
|
||||
|
||||
const postgresMigrations: MigrationDefinition[] = [
|
||||
{
|
||||
id: "001_v01_initial_durable_store",
|
||||
@@ -187,13 +234,18 @@ const postgresMigrations: MigrationDefinition[] = [
|
||||
checksum: checksumSql(hwlabManualDispatchMigrationSql),
|
||||
sql: hwlabManualDispatchMigrationSql,
|
||||
},
|
||||
{
|
||||
id: "004_v01_queue_q1",
|
||||
checksum: checksumSql(queueQ1MigrationSql),
|
||||
sql: queueQ1MigrationSql,
|
||||
},
|
||||
];
|
||||
|
||||
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_runner_jobs"],
|
||||
requiredTables: ["agentrun_schema_migrations", "agentrun_runs", "agentrun_commands", "agentrun_events", "agentrun_runners", "agentrun_backends", "agentrun_leases", "agentrun_sessions", "agentrun_runner_jobs", "agentrun_queue_tasks", "agentrun_queue_read_cursors"],
|
||||
checksums: Object.fromEntries(postgresMigrations.map((migration) => [migration.id, migration.checksum])),
|
||||
};
|
||||
}
|
||||
@@ -508,6 +560,96 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
return result.rows[0] ? sessionFromRow(result.rows[0]) : null;
|
||||
}
|
||||
|
||||
async createQueueTask(input: CreateQueueTaskInput): Promise<QueueTaskRecord> {
|
||||
const payloadHash = stableHash(input.payload);
|
||||
return this.withTransaction(async (client) => {
|
||||
if (input.idempotencyKey) {
|
||||
const existing = await client.query("SELECT * FROM agentrun_queue_tasks WHERE tenant_id = $1 AND project_id = $2 AND idempotency_key = $3 FOR UPDATE", [input.tenantId, input.projectId, input.idempotencyKey]);
|
||||
if (existing.rows[0]) {
|
||||
const record = queueTaskFromRow(existing.rows[0]);
|
||||
if (record.payloadHash !== payloadHash) throw new AgentRunError("schema-invalid", "queue task idempotency key reused with different payload", { httpStatus: 409 });
|
||||
return record;
|
||||
}
|
||||
}
|
||||
const at = nowIso();
|
||||
const version = await this.nextQueueVersion(client);
|
||||
const task: QueueTaskRecord = { ...input, id: newId("qt"), state: "pending", version, payloadHash, latestAttempt: null, sessionPath: null, createdAt: at, updatedAt: at, cancelledAt: null, cancelReason: null };
|
||||
const inserted = await client.query(
|
||||
`INSERT INTO agentrun_queue_tasks (id, tenant_id, project_id, queue, lane, title, priority, state, version, backend_profile, provider_id, workspace_ref, execution_policy, resource_bundle_ref, payload, payload_hash, references_json, metadata, idempotency_key, latest_attempt, session_path, created_at, updated_at, cancelled_at, cancel_reason)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12::jsonb, $13::jsonb, $14::jsonb, $15::jsonb, $16, $17::jsonb, $18::jsonb, $19, $20::jsonb, $21, $22, $23, $24, $25)
|
||||
RETURNING *`,
|
||||
[task.id, task.tenantId, task.projectId, task.queue, task.lane, task.title, task.priority, task.state, task.version, task.backendProfile, task.providerId, JSON.stringify(task.workspaceRef), JSON.stringify(task.executionPolicy), JSON.stringify(task.resourceBundleRef), JSON.stringify(task.payload), task.payloadHash, JSON.stringify(task.references), JSON.stringify(task.metadata), task.idempotencyKey ?? null, JSON.stringify(task.latestAttempt), task.sessionPath, task.createdAt, task.updatedAt, task.cancelledAt, task.cancelReason],
|
||||
);
|
||||
return queueTaskFromRow(inserted.rows[0]);
|
||||
});
|
||||
}
|
||||
|
||||
async listQueueTasks(input: ListQueueTasksInput): Promise<QueueTaskListResult> {
|
||||
const startVersion = parseQueueCursor(input.cursor) ?? input.updatedAfter ?? 0;
|
||||
const params: unknown[] = [startVersion];
|
||||
const where = ["version > $1"];
|
||||
if (input.queue) {
|
||||
params.push(input.queue);
|
||||
where.push(`queue = $${params.length}`);
|
||||
}
|
||||
if (input.state) {
|
||||
params.push(input.state);
|
||||
where.push(`state = $${params.length}`);
|
||||
}
|
||||
params.push(clampQueueLimit(input.limit));
|
||||
const result = await this.pool.query(`SELECT * FROM agentrun_queue_tasks WHERE ${where.join(" AND ")} ORDER BY priority DESC, created_at ASC, id ASC LIMIT $${params.length}`, params);
|
||||
const items = result.rows.map(queueTaskFromRow);
|
||||
return { items, count: items.length, cursor: items.length > 0 ? String(items[items.length - 1]?.version ?? startVersion) : null };
|
||||
}
|
||||
|
||||
async getQueueTask(taskId: string): Promise<QueueTaskRecord> {
|
||||
const result = await this.pool.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1", [taskId]);
|
||||
const row = result.rows[0];
|
||||
if (!row) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 });
|
||||
return queueTaskFromRow(row);
|
||||
}
|
||||
|
||||
async cancelQueueTask(taskId: string, reason = "cancel requested"): Promise<QueueTaskRecord> {
|
||||
return this.withTransaction(async (client) => {
|
||||
const existing = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]);
|
||||
const row = existing.rows[0];
|
||||
if (!row) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 });
|
||||
const task = queueTaskFromRow(row);
|
||||
if (isTerminalQueueTaskState(task.state)) return task;
|
||||
const at = nowIso();
|
||||
const version = await this.nextQueueVersion(client);
|
||||
const updated = await client.query("UPDATE agentrun_queue_tasks SET state = 'cancelled', version = $2, updated_at = $3, cancelled_at = $3, cancel_reason = $4 WHERE id = $1 RETURNING *", [taskId, version, at, reason]);
|
||||
return queueTaskFromRow(updated.rows[0]);
|
||||
});
|
||||
}
|
||||
|
||||
async markQueueTaskRead(taskId: string, readerId: string): Promise<QueueReadCursorRecord> {
|
||||
return this.withTransaction(async (client) => {
|
||||
const existing = await client.query("SELECT * FROM agentrun_queue_tasks WHERE id = $1 FOR UPDATE", [taskId]);
|
||||
const row = existing.rows[0];
|
||||
if (!row) throw new AgentRunError("schema-invalid", `queue task ${taskId} was not found`, { httpStatus: 404 });
|
||||
const task = queueTaskFromRow(row);
|
||||
const record: QueueReadCursorRecord = { taskId, readerId, taskVersion: task.version, readAt: nowIso() };
|
||||
await client.query(
|
||||
`INSERT INTO agentrun_queue_read_cursors (task_id, reader_id, task_version, read_at)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
ON CONFLICT (task_id, reader_id) DO UPDATE SET task_version = EXCLUDED.task_version, read_at = EXCLUDED.read_at`,
|
||||
[record.taskId, record.readerId, record.taskVersion, record.readAt],
|
||||
);
|
||||
return record;
|
||||
});
|
||||
}
|
||||
|
||||
async queueStats(queue?: string): Promise<QueueStats> {
|
||||
return buildQueueStats(await this.loadQueueTasksForProjection(queue), queue ?? null);
|
||||
}
|
||||
|
||||
async queueCommander(queue?: string): Promise<QueueCommanderSnapshot> {
|
||||
const tasks = await this.loadQueueTasksForProjection(queue);
|
||||
const generatedAt = nowIso();
|
||||
return { queue: queue ?? null, stats: buildQueueStats(tasks, queue ?? null, generatedAt), items: tasks.sort(queueTaskSort).slice(0, 20), generatedAt };
|
||||
}
|
||||
|
||||
async backends(): Promise<JsonRecord[]> {
|
||||
const result = await this.pool.query("SELECT * FROM agentrun_backends ORDER BY profile ASC");
|
||||
return result.rows.map((row) => {
|
||||
@@ -534,6 +676,20 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
return Number(result.rows[0]?.seq ?? 1);
|
||||
}
|
||||
|
||||
private async nextQueueVersion(client: PoolClient): Promise<number> {
|
||||
const result = await client.query<{ version: string | number }>("SELECT nextval('agentrun_queue_version_seq') AS version");
|
||||
return Number(result.rows[0]?.version ?? 1);
|
||||
}
|
||||
|
||||
private async loadQueueTasksForProjection(queue?: string): Promise<QueueTaskRecord[]> {
|
||||
if (queue) {
|
||||
const result = await this.pool.query("SELECT * FROM agentrun_queue_tasks WHERE queue = $1", [queue]);
|
||||
return result.rows.map(queueTaskFromRow);
|
||||
}
|
||||
const result = await this.pool.query("SELECT * FROM agentrun_queue_tasks");
|
||||
return result.rows.map(queueTaskFromRow);
|
||||
}
|
||||
|
||||
private async requireRunForUpdate(client: PoolClient, runId: string): Promise<RunRecord> {
|
||||
const result = await client.query("SELECT * FROM agentrun_runs WHERE id = $1 FOR UPDATE", [runId]);
|
||||
const row = result.rows[0];
|
||||
@@ -727,6 +883,36 @@ function runnerJobFromRow(row: QueryResultRow): RunnerJobRecord {
|
||||
};
|
||||
}
|
||||
|
||||
function queueTaskFromRow(row: QueryResultRow): QueueTaskRecord {
|
||||
return {
|
||||
id: stringValue(row.id),
|
||||
tenantId: stringValue(row.tenant_id),
|
||||
projectId: stringValue(row.project_id),
|
||||
queue: stringValue(row.queue),
|
||||
lane: stringValue(row.lane),
|
||||
title: stringValue(row.title),
|
||||
priority: Number(row.priority),
|
||||
state: stringValue(row.state) as QueueTaskState,
|
||||
version: Number(row.version),
|
||||
backendProfile: stringValue(row.backend_profile) as BackendProfile,
|
||||
providerId: nullableString(row.provider_id),
|
||||
workspaceRef: jsonValue(row.workspace_ref) as QueueTaskRecord["workspaceRef"],
|
||||
executionPolicy: jsonValue(row.execution_policy) as QueueTaskRecord["executionPolicy"],
|
||||
resourceBundleRef: jsonValue(row.resource_bundle_ref) as QueueTaskRecord["resourceBundleRef"],
|
||||
payload: jsonRecord(row.payload),
|
||||
payloadHash: stringValue(row.payload_hash),
|
||||
references: jsonArray(row.references_json) as JsonRecord[],
|
||||
metadata: jsonRecord(row.metadata),
|
||||
...(nullableString(row.idempotency_key) ? { idempotencyKey: stringValue(row.idempotency_key) } : {}),
|
||||
latestAttempt: jsonValue(row.latest_attempt) as QueueTaskRecord["latestAttempt"],
|
||||
sessionPath: nullableString(row.session_path),
|
||||
createdAt: iso(row.created_at),
|
||||
updatedAt: iso(row.updated_at),
|
||||
cancelledAt: nullableIso(row.cancelled_at),
|
||||
cancelReason: nullableString(row.cancel_reason),
|
||||
};
|
||||
}
|
||||
|
||||
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);
|
||||
@@ -749,6 +935,10 @@ function jsonRecord(value: unknown): JsonRecord {
|
||||
return typeof value === "object" && value !== null && !Array.isArray(value) ? value as JsonRecord : {};
|
||||
}
|
||||
|
||||
function jsonArray(value: unknown): JsonValue[] {
|
||||
return Array.isArray(value) ? value as JsonValue[] : [];
|
||||
}
|
||||
|
||||
function iso(value: unknown): string {
|
||||
if (value instanceof Date) return value.toISOString();
|
||||
if (typeof value === "string") return new Date(value).toISOString();
|
||||
|
||||
Reference in New Issue
Block a user