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

feat: 支持同 run runner 多轮 command

Browse Source
This commit is contained in:
Codex
2026-06-01 22:33:26 +08:00
parent 5fb008f5cf
commit 6fb8f7483a
14 changed files with 237 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mgr/postgres-store.ts
+9 -12
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, 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, ListQueueTasksInput, SaveRunnerJobInput, StoreHealth } from "./store.js";
import { assertSessionBoundary, buildQueueStats, clampQueueLimit, commandStateFromTerminal, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, parseQueueCursor, queueTaskSort, sessionRefFromRecord, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef } from "./store.js";
import { assertSessionBoundary, buildQueueStats, clampQueueLimit, commandStateFromTerminal, isLeaseExpired, 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";
@@ -332,7 +332,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
async createCommand(runId: string, input: CreateCommandInput): Promise<CommandRecord> {
const payloadHash = stableHash(input.payload);
return this.withTransaction(async (client) => {
await this.requireRunForUpdate(client, runId);
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]) {
@@ -436,7 +437,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
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 (run.claimedBy && run.claimedBy !== runnerId) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
if (run.claimedBy && run.claimedBy !== runnerId && !isLeaseExpired(run.leaseExpiresAt)) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
const leaseExpiresAt = new Date(Date.now() + leaseMs).toISOString();
const updated = await client.query(
`UPDATE agentrun_runs SET status = $2, claimed_by = $3, lease_expires_at = $4, updated_at = $5 WHERE id = $1 RETURNING *`,
@@ -478,7 +479,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
});
}
async finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): Promise<CommandRecord> {
async finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): Promise<CommandRecord> {
return this.withTransaction(async (client) => {
const existing = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [commandId]);
const row = existing.rows[0];
@@ -487,7 +488,9 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
if (isTerminalCommandState(command.state)) return command;
const state = commandStateFromTerminal(result.terminalStatus);
const updated = await client.query("UPDATE agentrun_commands SET state = $2, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, state, nowIso()]);
await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state, terminalStatus: result.terminalStatus, failureKind: result.failureKind });
const run = await this.requireRunForUpdate(client, command.runId);
if (result.threadId && run.sessionRef?.sessionId) await this.upsertSessionThread(client, run, result.threadId, result.turnId ?? null);
await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state, terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null });
return commandFromRow(updated.rows[0]);
});
}
@@ -544,13 +547,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
const command = commandFromRow(row);
if (isTerminalCommandState(command.state)) return command;
const updated = await client.query("UPDATE agentrun_commands SET state = 'cancelled', updated_at = $2 WHERE id = $1 RETURNING *", [commandId, nowIso()]);
await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled" });
const run = await this.requireRunForUpdate(client, command.runId);
if (!isTerminalRunStatus(run.status)) {
await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "cancel-requested", reason });
await client.query(`UPDATE agentrun_runs SET status = 'cancelled', terminal_status = 'cancelled', failure_kind = 'cancelled', failure_message = $2, updated_at = $3 WHERE id = $1`, [command.runId, reason, nowIso()]);
await this.appendEventWithLockedRun(client, command.runId, "terminal_status", { terminalStatus: "cancelled", failureKind: "cancelled", message: reason });
}
await this.appendEventWithLockedRun(client, command.runId, "backend_status", { phase: "command-terminal", commandId, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", message: reason });
return commandFromRow(updated.rows[0]);
});
}
src/mgr/result.ts
+23 -8
View File
@@ -6,14 +6,17 @@ export async function buildRunResult(store: AgentRunStore, runId: string, comman
const run = await store.getRun(runId);
const command = await selectCommand(store, runId, commandId);
const events = await store.listEvents(runId, 0, 500);
const scopedEvents = command ? eventsForCommand(events, command.id) : events;
const jobs = await store.listRunnerJobs(runId, command?.id);
const latestJob = jobs.at(-1) ?? null;
const terminalEventStatus = terminalFromEvents(events);
const terminal = terminalEventStatus ?? run.terminalStatus;
const terminalSource = terminalEventStatus ? "terminal_status-event" : run.terminalStatus ? "run-record" : "none";
const failureKind = run.failureKind ?? failureKindFromEvents(events);
const reply = assistantReply(events);
const blocker = terminal === "blocked" || terminal === "failed" ? { failureKind, message: run.failureMessage ?? messageFromEvents(events) } : null;
const commandTerminal = command ? terminalFromCommand(command) : null;
const terminalEventStatus = terminalFromEvents(scopedEvents);
const terminal = commandTerminal ?? terminalEventStatus ?? run.terminalStatus;
const terminalSource = commandTerminal ? "command-record" : terminalEventStatus ? "terminal_status-event" : run.terminalStatus ? "run-record" : "none";
const failureKind = command ? failureKindFromEvents(scopedEvents) : run.failureKind ?? failureKindFromEvents(scopedEvents);
const failureMessage = command ? messageFromEvents(scopedEvents) : run.failureMessage ?? messageFromEvents(scopedEvents);
const reply = assistantReply(scopedEvents);
const blocker = terminal === "blocked" || terminal === "failed" ? { failureKind, message: failureMessage } : null;
return {
runId: run.id,
commandId: command?.id ?? commandId ?? null,
@@ -29,11 +32,11 @@ export async function buildRunResult(store: AgentRunStore, runId: string, comman
completed: terminal === "completed",
reply,
failureKind,
failureMessage: run.failureMessage ?? messageFromEvents(events),
failureMessage,
blocker,
lastSeq: events.at(-1)?.seq ?? 0,
eventCount: events.length,
artifactSummary: artifactSummary(events),
artifactSummary: artifactSummary(scopedEvents),
sessionRef: sessionSummary(run),
resourceBundleRef: resourceBundleSummary(run, events),
runnerJobCount: jobs.length,
@@ -58,6 +61,18 @@ function terminalFromEvents(events: RunEvent[]): TerminalStatus | null {
return null;
}
function terminalFromCommand(command: CommandRecord): TerminalStatus | null {
if (command.state === "completed") return "completed";
if (command.state === "failed") return "failed";
if (command.state === "cancelled") return "cancelled";
return null;
}
function eventsForCommand(events: RunEvent[], commandId: string): RunEvent[] {
const scoped = events.filter((event) => event.payload.commandId === commandId || typeof event.payload.commandId !== "string");
return scoped.length > 0 ? scoped : events;
}
function failureKindFromEvents(events: RunEvent[]): string | null {
for (const event of [...events].reverse()) {
const value = event.payload.failureKind;
src/mgr/runner-job-status.ts
+4 -2
View File
@@ -1,7 +1,7 @@
import type { JsonRecord, RunEvent, RunnerJobRecord, TerminalStatus } from "../common/types.js";
export function runnerJobStatusSummary(job: RunnerJobRecord, events: RunEvent[] = []): JsonRecord {
const terminalEvent = latestTerminalEvent(events);
const terminalEvent = latestTerminalEvent(events, job.commandId);
const runner = recordAt(job.result, "runner");
const jobIdentity = recordAt(job.result, "jobIdentity");
const kubernetes = recordAt(job.result, "kubernetes");
@@ -36,9 +36,11 @@ export function runnerJobStatusSummary(job: RunnerJobRecord, events: RunEvent[]
};
}
function latestTerminalEvent(events: RunEvent[]): RunEvent | null {
function latestTerminalEvent(events: RunEvent[], commandId: string): RunEvent | null {
for (const event of [...events].reverse()) {
if (event.payload.commandId && event.payload.commandId !== commandId) continue;
if (event.type === "terminal_status") return event;
if (event.type === "backend_status" && event.payload.phase === "command-terminal" && event.payload.commandId === commandId) return event;
}
return null;
}
src/mgr/server.ts
+7 -1
View File
@@ -191,7 +191,13 @@ async function route({ method, url, body, store, sourceCommit, runnerJobDefaults
if (method === "PATCH" && commandStatusMatch) {
const record = asRecord(body, "commandStatus");
const terminalStatus = record.terminalStatus === "completed" || record.terminalStatus === "failed" || record.terminalStatus === "blocked" || record.terminalStatus === "cancelled" ? record.terminalStatus : "failed";
return await store.finishCommand(commandStatusMatch[1] ?? "", { terminalStatus, failureKind: typeof record.failureKind === "string" ? record.failureKind as never : null, failureMessage: typeof record.failureMessage === "string" ? record.failureMessage : null }) as unknown as JsonValue;
return await store.finishCommand(commandStatusMatch[1] ?? "", {
terminalStatus,
failureKind: typeof record.failureKind === "string" ? record.failureKind as never : null,
failureMessage: typeof record.failureMessage === "string" ? record.failureMessage : null,
...(typeof record.threadId === "string" ? { threadId: record.threadId } : {}),
...(typeof record.turnId === "string" ? { turnId: record.turnId } : {}),
}) as unknown as JsonValue;
}
const commandCancelMatch = path.match(/^\/api\/v1\/commands\/([^/]+)\/cancel$/u);
if (method === "POST" && commandCancelMatch) {
src/mgr/store.ts
+14 -7
View File
@@ -33,7 +33,7 @@ export interface AgentRunStore {
claimRun(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
heartbeat(runId: string, runnerId: string, leaseMs: number): MaybePromise<RunRecord>;
ackCommand(commandId: string): MaybePromise<CommandRecord>;
finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): MaybePromise<CommandRecord>;
finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): MaybePromise<CommandRecord>;
appendEvent(runId: string, type: RunEvent["type"], payload: JsonRecord): MaybePromise<RunEvent>;
finishRun(runId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): MaybePromise<RunRecord>;
cancelRun(runId: string, reason?: string): MaybePromise<RunRecord>;
@@ -122,7 +122,8 @@ export class MemoryAgentRunStore implements AgentRunStore {
}
createCommand(runId: string, input: CreateCommandInput): CommandRecord {
this.getRun(runId);
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);
@@ -185,7 +186,7 @@ export class MemoryAgentRunStore implements AgentRunStore {
claimRun(runId: string, runnerId: string, leaseMs: number): RunRecord {
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 });
if (run.claimedBy && run.claimedBy !== runnerId) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
if (run.claimedBy && run.claimedBy !== runnerId && !isLeaseExpired(run.leaseExpiresAt)) throw new AgentRunError("runner-lease-conflict", `run ${runId} is already claimed`, { httpStatus: 409 });
const next = this.updateRun(runId, { status: "claimed", claimedBy: runnerId, leaseExpiresAt: new Date(Date.now() + leaseMs).toISOString() });
this.appendEvent(runId, "backend_status", { phase: "run-claimed", runnerId });
return next;
@@ -206,12 +207,14 @@ export class MemoryAgentRunStore implements AgentRunStore {
return next;
}
finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage">): CommandRecord {
finishCommand(commandId: string, result: Pick<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): CommandRecord {
const command = this.getCommand(commandId);
if (isTerminalCommandState(command.state)) return command;
const next = { ...command, state: commandStateFromTerminal(result.terminalStatus), updatedAt: nowIso() };
this.commands.set(commandId, next);
this.appendEvent(command.runId, "backend_status", { phase: "command-terminal", commandId, state: next.state, terminalStatus: result.terminalStatus, failureKind: result.failureKind });
const run = this.getRun(command.runId);
if (result.threadId && run.sessionRef?.sessionId) this.upsertSessionThread(run, result.threadId, result.turnId ?? null);
this.appendEvent(command.runId, "backend_status", { phase: "command-terminal", commandId, state: next.state, terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage ?? null, threadId: result.threadId ?? null, turnId: result.turnId ?? null });
return next;
}
@@ -256,8 +259,7 @@ export class MemoryAgentRunStore implements AgentRunStore {
if (isTerminalCommandState(command.state)) return command;
const next = { ...command, state: "cancelled" as const, updatedAt: nowIso() };
this.commands.set(commandId, next);
this.appendEvent(command.runId, "backend_status", { phase: "command-terminal", commandId, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled" });
this.cancelRun(command.runId, reason);
this.appendEvent(command.runId, "backend_status", { phase: "command-terminal", commandId, state: "cancelled", terminalStatus: "cancelled", failureKind: "cancelled", message: reason });
return next;
}
@@ -419,6 +421,11 @@ export function isTerminalQueueTaskState(state: QueueTaskState): boolean {
return state === "completed" || state === "failed" || state === "blocked" || state === "cancelled";
}
export function isLeaseExpired(leaseExpiresAt: string | null): boolean {
if (!leaseExpiresAt) return true;
return new Date(leaseExpiresAt).getTime() <= Date.now();
}
export function sessionRefFromRecord(record: SessionRecord, fallback: SessionRef): SessionRef {
return {
sessionId: record.sessionId,