mirrors/pikasTech-agentrun
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 修复 Queue 已读投影和取消传播

Browse Source
This commit is contained in:
Codex
2026-06-09 01:55:42 +08:00
parent 424058e40b
commit 88ed2c1791
7 changed files with 132 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mgr/postgres-store.ts
+30 -6
View File
@@ -6,7 +6,7 @@ import { redactJson } from "../common/redaction.js";
import type { BackendProfile, BackendTurnResult, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, ListGcExpiredSessionsInput, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, 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, ListQueueTasksInput, ListSessionsInput, SaveRunnerJobInput, SessionEventPageInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js";
import { assertSessionBoundary, buildQueueStats, buildSessionSummary, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, parseQueueCursor, parseSessionCursor, queueTaskPayloadHash, queueTaskSort, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef, titleFromMetadata } from "./store.js";
import { assertSessionBoundary, buildQueueStats, buildQueueTaskSummary, buildSessionSummary, clampQueueLimit, clampSessionLimit, commandStateFromTerminal, isLeaseExpired, isSessionOutputEvent, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, parseQueueCursor, parseSessionCursor, queueTaskMatchesCommander, queueTaskPayloadHash, queueTaskSort, sessionListFilters, sessionMatchesListState, sessionRefFromRecord, sessionSort, sessionTitleFromCommand, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef, titleFromMetadata } from "./store.js";
import { backendCapabilitiesSqlValues, mergeBackendCapability } from "../common/backend-profiles.js";
import { normalizeRunEventPayload, requireEventType } from "../common/events.js";
@@ -983,15 +983,20 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
}
async cancelQueueTask(taskId: string, reason = "cancel requested"): Promise<QueueTaskRecord> {
const task = await this.getQueueTask(taskId);
if (isTerminalQueueTaskState(task.state)) return task;
if (task.latestAttempt?.commandId) await this.cancelCommand(task.latestAttempt.commandId, reason);
else if (task.latestAttempt?.runId) await this.cancelRun(task.latestAttempt.runId, reason);
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 lockedTask = queueTaskFromRow(row);
if (isTerminalQueueTaskState(lockedTask.state)) return lockedTask;
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]);
const latestAttempt = lockedTask.latestAttempt ? { ...lockedTask.latestAttempt, state: "cancelled" as const } : null;
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]);
});
}
@@ -1017,10 +1022,16 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
return buildQueueStats(await this.loadQueueTasksForProjection(queue), queue ?? null);
}
async queueCommander(queue?: string): Promise<QueueCommanderSnapshot> {
async queueCommander(queue?: string, readerId: string | null = null): 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 };
const cursors = readerId ? await this.loadQueueReadCursors(readerId, tasks.map((task) => task.id)) : new Map<string, QueueReadCursorRecord>();
const items = tasks
.map((task) => buildQueueTaskSummary(task, readerId, readerId ? cursors.get(task.id) ?? null : null))
.filter((task) => queueTaskMatchesCommander(task, readerId))
.sort(queueTaskSort)
.slice(0, 20);
return { queue: queue ?? null, readerId, stats: buildQueueStats(tasks, queue ?? null, generatedAt), items, generatedAt };
}
async backends(): Promise<JsonRecord[]> {
@@ -1079,6 +1090,15 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
}));
}
private async loadQueueReadCursors(readerId: string, taskIds: string[]): Promise<Map<string, QueueReadCursorRecord>> {
if (taskIds.length === 0) return new Map();
const result = await this.pool.query("SELECT * FROM agentrun_queue_read_cursors WHERE reader_id = $1 AND task_id = ANY($2::text[])", [readerId, taskIds]);
return new Map(result.rows.map((row) => {
const cursor = queueReadCursorFromRow(row);
return [cursor.taskId, cursor];
}));
}
private async loadQueueTasksForProjection(queue?: string): Promise<QueueTaskRecord[]> {
if (queue) {
const result = await this.pool.query("SELECT * FROM agentrun_queue_tasks WHERE queue = $1", [queue]);
@@ -1376,6 +1396,10 @@ function sessionReadCursorFromRow(row: QueryResultRow): SessionReadCursorRecord
return { sessionId: stringValue(row.session_id), readerId: stringValue(row.reader_id), sessionVersion: Number(row.session_version), readAt: iso(row.read_at) };
}
function queueReadCursorFromRow(row: QueryResultRow): QueueReadCursorRecord {
return { taskId: stringValue(row.task_id), readerId: stringValue(row.reader_id), taskVersion: Number(row.task_version), readAt: iso(row.read_at) };
}
function sessionExecutionState(value: unknown): SessionRecord["executionState"] {
if (value === "running" || value === "terminal") return value;
return "idle";
src/mgr/server.ts
+1 -1
View File
@@ -262,7 +262,7 @@ async function route({ method, url, body, store, sourceCommit, runnerJobDefaults
return await store.markQueueTaskRead(queueTaskReadMatch[1] ?? "", readerId) as unknown as JsonValue;
}
if (method === "GET" && path === "/api/v1/queue/stats") return await store.queueStats(url.searchParams.get("queue") ?? undefined) as unknown as JsonValue;
if (method === "GET" && path === "/api/v1/queue/commander") return await store.queueCommander(url.searchParams.get("queue") ?? undefined) as unknown as JsonValue;
if (method === "GET" && path === "/api/v1/queue/commander") return await store.queueCommander(url.searchParams.get("queue") ?? undefined, url.searchParams.get("readerId")) as unknown as JsonValue;
if (method === "POST" && path === "/api/v1/runs") return await store.createRun(validateCreateRun(body)) as unknown as JsonValue;
const runMatch = path.match(/^\/api\/v1\/runs\/([^/]+)$/u);
if (method === "GET" && runMatch) return await store.getRun(runMatch[1] ?? "") as unknown as JsonValue;
src/mgr/store.ts
+26 -6
View File
@@ -1,4 +1,4 @@
import type { BackendProfile, BackendTurnResult, CommandRecord, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, FailureKind, JsonRecord, ListGcExpiredSessionsInput, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerJobRecord, RunnerRecord, RunRecord, SessionEventPage, SessionListResult, SessionListState, SessionReadCursorRecord, SessionRecord, SessionRef, SessionStoragePatch, SessionSummary, TerminalStatus, UpsertSessionInput } from "../common/types.js";
import type { BackendProfile, BackendTurnResult, CommandRecord, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, FailureKind, JsonRecord, ListGcExpiredSessionsInput, QueueAttemptRef, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, QueueTaskSummary, RunEvent, 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";
@@ -55,7 +55,7 @@ export interface AgentRunStore {
cancelQueueTask(taskId: string, reason?: string): MaybePromise<QueueTaskRecord>;
markQueueTaskRead(taskId: string, readerId: string): MaybePromise<QueueReadCursorRecord>;
queueStats(queue?: string): MaybePromise<QueueStats>;
queueCommander(queue?: string): MaybePromise<QueueCommanderSnapshot>;
queueCommander(queue?: string, readerId?: string | null): MaybePromise<QueueCommanderSnapshot>;
backends(): MaybePromise<JsonRecord[]>;
close?(): MaybePromise<void>;
}
@@ -495,8 +495,11 @@ export class MemoryAgentRunStore implements AgentRunStore {
cancelQueueTask(taskId: string, reason = "cancel requested"): QueueTaskRecord {
const task = this.getQueueTask(taskId);
if (isTerminalQueueTaskState(task.state)) return task;
if (task.latestAttempt?.commandId) this.cancelCommand(task.latestAttempt.commandId, reason);
else if (task.latestAttempt?.runId) this.cancelRun(task.latestAttempt.runId, reason);
const at = nowIso();
const next: QueueTaskRecord = { ...task, state: "cancelled", version: this.nextQueueVersion(), updatedAt: at, cancelledAt: at, cancelReason: reason };
const latestAttempt = task.latestAttempt ? { ...task.latestAttempt, state: "cancelled" as const } : null;
const next: QueueTaskRecord = { ...task, state: "cancelled", latestAttempt, version: this.nextQueueVersion(), updatedAt: at, cancelledAt: at, cancelReason: reason };
this.queueTasks.set(taskId, next);
return next;
}
@@ -512,10 +515,15 @@ export class MemoryAgentRunStore implements AgentRunStore {
return buildQueueStats(Array.from(this.queueTasks.values()).filter((task) => !queue || task.queue === queue), queue ?? null);
}
queueCommander(queue?: string): QueueCommanderSnapshot {
const items = Array.from(this.queueTasks.values()).filter((task) => !queue || task.queue === queue).sort(queueTaskSort).slice(0, 20);
queueCommander(queue?: string, readerId: string | null = null): QueueCommanderSnapshot {
const tasks = Array.from(this.queueTasks.values()).filter((task) => !queue || task.queue === queue);
const generatedAt = nowIso();
return { queue: queue ?? null, stats: buildQueueStats(Array.from(this.queueTasks.values()).filter((task) => !queue || task.queue === queue), queue ?? null, generatedAt), items, generatedAt };
const items = tasks
.map((task) => buildQueueTaskSummary(task, readerId, readerId ? this.queueReadCursors.get(queueReadKey(task.id, readerId)) ?? null : null))
.filter((task) => queueTaskMatchesCommander(task, readerId))
.sort(queueTaskSort)
.slice(0, 20);
return { queue: queue ?? null, readerId, stats: buildQueueStats(tasks, queue ?? null, generatedAt), items, generatedAt };
}
backends(): JsonRecord[] {
@@ -686,6 +694,18 @@ export function buildSessionSummary(record: SessionRecord, readerId: string | nu
return { ...record, sessionPath: `${record.tenantId}/${record.projectId}/${record.sessionId}`, readerId, readCursor, attentionState, unread, active };
}
export function buildQueueTaskSummary(record: QueueTaskRecord, readerId: string | null, readCursor: QueueReadCursorRecord | null): QueueTaskSummary {
const active = !isTerminalQueueTaskState(record.state);
const unread = !active && (!readCursor || readCursor.taskVersion < record.version);
const attentionState = active ? "active" : unread ? "unread" : "read";
return { ...record, readerId, readCursor, attentionState, unread, active };
}
export function queueTaskMatchesCommander(task: QueueTaskSummary, readerId: string | null): boolean {
if (!readerId) return true;
return task.active || task.unread;
}
export function sessionMatchesListState(session: SessionSummary, state: SessionListState): boolean {
if (state === "all") return true;
if (state === "default") return session.active || session.unread;