From 49cf4abaa1bfc0c775981a31a4ac76cc5dd0fb7f Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 12 Jul 2026 23:18:53 +0200 Subject: [PATCH] =?UTF-8?q?fix:=20=E7=BB=9F=E4=B8=80=20Session=20=E7=BB=88?= =?UTF-8?q?=E6=80=81=E5=B9=B6=E5=8F=91=E9=94=81=E5=BA=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mgr/postgres-store.ts | 15 ++-- .../postgres-session-turn-admission.ts | 72 +++++++++++++++++++ 2 files changed, 81 insertions(+), 6 deletions(-) diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 1e5f395..4ddaff6 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -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 }; }); } diff --git a/src/selftest/integration/postgres-session-turn-admission.ts b/src/selftest/integration/postgres-session-turn-admission.ts index 2d87218..c72ee2e 100644 --- a/src/selftest/integration/postgres-session-turn-admission.ts +++ b/src/selftest/integration/postgres-session-turn-admission.ts @@ -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 { assert.equal((await primary.getLatestSessionResourceRun(resourceSessionId))?.id, initial.id); } +async function assertTerminalAndFollowUpLockOrder(): Promise { + 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>> | 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 { + 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 { + await new Promise((resolve) => setTimeout(resolve, ms)); +} + function admissionInput(sessionId: string, prompt: string, idempotencyKey: string): SessionTurnAdmissionInput { return { sessionId,