diff --git a/package.json b/package.json index 9d2bcb5..671568c 100644 --- a/package.json +++ b/package.json @@ -7,6 +7,7 @@ "check": "tsc --noEmit", "self-test": "bun run src/selftest/run.ts", "self-test:postgres-kafka-order": "bun run src/selftest/integration/postgres-kafka-order.ts", + "self-test:postgres-events-long-poll": "bun run src/selftest/integration/postgres-events-long-poll.ts", "self-test:postgres-queue-retry": "bun run src/selftest/integration/postgres-queue-retry.ts", "self-test:postgres-runner-retention-fence": "bun run src/selftest/integration/postgres-runner-retention-fence.ts", "self-test:postgres-session-turn-admission": "bun run src/selftest/integration/postgres-session-turn-admission.ts", diff --git a/src/mgr/postgres-store.ts b/src/mgr/postgres-store.ts index 41dd02a..62c5aa1 100644 --- a/src/mgr/postgres-store.ts +++ b/src/mgr/postgres-store.ts @@ -18,6 +18,7 @@ interface PostgresStoreOptions { connectionString: string; poolMax?: number; eventOutbox?: EventOutboxConfig; + onWaitForRunChangeListening?: (runId: string) => Promise; } interface MigrationDefinition { @@ -123,6 +124,8 @@ ON CONFLICT (profile) DO UPDATE SET updated_at = EXCLUDED.updated_at; `; +const runChangeNotificationChannel = "agentrun_run_change"; + const backendProfilesMigrationSql = ` INSERT INTO agentrun_backends (profile, capabilities, capacity, health, updated_at) VALUES ${backendCapabilitiesSqlValues(["codex", "deepseek"])} @@ -539,6 +542,7 @@ export class PostgresAgentRunStore implements AgentRunStore { private readonly pool: Pool; private readonly runnerCreateFencePool: Pool; private readonly eventOutboxConfig: EventOutboxConfig; + private readonly onWaitForRunChangeListening: ((runId: string) => Promise) | null; private migrationReady = false; private appliedMigrationId: string | null = null; @@ -546,6 +550,7 @@ export class PostgresAgentRunStore implements AgentRunStore { this.pool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-v01", ...(options.poolMax === undefined ? {} : { max: options.poolMax }) }); this.runnerCreateFencePool = new Pool({ connectionString: options.connectionString, application_name: "agentrun-mgr-runner-capacity-fence", max: Math.max(2, options.poolMax ?? 10) }); this.eventOutboxConfig = options.eventOutbox ?? { enabled: false, topic: null, source: null }; + this.onWaitForRunChangeListening = options.onWaitForRunChangeListening ?? null; } async migrate(): Promise { @@ -601,6 +606,54 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations ( return result.rows.map(eventFromRow); } + async countEvents(runId: string, afterSeq: number): Promise { + await this.getRun(runId); + const result = await this.pool.query<{ count: string }>("SELECT COUNT(*)::text AS count FROM agentrun_events WHERE run_id = $1 AND seq > $2", [runId, afterSeq]); + return Number(result.rows[0]?.count ?? 0); + } + + async waitForRunChange(runId: string, afterSeq: number, timeoutMs: number, signal?: AbortSignal): Promise<"changed" | "timeout" | "aborted"> { + await this.getRun(runId); + if (signal?.aborted) return "aborted"; + const client = await this.pool.connect(); + let notificationHandler: ((message: import("pg").Notification) => void) | null = null; + let abortHandler: (() => void) | null = null; + let timer: ReturnType | null = null; + try { + await client.query(`LISTEN ${runChangeNotificationChannel}`); + let finish: (result: "changed" | "timeout" | "aborted") => void = () => {}; + const wait = new Promise<"changed" | "timeout" | "aborted">((resolve) => { + let settled = false; + finish = (result) => { + if (settled) return; + settled = true; + resolve(result); + }; + notificationHandler = (message) => { + if (message.channel === runChangeNotificationChannel && message.payload === runId) finish("changed"); + }; + abortHandler = () => finish("aborted"); + timer = setTimeout(() => finish("timeout"), timeoutMs); + client.on("notification", notificationHandler); + signal?.addEventListener("abort", abortHandler, { once: true }); + if (signal?.aborted) finish("aborted"); + }); + await this.onWaitForRunChangeListening?.(runId); + const changed = await client.query("SELECT EXISTS(SELECT 1 FROM agentrun_events WHERE run_id = $1 AND seq > $2) AS changed", [runId, afterSeq]); + if (changed.rows[0]?.changed === true) finish("changed"); + return await wait; + } finally { + if (timer !== null) clearTimeout(timer); + if (abortHandler !== null) signal?.removeEventListener("abort", abortHandler); + if (notificationHandler !== null) client.removeListener("notification", notificationHandler); + try { + await client.query(`UNLISTEN ${runChangeNotificationChannel}`); + } finally { + client.release(); + } + } + } + async listEventsForCommand(runId: string, commandId: string, limit: number): Promise { await this.getRun(runId); const result = await this.pool.query( @@ -1801,6 +1854,7 @@ 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: fenced.type, 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("SELECT pg_notify($1, $2)", [runChangeNotificationChannel, runId]); await client.query( `UPDATE agentrun_sessions SET last_event_seq = $2, last_activity_at = $3, updated_at = $3 diff --git a/src/mgr/server.ts b/src/mgr/server.ts index 89c1802..1e9f9b0 100644 --- a/src/mgr/server.ts +++ b/src/mgr/server.ts @@ -2,7 +2,7 @@ import type { Server } from "node:http"; import { createServer } from "node:http"; import type { AddressInfo } from "node:net"; import type { AgentRunStore, ListQueueTasksInput, ListSessionsInput, SessionEventPageInput } from "./store.js"; -import { assertSessionBoundary, openAgentRunStoreFromEnv, receivableActiveSessionTurn } from "./store.js"; +import { assertSessionBoundary, isTerminalCommandState, isTerminalRunStatus, openAgentRunStoreFromEnv, receivableActiveSessionTurn } from "./store.js"; import { agentRunKafkaConfig, type AgentRunKafkaConfig } from "../common/kafka-events.js"; import { AgentRunError, errorToJson } from "../common/errors.js"; import { asRecord, stableHash, validateBackendProfile, validateCreateCommand, validateCreateQueueTask, validateCreateRun, validateQueueTaskState, validateSessionListState, validateSessionRunnerJobInput } from "../common/validation.js"; @@ -275,6 +275,11 @@ export async function startManagerServer(options: ManagerServerOptions = {}): Pr const method = req.method ?? "GET"; const url = new URL(req.url ?? "/", "http://agentrun.local"); let diagnosticTraceContext: AgentRunDiagnosticOtelTraceContext | null = null; + const requestAbort = new AbortController(); + req.once("aborted", () => requestAbort.abort()); + res.once("close", () => { + if (!res.writableEnded) requestAbort.abort(); + }); try { assertManagerRequestAuthorized(req, url.pathname, auth); diagnosticTraceContext = agentRunDiagnosticOtelTraceContext(req.headers); @@ -284,7 +289,7 @@ export async function startManagerServer(options: ManagerServerOptions = {}): Pr if (diagnosticTraceContext !== null && !isSupportedDiagnosticReadPath(url.pathname)) { throw new AgentRunError("schema-invalid", "diagnostic trace context is not supported for this GET route", { httpStatus: 400, details: { reason: "diagnostic-trace-route-unsupported", route: url.pathname, valuesPrinted: false } }); } - const data = await route({ method, url, body: await readBody(req), store, sourceCommit, authSummary, diagnosticTraceContext, runnerReconcilerOptions, runnerDispatcherOptions, kafkaOutboxRelayOptions, resolvedRunnerJobDefaults, ...(runnerJobDefaults ? { runnerJobDefaults } : {}), ...(sessionPvcDefaults ? { sessionPvcDefaults } : {}), ...(providerProfileDefaults ? { providerProfileDefaults } : {}), ...(toolCredentialDefaults ? { toolCredentialDefaults } : {}), ...(aipodSpecDir ? { aipodSpecDir } : {}) }); + const data = await route({ method, url, body: await readBody(req), signal: requestAbort.signal, store, sourceCommit, authSummary, diagnosticTraceContext, runnerReconcilerOptions, runnerDispatcherOptions, kafkaOutboxRelayOptions, resolvedRunnerJobDefaults, ...(runnerJobDefaults ? { runnerJobDefaults } : {}), ...(sessionPvcDefaults ? { sessionPvcDefaults } : {}), ...(providerProfileDefaults ? { providerProfileDefaults } : {}), ...(toolCredentialDefaults ? { toolCredentialDefaults } : {}), ...(aipodSpecDir ? { aipodSpecDir } : {}) }); writeJson(res, 200, { ok: true, data, traceId }); } catch (error) { const agentError = normalizeError(error); @@ -641,7 +646,7 @@ function commandResultOtelAttributes(result: JsonValue): JsonRecord { }; } -async function route({ method, url, body, store, sourceCommit, authSummary, diagnosticTraceContext, runnerJobDefaults, sessionPvcDefaults, providerProfileDefaults, toolCredentialDefaults, runnerReconcilerOptions, runnerDispatcherOptions, kafkaOutboxRelayOptions, resolvedRunnerJobDefaults, aipodSpecDir }: { method: string; url: URL; body: unknown; store: AgentRunStore; sourceCommit: string; authSummary?: JsonRecord; diagnosticTraceContext: AgentRunDiagnosticOtelTraceContext | null; runnerJobDefaults?: NonNullable; sessionPvcDefaults?: NonNullable; providerProfileDefaults?: NonNullable; toolCredentialDefaults?: NonNullable; runnerReconcilerOptions: RunnerReconcilerRuntimeOptions; runnerDispatcherOptions: RunnerDispatcherOptions; kafkaOutboxRelayOptions: KafkaOutboxRelayOptions; resolvedRunnerJobDefaults: RunnerJobDefaults; aipodSpecDir?: string }): Promise { +async function route({ method, url, body, signal, store, sourceCommit, authSummary, diagnosticTraceContext, runnerJobDefaults, sessionPvcDefaults, providerProfileDefaults, toolCredentialDefaults, runnerReconcilerOptions, runnerDispatcherOptions, kafkaOutboxRelayOptions, resolvedRunnerJobDefaults, aipodSpecDir }: { method: string; url: URL; body: unknown; signal: AbortSignal; store: AgentRunStore; sourceCommit: string; authSummary?: JsonRecord; diagnosticTraceContext: AgentRunDiagnosticOtelTraceContext | null; runnerJobDefaults?: NonNullable; sessionPvcDefaults?: NonNullable; providerProfileDefaults?: NonNullable; toolCredentialDefaults?: NonNullable; runnerReconcilerOptions: RunnerReconcilerRuntimeOptions; runnerDispatcherOptions: RunnerDispatcherOptions; kafkaOutboxRelayOptions: KafkaOutboxRelayOptions; resolvedRunnerJobDefaults: RunnerJobDefaults; aipodSpecDir?: string }): Promise { const path = url.pathname; if (method === "GET" && (path === "/health" || path === "/health/live" || path === "/health/readiness")) { const [database, runnerDispatch, kafkaEventOutbox] = await Promise.all([store.health(), store.runnerDispatchStatus(), store.kafkaEventOutboxStatus()]); @@ -927,9 +932,18 @@ async function route({ method, url, body, store, sourceCommit, authSummary, diag const limit = integerQuery(url, "limit", 100); const run = await store.getRun(runId); const traceContext = await validatedDiagnosticTraceContext({ store, context: diagnosticTraceContext, run }); - const items = await store.listEvents(runId, afterSeq, limit); - void emitAgentRunOtelSpan("projection_sync", run, process.env, { ...(traceContext ? { traceContext } : {}), startTimeMs: startedAt, kind: 2, attributes: { "http.method": "GET", "http.route": "/api/v1/runs/:runId/events", "http.status_code": 200, afterSeq, limit, eventCount: items.length } }); - return { items: items as unknown as JsonValue }; + const expectRaw = url.searchParams.get("expect"); + if (expectRaw === null) { + if (url.searchParams.has("timeoutMs")) throw new AgentRunError("schema-invalid", "timeoutMs requires expect", { httpStatus: 400 }); + const items = await store.listEvents(runId, afterSeq, limit); + void emitAgentRunOtelSpan("projection_sync", run, process.env, { ...(traceContext ? { traceContext } : {}), startTimeMs: startedAt, kind: 2, attributes: { "http.method": "GET", "http.route": "/api/v1/runs/:runId/events", "http.status_code": 200, afterSeq, limit, eventCount: items.length } }); + return { items: items as unknown as JsonValue }; + } + const expect = positiveIntegerQuery(url, "expect"); + const timeoutMs = positiveIntegerQuery(url, "timeoutMs", 120_000, 3_600_000); + const page = await waitForEventsPage({ store, runId, afterSeq, expect, timeoutMs, limit, signal }); + void emitAgentRunOtelSpan("projection_sync", run, process.env, { ...(traceContext ? { traceContext } : {}), startTimeMs: startedAt, kind: 2, attributes: { "http.method": "GET", "http.route": "/api/v1/runs/:runId/events", "http.status_code": 200, afterSeq, expect, timeoutMs, limit, eventCount: page.items.length, observed: page.observed, timedOut: page.timedOut, completionReason: page.completionReason } }); + return page as unknown as JsonValue; } const runResultMatch = path.match(/^\/api\/v1\/runs\/([^/]+)\/result$/u); if (method === "GET" && runResultMatch) { @@ -1451,6 +1465,52 @@ function integerQuery(url: URL, key: string, fallback: number): number { return Number.isInteger(value) && value >= 0 ? value : fallback; } +function positiveIntegerQuery(url: URL, key: string, fallback?: number, maximum = Number.MAX_SAFE_INTEGER): number { + const raw = url.searchParams.get(key); + if (raw === null && fallback !== undefined) return fallback; + if (raw === null || !/^\d+$/u.test(raw)) throw new AgentRunError("schema-invalid", `${key} must be a positive integer`, { httpStatus: 400 }); + const value = Number(raw); + if (!Number.isSafeInteger(value) || value < 1 || value > maximum) { + throw new AgentRunError("schema-invalid", `${key} must be between 1 and ${maximum}`, { httpStatus: 400 }); + } + return value; +} + +async function waitForEventsPage(input: { store: AgentRunStore; runId: string; afterSeq: number; expect: number; timeoutMs: number; limit: number; signal: AbortSignal }): Promise<{ items: RunEvent[]; timedOut: boolean; observed: number; nextAfterSeq: number; hasMore: boolean; completionReason: "expect-reached" | "run-terminal" | "command-terminal" | "timeout" }> { + const deadline = Date.now() + input.timeoutMs; + while (true) { + const [run, observed, commands] = await Promise.all([ + input.store.getRun(input.runId), + input.store.countEvents(input.runId, input.afterSeq), + input.store.listCommands(input.runId, 0, 500), + ]); + const latestCommand = commands.at(-1) ?? null; + let completionReason: "expect-reached" | "run-terminal" | "command-terminal" | "timeout" | null = null; + let timedOut = false; + if (observed >= input.expect) completionReason = "expect-reached"; + else if (isTerminalRunStatus(run.status)) completionReason = "run-terminal"; + else if (latestCommand !== null && isTerminalCommandState(latestCommand.state)) completionReason = "command-terminal"; + else if (Date.now() >= deadline) { + completionReason = "timeout"; + timedOut = true; + } + if (completionReason !== null) { + const items = await input.store.listEvents(input.runId, input.afterSeq, input.limit); + return { + items, + timedOut, + observed, + nextAfterSeq: items.at(-1)?.seq ?? input.afterSeq, + hasMore: observed > items.length, + completionReason, + }; + } + const remainingMs = Math.max(1, deadline - Date.now()); + const waitResult = await input.store.waitForRunChange(input.runId, input.afterSeq + observed, remainingMs, input.signal); + if (waitResult === "aborted") throw new AgentRunError("cancelled", "events long poll request disconnected", { httpStatus: 499 }); + } +} + function numberField(record: JsonRecord, key: string, fallback: number): number { const value = record[key]; return typeof value === "number" && Number.isFinite(value) ? value : fallback; diff --git a/src/mgr/store.ts b/src/mgr/store.ts index 596a2b8..2ee6790 100644 --- a/src/mgr/store.ts +++ b/src/mgr/store.ts @@ -79,6 +79,8 @@ export interface AgentRunStore { createRun(input: CreateRunInput, identity?: CreateRunIdentity): MaybePromise; getRun(runId: string): MaybePromise; listEvents(runId: string, afterSeq: number, limit: number): MaybePromise; + countEvents(runId: string, afterSeq: number): MaybePromise; + waitForRunChange(runId: string, afterSeq: number, timeoutMs: number, signal?: AbortSignal): Promise<"changed" | "timeout" | "aborted">; listEventsForCommand(runId: string, commandId: string, limit: number): MaybePromise; createCommand(runId: string, input: CreateCommandInput): MaybePromise; admitSessionSteer(input: SessionSteerAdmissionInput): MaybePromise; @@ -262,6 +264,7 @@ export class MemoryAgentRunStore implements AgentRunStore { private readonly queueReadCursors = new Map(); private readonly runnerCreateFenceTails = new Map>(); private readonly runnerRetentionFences = new Set(); + private readonly runChangeWaiters = new Map void>>(); private queueVersion = 0; private sessionVersion = 0; private kafkaOutboxSeq = 0; @@ -302,6 +305,38 @@ export class MemoryAgentRunStore implements AgentRunStore { return (this.eventsByRun.get(runId) ?? []).filter((event) => event.seq > afterSeq).slice(0, Math.max(1, Math.min(limit, 500))); } + countEvents(runId: string, afterSeq: number): number { + this.getRun(runId); + return (this.eventsByRun.get(runId) ?? []).filter((event) => event.seq > afterSeq).length; + } + + async waitForRunChange(runId: string, afterSeq: number, timeoutMs: number, signal?: AbortSignal): Promise<"changed" | "timeout" | "aborted"> { + this.getRun(runId); + if (this.countEvents(runId, afterSeq) > 0) return "changed"; + if (signal?.aborted) return "aborted"; + return await new Promise((resolve) => { + const waiters = this.runChangeWaiters.get(runId) ?? new Set<() => void>(); + let settled = false; + const finish = (result: "changed" | "timeout" | "aborted") => { + if (settled) return; + settled = true; + clearTimeout(timer); + waiters.delete(onChange); + if (waiters.size === 0) this.runChangeWaiters.delete(runId); + signal?.removeEventListener("abort", onAbort); + resolve(result); + }; + const onChange = () => finish("changed"); + const onAbort = () => finish("aborted"); + const timer = setTimeout(() => finish("timeout"), timeoutMs); + waiters.add(onChange); + this.runChangeWaiters.set(runId, waiters); + signal?.addEventListener("abort", onAbort, { once: true }); + if (signal?.aborted) finish("aborted"); + if (this.countEvents(runId, afterSeq) > 0) finish("changed"); + }); + } + listEventsForCommand(runId: string, commandId: string, limit: number): RunEvent[] { this.getRun(runId); const clamped = Math.max(1, Math.min(limit, 2_000)); @@ -802,6 +837,7 @@ export class MemoryAgentRunStore implements AgentRunStore { this.kafkaEventOutbox.set(outbox.id, outbox); } this.touchSessionForRun(this.getRun(run.id), { lastEventSeq: event.seq, lastActivityAt: event.createdAt }, { bumpVersion: false, at: event.createdAt }); + for (const notify of this.runChangeWaiters.get(run.id) ?? []) notify(); } finishRun(runId: string, result: Pick): RunRecord { diff --git a/src/selftest/cases/10-manager-memory.ts b/src/selftest/cases/10-manager-memory.ts index 0334ec9..da8c2ff 100644 --- a/src/selftest/cases/10-manager-memory.ts +++ b/src/selftest/cases/10-manager-memory.ts @@ -19,8 +19,9 @@ const selfTest: SelfTestCase = async () => { assert.ok((((health.runnerWorkReady as { requiredImageTools?: string[] } | undefined)?.requiredImageTools) ?? []).includes("npm")); assert.equal(health.secretRefs?.valuesPrinted, false); await assertManagerAuthBoundary(); + await assertEventsLongPolling(client, store); await assertLongResultUsesTerminalAssistant(client, store); - return { name: "manager-memory", tests: ["manager-memory-lifecycle", "manager-auth-boundary", "manager-result-long-trace"] }; + return { name: "manager-memory", tests: ["manager-memory-lifecycle", "manager-auth-boundary", "manager-events-long-poll", "manager-result-long-trace"] }; } finally { await new Promise((resolve) => server.server.close(() => resolve())); } @@ -48,6 +49,75 @@ async function assertManagerAuthBoundary(): Promise { } } +async function assertEventsLongPolling(client: ManagerClient, store: MemoryAgentRunStore): Promise { + const run = createLongPollRun(store, "memory-expect-limit"); + const afterSeq = store.listEvents(run.id, 0, 500).at(-1)?.seq ?? 0; + const expected = client.get(`/api/v1/runs/${encodeURIComponent(run.id)}/events?afterSeq=${afterSeq}&expect=3&timeoutMs=1000&limit=1`) as Promise; + setTimeout(() => { + store.appendEvent(run.id, "backend_status", { phase: "first" }); + store.appendEvent(run.id, "backend_status", { phase: "second" }); + store.appendEvent(run.id, "backend_status", { phase: "third" }); + }, 10); + const expectedPage = await expected; + assert.equal(expectedPage.timedOut, false); + assert.equal(expectedPage.observed, 3); + assert.equal(expectedPage.hasMore, true); + assert.equal((expectedPage.items as JsonRecord[]).length, 1); + assert.equal(expectedPage.completionReason, "expect-reached"); + assert.equal(expectedPage.nextAfterSeq, afterSeq + 1); + + const partialAfterSeq = store.listEvents(run.id, 0, 500).at(-1)?.seq ?? afterSeq; + const partial = client.get(`/api/v1/runs/${encodeURIComponent(run.id)}/events?afterSeq=${partialAfterSeq}&expect=2&timeoutMs=40&limit=20`) as Promise; + setTimeout(() => store.appendEvent(run.id, "backend_status", { phase: "partial" }), 10); + const partialPage = await partial; + assert.equal(partialPage.timedOut, true); + assert.equal(partialPage.observed, 1); + assert.equal((partialPage.items as JsonRecord[]).length, 1); + assert.equal(partialPage.completionReason, "timeout"); + + const emptyAfterSeq = store.listEvents(run.id, 0, 500).at(-1)?.seq ?? partialAfterSeq; + const emptyPage = await client.get(`/api/v1/runs/${encodeURIComponent(run.id)}/events?afterSeq=${emptyAfterSeq}&expect=1&timeoutMs=10&limit=20`) as JsonRecord; + assert.equal(emptyPage.timedOut, true); + assert.equal(emptyPage.observed, 0); + assert.equal(emptyPage.nextAfterSeq, emptyAfterSeq); + + const commandRun = createLongPollRun(store, "memory-command-terminal"); + const command = store.createCommand(commandRun.id, { type: "turn", payload: { prompt: "terminal wake" }, idempotencyKey: "memory-command-terminal" }); + const commandAfterSeq = store.listEvents(commandRun.id, 0, 500).at(-1)?.seq ?? 0; + const commandWait = client.get(`/api/v1/runs/${encodeURIComponent(commandRun.id)}/events?afterSeq=${commandAfterSeq}&expect=10&timeoutMs=1000&limit=20`) as Promise; + setTimeout(() => store.finishCommand(command.id, { terminalStatus: "completed", failureKind: null, failureMessage: null }), 10); + const commandPage = await commandWait; + assert.equal(commandPage.timedOut, false); + assert.equal(commandPage.completionReason, "command-terminal"); + + const runTerminal = createLongPollRun(store, "memory-run-terminal"); + const runAfterSeq = store.listEvents(runTerminal.id, 0, 500).at(-1)?.seq ?? 0; + const runWait = client.get(`/api/v1/runs/${encodeURIComponent(runTerminal.id)}/events?afterSeq=${runAfterSeq}&expect=10&timeoutMs=1000&limit=20`) as Promise; + setTimeout(() => store.finishRun(runTerminal.id, { terminalStatus: "completed", failureKind: null, failureMessage: null }), 10); + const runPage = await runWait; + assert.equal(runPage.timedOut, false); + assert.equal(runPage.completionReason, "run-terminal"); +} + +function createLongPollRun(store: MemoryAgentRunStore, idempotencyKey: string) { + return store.createRun({ + tenantId: "unidesk", + projectId: "pikasTech/agentrun", + workspaceRef: { kind: "host-path", path: "/tmp/agentrun-selftest" }, + providerId: "NC01", + backendProfile: "codex", + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 1000, + network: "none", + secretScope: { allowCredentialEcho: false, providerCredentials: [] }, + }, + traceSink: null, + idempotencyKey, + }); +} + async function assertLongResultUsesTerminalAssistant(client: ManagerClient, store: MemoryAgentRunStore): Promise { const run = store.createRun({ tenantId: "unidesk", diff --git a/src/selftest/integration/postgres-events-long-poll.ts b/src/selftest/integration/postgres-events-long-poll.ts new file mode 100644 index 0000000..8aa3ab2 --- /dev/null +++ b/src/selftest/integration/postgres-events-long-poll.ts @@ -0,0 +1,79 @@ +import assert from "node:assert/strict"; +import { ManagerClient } from "../../mgr/client.js"; +import { createPostgresAgentRunStore } from "../../mgr/postgres-store.js"; +import { startManagerServer } from "../../mgr/server.js"; +import type { CreateRunInput, JsonRecord, RunRecord } from "../../common/types.js"; + +const connectionString = process.env.AGENTRUN_SELFTEST_DATABASE_URL?.trim(); +if (!connectionString) throw new Error("AGENTRUN_SELFTEST_DATABASE_URL is required; this integration test never falls back to memory"); + +const concurrent = await createPostgresAgentRunStore({ connectionString, poolMax: 4 }); +let raceRun: RunRecord | null = null; +const primary = await createPostgresAgentRunStore({ + connectionString, + poolMax: 6, + onWaitForRunChangeListening: async (runId) => { + if (raceRun?.id === runId) await concurrent.appendEvent(runId, "backend_status", { phase: "listen-query-race" }); + }, +}); +const server = await startManagerServer({ port: 0, host: "127.0.0.1", sourceCommit: "self-test", store: primary }); + +try { + raceRun = await primary.createRun(createRunInput("postgres-listen-query-race")); + const raceAfterSeq = (await primary.listEvents(raceRun.id, 0, 500)).at(-1)?.seq ?? 0; + const raceResult = await primary.waitForRunChange(raceRun.id, raceAfterSeq, 1_000); + assert.equal(raceResult, "changed"); + assert.equal(await primary.countEvents(raceRun.id, raceAfterSeq), 1); + + const client = new ManagerClient(server.baseUrl); + const run = await primary.createRun(createRunInput("postgres-expect-limit")); + const afterSeq = (await primary.listEvents(run.id, 0, 500)).at(-1)?.seq ?? 0; + const expected = client.get(`/api/v1/runs/${encodeURIComponent(run.id)}/events?afterSeq=${afterSeq}&expect=3&timeoutMs=2000&limit=1`) as Promise; + setTimeout(() => { + void Promise.all([ + concurrent.appendEvent(run.id, "backend_status", { phase: "first" }), + concurrent.appendEvent(run.id, "backend_status", { phase: "second" }), + concurrent.appendEvent(run.id, "backend_status", { phase: "third" }), + ]); + }, 20); + const page = await expected; + assert.equal(page.timedOut, false); + assert.equal(page.observed, 3); + assert.equal(page.hasMore, true); + assert.equal((page.items as JsonRecord[]).length, 1); + assert.equal(page.completionReason, "expect-reached"); + + const terminalRun = await primary.createRun(createRunInput("postgres-command-terminal")); + const command = await primary.createCommand(terminalRun.id, { type: "turn", payload: { prompt: "terminal wake" }, idempotencyKey: "postgres-command-terminal" }); + const terminalAfterSeq = (await primary.listEvents(terminalRun.id, 0, 500)).at(-1)?.seq ?? 0; + const terminalWait = client.get(`/api/v1/runs/${encodeURIComponent(terminalRun.id)}/events?afterSeq=${terminalAfterSeq}&expect=10&timeoutMs=2000&limit=20`) as Promise; + setTimeout(() => { + void concurrent.finishCommand(command.id, { terminalStatus: "completed", failureKind: null, failureMessage: null }); + }, 20); + const terminalPage = await terminalWait; + assert.equal(terminalPage.timedOut, false); + assert.equal(terminalPage.completionReason, "command-terminal"); + + console.log(JSON.stringify({ ok: true, test: "postgres-events-long-poll", race: "listener-before-recheck", observed: page.observed, hasMore: page.hasMore, terminalReason: terminalPage.completionReason })); +} finally { + await new Promise((resolve) => server.server.close(() => resolve())); + await Promise.all([primary.close(), concurrent.close()]); +} + +function createRunInput(label: string): CreateRunInput { + return { + tenantId: "unidesk", + projectId: `pikasTech/agentrun-${label}`, + workspaceRef: { kind: "host-path", path: "/tmp/agentrun-postgres-events-long-poll" }, + providerId: "NC01", + backendProfile: "codex", + executionPolicy: { + sandbox: "workspace-write", + approval: "never", + timeoutMs: 1000, + network: "none", + secretScope: { allowCredentialEcho: false, providerCredentials: [] }, + }, + traceSink: null, + }; +}