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

fix: 统一 Session 终态并发锁序

Browse Source
This commit is contained in:
Codex
2026-07-12 23:18:53 +02:00
parent 741e6bd713
commit 49cf4abaa1
2 changed files with 81 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mgr/postgres-store.ts
+9 -6
View File
@@ -615,18 +615,21 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
assertSessionSteerAdmissionContract(input);
return this.withTransaction(async (client) => {
await client.query("SELECT pg_advisory_xact_lock(hashtext($1), hashtext($2))", ["agentrun-session-turn-admission", input.sessionId]);
const sessionResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionId]);
const session = sessionResult.rows[0] ? sessionFromRow(sessionResult.rows[0]) : null;
if (!session?.activeRunId || !session.activeCommandId) return null;
const runResult = await client.query("SELECT * FROM agentrun_runs WHERE id = $1 FOR UPDATE", [session.activeRunId]);
const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [session.activeCommandId]);
const observedSessionResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1", [input.sessionId]);
const observedSession = observedSessionResult.rows[0] ? sessionFromRow(observedSessionResult.rows[0]) : null;
if (!observedSession?.activeRunId || !observedSession.activeCommandId) return null;
const runResult = await client.query("SELECT * FROM agentrun_runs WHERE id = $1 FOR UPDATE", [observedSession.activeRunId]);
const commandResult = await client.query("SELECT * FROM agentrun_commands WHERE id = $1 FOR UPDATE", [observedSession.activeCommandId]);
if (!runResult.rows[0] || !commandResult.rows[0]) return null;
const run = runFromRow(runResult.rows[0]);
const targetCommand = commandFromRow(commandResult.rows[0]);
const sessionResult = await client.query("SELECT * FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [input.sessionId]);
const session = sessionResult.rows[0] ? sessionFromRow(sessionResult.rows[0]) : null;
if (!session) return null;
const receivable = receivableActiveSessionTurn(session, run, targetCommand);
if (!receivable) return null;
const command = await this.createCommandWithClient(client, run.id, input.command);
return { run: await this.requireRunForUpdate(client, run.id), command, targetCommand, ...receivable };
return { run, command, targetCommand, ...receivable };
});
}
src/selftest/integration/postgres-session-turn-admission.ts
+72
View File
@@ -13,6 +13,7 @@ const concurrentSessionId = `ses_pg_admission_${suffix}`;
const legacySessionId = `ses_pg_admission_legacy_${suffix}`;
const rollbackSessionId = `ses_pg_admission_rollback_${suffix}`;
const resourceSessionId = `ses_pg_admission_resource_${suffix}`;
const terminalRaceSessionId = `ses_pg_admission_terminal_race_${suffix}`;
const eventOutbox = { enabled: true, topic: "agentrun.event.v1", source: "session-admission-selftest" } as const;
const primary = await createPostgresAgentRunStore({ connectionString, poolMax: 4, eventOutbox });
const concurrent = await createPostgresAgentRunStore({ connectionString, poolMax: 4, eventOutbox });
@@ -24,6 +25,7 @@ try {
await assertLegacyHalfCommitRecoveryAfterRestart();
await assertAdmissionTransactionRollback();
await assertLatestSessionResourceRunSkipsJsonNull();
await assertTerminalAndFollowUpLockOrder();
console.log(JSON.stringify({
ok: true,
tests: [
@@ -32,6 +34,7 @@ try {
"postgres-session-turn-active-conflict",
"postgres-session-turn-same-session-continuity",
"postgres-session-follow-up-terminal-race-preserves-new-active-run",
"postgres-session-terminal-follow-up-lock-order",
"postgres-legacy-half-commit-recovery-after-restart",
"postgres-session-turn-transaction-rollback",
"postgres-session-resource-run-skips-json-null",
@@ -44,6 +47,7 @@ try {
await cleanupSession(legacySessionId).catch(() => undefined);
await cleanupSession(rollbackSessionId).catch(() => undefined);
await cleanupSession(resourceSessionId).catch(() => undefined);
await cleanupSession(terminalRaceSessionId).catch(() => undefined);
await Promise.all(reopenedStores.map(async (store) => await store.close()));
await Promise.all([primary.close(), concurrent.close(), control.end()]);
}
@@ -157,6 +161,74 @@ async function assertLatestSessionResourceRunSkipsJsonNull(): Promise<void> {
assert.equal((await primary.getLatestSessionResourceRun(resourceSessionId))?.id, initial.id);
}
async function assertTerminalAndFollowUpLockOrder(): Promise<void> {
const run = await primary.createRun(validateCreateRun(runInput(terminalRaceSessionId)));
const command = await primary.createCommand(run.id, validateCreateCommand({ type: "turn", payload: { prompt: "terminal lock order" }, idempotencyKey: `pg-terminal-lock-order-${suffix}` }));
const runner = await primary.registerRunner({ runId: run.id, attemptId: `attempt_terminal_lock_${suffix}`, backendProfile: "codex", placement: "selftest", sourceCommit: "selftest" });
await primary.claimRun(run.id, runner.id, 60_000);
await primary.ackCommand(command.id);
const terminalClient = await control.connect();
let steerPromise: Promise<Awaited<ReturnType<typeof concurrent.admitSessionSteer>>> | null = null;
try {
await terminalClient.query("BEGIN");
await terminalClient.query("SET LOCAL lock_timeout = '2s'");
await terminalClient.query("SELECT id FROM agentrun_runs WHERE id = $1 FOR UPDATE", [run.id]);
await terminalClient.query("SELECT id FROM agentrun_commands WHERE id = $1 FOR UPDATE", [command.id]);
steerPromise = concurrent.admitSessionSteer({
sessionId: terminalRaceSessionId,
command: validateCreateCommand({ type: "steer", payload: { prompt: "follow up at terminal boundary" }, idempotencyKey: `pg-terminal-lock-steer-${suffix}` }),
});
await waitForBlockedRunLock();
await terminalClient.query("SELECT session_id FROM agentrun_sessions WHERE session_id = $1 FOR UPDATE", [terminalRaceSessionId]);
await terminalClient.query("UPDATE agentrun_commands SET state = 'completed', updated_at = now() WHERE id = $1", [command.id]);
await terminalClient.query(
`UPDATE agentrun_sessions
SET execution_state = 'terminal', active_run_id = NULL, active_command_id = NULL,
last_run_id = $2, last_command_id = $3, terminal_status = 'completed', failure_kind = NULL, updated_at = now()
WHERE session_id = $1`,
[terminalRaceSessionId, run.id, command.id],
);
await terminalClient.query("COMMIT");
assert.equal(await steerPromise, null);
assert.deepEqual((await primary.listCommands(run.id, 0, 100)).map((item) => item.id), [command.id]);
const next = await primary.admitSessionTurn(admissionInput(terminalRaceSessionId, "new run after terminal boundary", `pg-terminal-lock-next-${suffix}`));
await concurrent.finishRun(run.id, { terminalStatus: "completed", failureKind: null, failureMessage: null });
const session = await primary.getSession(terminalRaceSessionId);
assert.equal(session?.activeRunId, next.run.id);
assert.equal(session?.activeCommandId, next.command.id);
assert.equal(session?.lastRunId, next.run.id);
assert.equal(session?.lastCommandId, next.command.id);
} finally {
await terminalClient.query("ROLLBACK").catch(() => undefined);
terminalClient.release();
if (steerPromise) await steerPromise.catch(() => undefined);
}
}
async function waitForBlockedRunLock(): Promise<void> {
const deadline = Date.now() + 2_000;
while (Date.now() < deadline) {
const result = await control.query<{ blocked: number }>(
`SELECT count(*)::int AS blocked
FROM pg_stat_activity
WHERE application_name = 'agentrun-mgr-v01'
AND wait_event_type = 'Lock'
AND query LIKE 'SELECT * FROM agentrun_runs WHERE id = $1 FOR UPDATE%'`,
);
if ((result.rows[0]?.blocked ?? 0) > 0) return;
await delay(10);
}
throw new Error("session steer did not block on the run lock before the terminal path acquired the session lock");
}
async function delay(ms: number): Promise<void> {
await new Promise<void>((resolve) => setTimeout(resolve, ms));
}
function admissionInput(sessionId: string, prompt: string, idempotencyKey: string): SessionTurnAdmissionInput {
return {
sessionId,