feat: add session subagent cli control
This commit is contained in:
Codex
+261
-12
@@ -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, 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 type { BackendProfile, BackendTurnResult, CommandRecord, CommandState, CreateCommandInput, CreateQueueTaskInput, CreateRunInput, EventType, FailureKind, JsonRecord, JsonValue, QueueCommanderSnapshot, QueueReadCursorRecord, QueueStats, QueueTaskListResult, QueueTaskRecord, QueueTaskState, RunEvent, RunnerJobRecord, RunnerRecord, RunRecord, RunStatus, SessionEventPage, SessionListResult, SessionReadCursorRecord, SessionRecord, SessionRef, SessionSummary, TerminalStatus } from "../common/types.js";
|
||||
import { newId, nowIso, stableHash } from "../common/validation.js";
|
||||
import type { AgentRunStore, ListQueueTasksInput, SaveRunnerJobInput, StoreHealth, UpdateQueueTaskAttemptInput } from "./store.js";
|
||||
import { assertSessionBoundary, buildQueueStats, clampQueueLimit, commandStateFromTerminal, isLeaseExpired, isTerminalCommandState, isTerminalQueueTaskState, isTerminalRunStatus, parseQueueCursor, queueTaskSort, sessionRefFromRecord, statusFromTerminal, summarizeResourceBundleRef, summarizeSessionRef } from "./store.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, 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";
|
||||
|
||||
@@ -137,6 +137,44 @@ ON CONFLICT (profile) DO UPDATE SET
|
||||
updated_at = EXCLUDED.updated_at;
|
||||
`;
|
||||
|
||||
const sessionControlMigrationSql = `
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS version bigint NOT NULL DEFAULT 1;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS execution_state text NOT NULL DEFAULT 'idle';
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS last_run_id text;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS last_command_id text;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS active_run_id text;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS active_command_id text;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS last_event_seq integer NOT NULL DEFAULT 0;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS terminal_status text;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS failure_kind text;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS title text;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS summary jsonb NOT NULL DEFAULT '{}'::jsonb;
|
||||
ALTER TABLE agentrun_sessions ADD COLUMN IF NOT EXISTS last_activity_at timestamptz;
|
||||
|
||||
CREATE SEQUENCE IF NOT EXISTS agentrun_session_version_seq;
|
||||
|
||||
SELECT setval(
|
||||
'agentrun_session_version_seq',
|
||||
GREATEST(
|
||||
COALESCE((SELECT MAX(version) FROM agentrun_sessions), 0),
|
||||
1
|
||||
),
|
||||
true
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS agentrun_session_read_cursors (
|
||||
session_id text NOT NULL REFERENCES agentrun_sessions(session_id) ON DELETE CASCADE,
|
||||
reader_id text NOT NULL,
|
||||
session_version bigint NOT NULL,
|
||||
read_at timestamptz NOT NULL,
|
||||
PRIMARY KEY (session_id, reader_id)
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS agentrun_sessions_control_idx ON agentrun_sessions (execution_state, backend_profile, updated_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS agentrun_sessions_version_idx ON agentrun_sessions (version, updated_at DESC);
|
||||
CREATE INDEX IF NOT EXISTS agentrun_runs_session_idx ON agentrun_runs ((session_ref->>'sessionId'), updated_at DESC);
|
||||
`;
|
||||
|
||||
const hwlabManualDispatchMigrationSql = `
|
||||
ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS session_ref jsonb;
|
||||
ALTER TABLE agentrun_runs ADD COLUMN IF NOT EXISTS resource_bundle_ref jsonb;
|
||||
@@ -254,13 +292,18 @@ const postgresMigrations: MigrationDefinition[] = [
|
||||
checksum: checksumSql(minimaxM3BackendProfileMigrationSql),
|
||||
sql: minimaxM3BackendProfileMigrationSql,
|
||||
},
|
||||
{
|
||||
id: "006_v01_session_control",
|
||||
checksum: checksumSql(sessionControlMigrationSql),
|
||||
sql: sessionControlMigrationSql,
|
||||
},
|
||||
];
|
||||
|
||||
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", "agentrun_queue_tasks", "agentrun_queue_read_cursors"],
|
||||
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_queue_tasks", "agentrun_queue_read_cursors"],
|
||||
checksums: Object.fromEntries(postgresMigrations.map((migration) => [migration.id, migration.checksum])),
|
||||
};
|
||||
}
|
||||
@@ -326,6 +369,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
VALUES ($1, $2, $3, $4::jsonb, $5::jsonb, $6::jsonb, $7, $8, $9::jsonb, $10::jsonb, $11, $12, $13, $14, $15, $16, $17, $18)`,
|
||||
[run.id, run.tenantId, run.projectId, JSON.stringify(run.workspaceRef), JSON.stringify(run.sessionRef), JSON.stringify(run.resourceBundleRef), run.providerId, run.backendProfile, JSON.stringify(run.executionPolicy), JSON.stringify(run.traceSink), run.status, run.terminalStatus, run.failureKind, run.failureMessage, run.createdAt, run.updatedAt, run.claimedBy, run.leaseExpiresAt],
|
||||
);
|
||||
await this.touchSessionForRun(client, run, { lastRunId: run.id, lastActivityAt: at }, { bumpVersion: false, at });
|
||||
await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "run-created", backendProfile: run.backendProfile, sessionRef: summarizeSessionRef(run.sessionRef ?? null), resourceBundleRef: summarizeResourceBundleRef(run.resourceBundleRef ?? null) });
|
||||
return run;
|
||||
});
|
||||
@@ -365,6 +409,8 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
VALUES ($1, $2, $3, $4, $5::jsonb, $6, $7, $8, $9, $10, $11)`,
|
||||
[command.id, command.runId, command.seq, command.type, JSON.stringify(command.payload), command.payloadHash, command.idempotencyKey ?? null, command.state, command.createdAt, command.updatedAt, command.acknowledgedAt],
|
||||
);
|
||||
if (command.type === "turn") await this.touchSessionForRun(client, run, { executionState: "running", lastRunId: run.id, lastCommandId: command.id, activeRunId: run.id, activeCommandId: command.id, terminalStatus: null, failureKind: null, title: sessionTitleFromCommand(command), lastActivityAt: at }, { bumpVersion: true, at });
|
||||
else if (command.type === "steer") await this.touchSessionForRun(client, run, { executionState: "running", lastRunId: run.id, lastCommandId: command.id, activeRunId: run.id, lastActivityAt: at }, { bumpVersion: true, at });
|
||||
await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "command-created", commandId: command.id, commandType: command.type });
|
||||
return command;
|
||||
});
|
||||
@@ -464,8 +510,10 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
ON CONFLICT (run_id) DO UPDATE SET runner_id = EXCLUDED.runner_id, lease_expires_at = EXCLUDED.lease_expires_at, updated_at = EXCLUDED.updated_at`,
|
||||
[runId, runnerId, leaseExpiresAt, null, nowIso()],
|
||||
);
|
||||
const next = runFromRow(updated.rows[0]);
|
||||
await this.touchSessionForRun(client, next, { executionState: "running", activeRunId: runId, lastRunId: runId, lastActivityAt: next.updatedAt }, { bumpVersion: false, at: next.updatedAt });
|
||||
await this.appendEventWithLockedRun(client, runId, "backend_status", { phase: "run-claimed", runnerId });
|
||||
return runFromRow(updated.rows[0]);
|
||||
return next;
|
||||
});
|
||||
}
|
||||
|
||||
@@ -505,6 +553,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
const updated = await client.query("UPDATE agentrun_commands SET state = $2, updated_at = $3 WHERE id = $1 RETURNING *", [commandId, state, nowIso()]);
|
||||
const run = await this.requireRunForUpdate(client, command.runId);
|
||||
if (result.threadId && run.sessionRef?.sessionId) await this.upsertSessionThread(client, run, result.threadId, result.turnId ?? null);
|
||||
if (command.type === "turn") await this.touchSessionForRun(client, run, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: command.runId, lastCommandId: command.id, terminalStatus: result.terminalStatus, failureKind: result.failureKind, lastActivityAt: nowIso() }, { bumpVersion: true });
|
||||
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]);
|
||||
});
|
||||
@@ -529,6 +578,7 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
const run = runFromRow(updated.rows[0]);
|
||||
if (result.threadId && run.sessionRef?.sessionId) await this.upsertSessionThread(client, run, result.threadId, result.turnId ?? null);
|
||||
await this.appendEventWithLockedRun(client, runId, "terminal_status", { terminalStatus: result.terminalStatus, failureKind: result.failureKind, message: result.failureMessage });
|
||||
await this.touchSessionForRun(client, run, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: runId, terminalStatus: result.terminalStatus, failureKind: result.failureKind, lastActivityAt: run.updatedAt }, { bumpVersion: true, at: run.updatedAt });
|
||||
return run;
|
||||
});
|
||||
}
|
||||
@@ -550,7 +600,9 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
[runId, reason, at],
|
||||
);
|
||||
await this.appendEventWithLockedRun(client, runId, "terminal_status", { terminalStatus: "cancelled", failureKind: "cancelled", message: reason });
|
||||
return runFromRow(updated.rows[0]);
|
||||
const next = runFromRow(updated.rows[0]);
|
||||
await this.touchSessionForRun(client, next, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: runId, terminalStatus: "cancelled", failureKind: "cancelled", lastActivityAt: at }, { bumpVersion: true, at });
|
||||
return next;
|
||||
});
|
||||
}
|
||||
|
||||
@@ -563,6 +615,10 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
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", message: reason });
|
||||
if (command.type === "turn") {
|
||||
const run = await this.requireRunForUpdate(client, command.runId);
|
||||
await this.touchSessionForRun(client, run, { executionState: "terminal", activeRunId: null, activeCommandId: null, lastRunId: command.runId, lastCommandId: command.id, terminalStatus: "cancelled", failureKind: "cancelled", lastActivityAt: nowIso() }, { bumpVersion: true });
|
||||
}
|
||||
return commandFromRow(updated.rows[0]);
|
||||
});
|
||||
}
|
||||
@@ -572,6 +628,63 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
return result.rows[0] ? sessionFromRow(result.rows[0]) : null;
|
||||
}
|
||||
|
||||
async getSessionSummary(sessionId: string, readerId: string | null = null): Promise<SessionSummary> {
|
||||
const session = await this.getSession(sessionId);
|
||||
if (!session) throw new AgentRunError("schema-invalid", `session ${sessionId} was not found`, { httpStatus: 404 });
|
||||
const cursor = readerId ? await this.getSessionReadCursor(sessionId, readerId) : null;
|
||||
return buildSessionSummary(session, readerId, cursor);
|
||||
}
|
||||
|
||||
async listSessions(input: ListSessionsInput): Promise<SessionListResult> {
|
||||
const startVersion = parseSessionCursor(input.cursor) ?? 0;
|
||||
const state = input.state ?? "default";
|
||||
const params: unknown[] = [startVersion];
|
||||
const where = ["version > $1"];
|
||||
if (input.backendProfile) {
|
||||
params.push(input.backendProfile);
|
||||
where.push(`backend_profile = $${params.length}`);
|
||||
}
|
||||
const result = await this.pool.query(`SELECT * FROM agentrun_sessions WHERE ${where.join(" AND ")} ORDER BY updated_at DESC, session_id ASC LIMIT 500`, params);
|
||||
const cursors = input.readerId ? await this.loadSessionReadCursors(input.readerId, result.rows.map((row) => stringValue(row.session_id))) : new Map<string, SessionReadCursorRecord>();
|
||||
const items = result.rows
|
||||
.map(sessionFromRow)
|
||||
.map((session) => buildSessionSummary(session, input.readerId ?? null, input.readerId ? cursors.get(session.sessionId) ?? null : null))
|
||||
.filter((session) => sessionMatchesListState(session, state))
|
||||
.sort(sessionSort)
|
||||
.slice(0, clampSessionLimit(input.limit));
|
||||
return { items, count: items.length, cursor: items.length > 0 ? String(items[items.length - 1]?.version ?? startVersion) : null, filters: sessionListFilters(input) };
|
||||
}
|
||||
|
||||
async listSessionTrace(sessionId: string, input: SessionEventPageInput): Promise<SessionEventPage> {
|
||||
const runId = await this.resolveSessionRunId(sessionId, input.runId ?? null);
|
||||
if (!runId) return { sessionId, runId: null, items: [], count: 0, cursor: null };
|
||||
const items = await this.listEvents(runId, input.afterSeq, input.limit);
|
||||
return { sessionId, runId, items, count: items.length, cursor: items.length > 0 ? String(items[items.length - 1]?.seq ?? input.afterSeq) : null };
|
||||
}
|
||||
|
||||
async listSessionOutput(sessionId: string, input: SessionEventPageInput): Promise<SessionEventPage> {
|
||||
const page = await this.listSessionTrace(sessionId, input);
|
||||
const items = page.items.filter(isSessionOutputEvent);
|
||||
return { ...page, items, count: items.length, cursor: items.length > 0 ? String(items[items.length - 1]?.seq ?? input.afterSeq) : null };
|
||||
}
|
||||
|
||||
async markSessionRead(sessionId: string, readerId: string): Promise<SessionReadCursorRecord> {
|
||||
return this.withTransaction(async (client) => {
|
||||
const result = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [sessionId]);
|
||||
const row = result.rows[0];
|
||||
if (!row) throw new AgentRunError("schema-invalid", `session ${sessionId} was not found`, { httpStatus: 404 });
|
||||
const session = sessionFromRow(row);
|
||||
const record: SessionReadCursorRecord = { sessionId, readerId, sessionVersion: session.version, readAt: nowIso() };
|
||||
await client.query(
|
||||
`INSERT INTO agentrun_session_read_cursors (session_id, reader_id, session_version, read_at)
|
||||
VALUES ($1, $2, $3, $4)
|
||||
ON CONFLICT (session_id, reader_id) DO UPDATE SET session_version = EXCLUDED.session_version, read_at = EXCLUDED.read_at`,
|
||||
[record.sessionId, record.readerId, record.sessionVersion, record.readAt],
|
||||
);
|
||||
return record;
|
||||
});
|
||||
}
|
||||
|
||||
async createQueueTask(input: CreateQueueTaskInput): Promise<QueueTaskRecord> {
|
||||
const payloadHash = stableHash(input.payload);
|
||||
return this.withTransaction(async (client) => {
|
||||
@@ -697,6 +810,12 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
const seq = await this.nextSeq(client, "agentrun_events", runId);
|
||||
const event: RunEvent = { id: newId("evt"), runId, seq, type: eventType, payload: redactJson(eventPayload), createdAt: nowIso() };
|
||||
await client.query("INSERT INTO agentrun_events (id, run_id, seq, type, payload, created_at) VALUES ($1, $2, $3, $4, $5::jsonb, $6)", [event.id, event.runId, event.seq, event.type, JSON.stringify(event.payload), event.createdAt]);
|
||||
await client.query(
|
||||
`UPDATE agentrun_sessions
|
||||
SET last_event_seq = $2, last_activity_at = $3, updated_at = $3
|
||||
WHERE session_id = (SELECT session_ref->>'sessionId' FROM agentrun_runs WHERE id = $1)`,
|
||||
[runId, event.seq, event.createdAt],
|
||||
);
|
||||
return event;
|
||||
}
|
||||
|
||||
@@ -710,6 +829,25 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
return Number(result.rows[0]?.version ?? 1);
|
||||
}
|
||||
|
||||
private async nextSessionVersion(client: PoolClient): Promise<number> {
|
||||
const result = await client.query<{ version: string | number }>("SELECT nextval('agentrun_session_version_seq') AS version");
|
||||
return Number(result.rows[0]?.version ?? 1);
|
||||
}
|
||||
|
||||
private async getSessionReadCursor(sessionId: string, readerId: string): Promise<SessionReadCursorRecord | null> {
|
||||
const result = await this.pool.query("SELECT * FROM agentrun_session_read_cursors WHERE session_id = $1 AND reader_id = $2", [sessionId, readerId]);
|
||||
return result.rows[0] ? sessionReadCursorFromRow(result.rows[0]) : null;
|
||||
}
|
||||
|
||||
private async loadSessionReadCursors(readerId: string, sessionIds: string[]): Promise<Map<string, SessionReadCursorRecord>> {
|
||||
if (sessionIds.length === 0) return new Map();
|
||||
const result = await this.pool.query("SELECT * FROM agentrun_session_read_cursors WHERE reader_id = $1 AND session_id = ANY($2::text[])", [readerId, sessionIds]);
|
||||
return new Map(result.rows.map((row) => {
|
||||
const cursor = sessionReadCursorFromRow(row);
|
||||
return [cursor.sessionId, 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]);
|
||||
@@ -742,14 +880,26 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
conversationId: input.sessionRef.conversationId ?? null,
|
||||
threadId: input.sessionRef.threadId ?? null,
|
||||
metadata: input.sessionRef.metadata ?? {},
|
||||
version: await this.nextSessionVersion(client),
|
||||
executionState: "idle",
|
||||
lastRunId: null,
|
||||
lastCommandId: null,
|
||||
activeRunId: null,
|
||||
activeCommandId: null,
|
||||
lastEventSeq: 0,
|
||||
terminalStatus: null,
|
||||
failureKind: null,
|
||||
title: titleFromMetadata(input.sessionRef.metadata ?? {}),
|
||||
summary: {},
|
||||
lastActivityAt: at,
|
||||
createdAt: at,
|
||||
updatedAt: at,
|
||||
expiresAt: input.sessionRef.expiresAt ?? null,
|
||||
};
|
||||
await client.query(
|
||||
`INSERT INTO agentrun_sessions (session_id, tenant_id, project_id, backend_profile, conversation_id, thread_id, metadata, created_at, updated_at, expires_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10)`,
|
||||
[record.sessionId, record.tenantId, record.projectId, record.backendProfile, record.conversationId, record.threadId, JSON.stringify(record.metadata), record.createdAt, record.updatedAt, record.expiresAt],
|
||||
`INSERT INTO agentrun_sessions (session_id, tenant_id, project_id, backend_profile, conversation_id, thread_id, metadata, version, execution_state, last_run_id, last_command_id, active_run_id, active_command_id, last_event_seq, terminal_status, failure_kind, title, summary, last_activity_at, created_at, updated_at, expires_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18::jsonb, $19, $20, $21, $22)`,
|
||||
[record.sessionId, record.tenantId, record.projectId, record.backendProfile, record.conversationId, record.threadId, JSON.stringify(record.metadata), record.version, record.executionState, record.lastRunId, record.lastCommandId, record.activeRunId, record.activeCommandId, record.lastEventSeq, record.terminalStatus, record.failureKind, record.title, JSON.stringify(record.summary), record.lastActivityAt, record.createdAt, record.updatedAt, record.expiresAt],
|
||||
);
|
||||
return sessionRefFromRecord(record, input.sessionRef);
|
||||
}
|
||||
@@ -768,13 +918,25 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
conversationId: run.sessionRef.conversationId ?? existing?.conversationId ?? null,
|
||||
threadId,
|
||||
metadata,
|
||||
version: await this.nextSessionVersion(client),
|
||||
executionState: existing?.executionState ?? "idle",
|
||||
lastRunId: existing?.lastRunId ?? run.id,
|
||||
lastCommandId: existing?.lastCommandId ?? null,
|
||||
activeRunId: existing?.activeRunId ?? null,
|
||||
activeCommandId: existing?.activeCommandId ?? null,
|
||||
lastEventSeq: existing?.lastEventSeq ?? 0,
|
||||
terminalStatus: existing?.terminalStatus ?? null,
|
||||
failureKind: existing?.failureKind ?? null,
|
||||
title: existing?.title ?? titleFromMetadata(run.sessionRef.metadata ?? {}),
|
||||
summary: existing?.summary ?? {},
|
||||
lastActivityAt: at,
|
||||
createdAt: existing?.createdAt ?? at,
|
||||
updatedAt: at,
|
||||
expiresAt: run.sessionRef.expiresAt ?? existing?.expiresAt ?? null,
|
||||
};
|
||||
await client.query(
|
||||
`INSERT INTO agentrun_sessions (session_id, tenant_id, project_id, backend_profile, conversation_id, thread_id, metadata, created_at, updated_at, expires_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10)
|
||||
`INSERT INTO agentrun_sessions (session_id, tenant_id, project_id, backend_profile, conversation_id, thread_id, metadata, version, execution_state, last_run_id, last_command_id, active_run_id, active_command_id, last_event_seq, terminal_status, failure_kind, title, summary, last_activity_at, created_at, updated_at, expires_at)
|
||||
VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18::jsonb, $19, $20, $21, $22)
|
||||
ON CONFLICT (session_id) DO UPDATE SET
|
||||
tenant_id = EXCLUDED.tenant_id,
|
||||
project_id = EXCLUDED.project_id,
|
||||
@@ -782,15 +944,81 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
|
||||
conversation_id = EXCLUDED.conversation_id,
|
||||
thread_id = EXCLUDED.thread_id,
|
||||
metadata = EXCLUDED.metadata,
|
||||
version = EXCLUDED.version,
|
||||
execution_state = EXCLUDED.execution_state,
|
||||
last_run_id = EXCLUDED.last_run_id,
|
||||
last_command_id = EXCLUDED.last_command_id,
|
||||
active_run_id = EXCLUDED.active_run_id,
|
||||
active_command_id = EXCLUDED.active_command_id,
|
||||
last_event_seq = EXCLUDED.last_event_seq,
|
||||
terminal_status = EXCLUDED.terminal_status,
|
||||
failure_kind = EXCLUDED.failure_kind,
|
||||
title = EXCLUDED.title,
|
||||
summary = EXCLUDED.summary,
|
||||
last_activity_at = EXCLUDED.last_activity_at,
|
||||
updated_at = EXCLUDED.updated_at,
|
||||
expires_at = EXCLUDED.expires_at`,
|
||||
[record.sessionId, record.tenantId, record.projectId, record.backendProfile, record.conversationId, record.threadId, JSON.stringify(record.metadata), record.createdAt, record.updatedAt, record.expiresAt],
|
||||
[record.sessionId, record.tenantId, record.projectId, record.backendProfile, record.conversationId, record.threadId, JSON.stringify(record.metadata), record.version, record.executionState, record.lastRunId, record.lastCommandId, record.activeRunId, record.activeCommandId, record.lastEventSeq, record.terminalStatus, record.failureKind, record.title, JSON.stringify(record.summary), record.lastActivityAt, record.createdAt, record.updatedAt, record.expiresAt],
|
||||
);
|
||||
const nextSessionRef = sessionRefFromRecord(record, run.sessionRef);
|
||||
await client.query("UPDATE agentrun_runs SET session_ref = $2::jsonb, updated_at = $3 WHERE id = $1", [run.id, JSON.stringify(nextSessionRef), at]);
|
||||
await this.appendEventWithLockedRun(client, run.id, "backend_status", { phase: "session-updated", sessionRef: summarizeSessionRef(nextSessionRef), turnId });
|
||||
}
|
||||
|
||||
private async touchSessionForRun(client: PoolClient, run: RunRecord, patch: Partial<SessionRecord>, options: { bumpVersion: boolean; at?: string }): Promise<void> {
|
||||
const sessionId = run.sessionRef?.sessionId;
|
||||
if (!sessionId) return;
|
||||
const existingResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [sessionId]);
|
||||
const existing = existingResult.rows[0] ? sessionFromRow(existingResult.rows[0]) : null;
|
||||
if (!existing) return;
|
||||
const at = options.at ?? nowIso();
|
||||
const version = options.bumpVersion ? await this.nextSessionVersion(client) : existing.version;
|
||||
await client.query(
|
||||
`UPDATE agentrun_sessions SET
|
||||
version = $2,
|
||||
execution_state = $3,
|
||||
last_run_id = $4,
|
||||
last_command_id = $5,
|
||||
active_run_id = $6,
|
||||
active_command_id = $7,
|
||||
last_event_seq = $8,
|
||||
terminal_status = $9,
|
||||
failure_kind = $10,
|
||||
title = $11,
|
||||
summary = $12::jsonb,
|
||||
last_activity_at = $13,
|
||||
updated_at = $14
|
||||
WHERE session_id = $1`,
|
||||
[
|
||||
sessionId,
|
||||
version,
|
||||
patch.executionState ?? existing.executionState,
|
||||
patch.lastRunId === undefined ? existing.lastRunId : patch.lastRunId,
|
||||
patch.lastCommandId === undefined ? existing.lastCommandId : patch.lastCommandId,
|
||||
patch.activeRunId === undefined ? existing.activeRunId : patch.activeRunId,
|
||||
patch.activeCommandId === undefined ? existing.activeCommandId : patch.activeCommandId,
|
||||
patch.lastEventSeq ?? existing.lastEventSeq,
|
||||
patch.terminalStatus === undefined ? existing.terminalStatus : patch.terminalStatus,
|
||||
patch.failureKind === undefined ? existing.failureKind : patch.failureKind,
|
||||
patch.title === undefined ? existing.title : patch.title,
|
||||
JSON.stringify(patch.summary ?? existing.summary),
|
||||
patch.lastActivityAt ?? at,
|
||||
at,
|
||||
],
|
||||
);
|
||||
}
|
||||
|
||||
private async resolveSessionRunId(sessionId: string, requestedRunId: string | null): Promise<string | null> {
|
||||
const session = await this.getSession(sessionId);
|
||||
if (!session) throw new AgentRunError("schema-invalid", `session ${sessionId} was not found`, { httpStatus: 404 });
|
||||
if (requestedRunId) {
|
||||
const run = await this.getRun(requestedRunId);
|
||||
if (run.sessionRef?.sessionId !== sessionId) throw new AgentRunError("schema-invalid", `run ${requestedRunId} does not belong to session ${sessionId}`, { httpStatus: 404 });
|
||||
return requestedRunId;
|
||||
}
|
||||
return session.activeRunId ?? session.lastRunId;
|
||||
}
|
||||
|
||||
private async withTransaction<T>(fn: (client: PoolClient) => Promise<T>): Promise<T> {
|
||||
const client = await this.pool.connect();
|
||||
try {
|
||||
@@ -885,12 +1113,33 @@ function sessionFromRow(row: QueryResultRow): SessionRecord {
|
||||
conversationId: nullableString(row.conversation_id),
|
||||
threadId: nullableString(row.thread_id),
|
||||
metadata: jsonRecord(row.metadata),
|
||||
version: Number(row.version ?? 1),
|
||||
executionState: sessionExecutionState(row.execution_state),
|
||||
lastRunId: nullableString(row.last_run_id),
|
||||
lastCommandId: nullableString(row.last_command_id),
|
||||
activeRunId: nullableString(row.active_run_id),
|
||||
activeCommandId: nullableString(row.active_command_id),
|
||||
lastEventSeq: Number(row.last_event_seq ?? 0),
|
||||
terminalStatus: nullableString(row.terminal_status) as TerminalStatus | null,
|
||||
failureKind: nullableString(row.failure_kind) as FailureKind | null,
|
||||
title: nullableString(row.title),
|
||||
summary: jsonRecord(row.summary),
|
||||
lastActivityAt: nullableIso(row.last_activity_at),
|
||||
createdAt: iso(row.created_at),
|
||||
updatedAt: iso(row.updated_at),
|
||||
expiresAt: nullableIso(row.expires_at),
|
||||
};
|
||||
}
|
||||
|
||||
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 sessionExecutionState(value: unknown): SessionRecord["executionState"] {
|
||||
if (value === "running" || value === "terminal") return value;
|
||||
return "idle";
|
||||
}
|
||||
|
||||
function runnerJobFromRow(row: QueryResultRow): RunnerJobRecord {
|
||||
return {
|
||||
id: stringValue(row.id),
|
||||
|
||||
Reference in New Issue
Block a user