feat: add manager-held events long polling

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",

src/mgr/postgres-store.ts

@@ -18,6 +18,7 @@ interface PostgresStoreOptions {
   connectionString: string;
   poolMax?: number;
   eventOutbox?: EventOutboxConfig;
+  onWaitForRunChangeListening?: (runId: string) => Promise<void>;
 }
 
 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<void>) | 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<void> {
@@ -601,6 +606,54 @@ CREATE TABLE IF NOT EXISTS agentrun_schema_migrations (
     return result.rows.map(eventFromRow);
   }
 
+  async countEvents(runId: string, afterSeq: number): Promise<number> {
+    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<typeof setTimeout> | 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<RunEvent[]> {
     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

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<ManagerServerOptions["runnerJobDefaults"]>; sessionPvcDefaults?: NonNullable<ManagerServerOptions["sessionPvcOptions"]>; providerProfileDefaults?: NonNullable<ManagerServerOptions["providerProfileOptions"]>; toolCredentialDefaults?: NonNullable<ManagerServerOptions["toolCredentialOptions"]>; runnerReconcilerOptions: RunnerReconcilerRuntimeOptions; runnerDispatcherOptions: RunnerDispatcherOptions; kafkaOutboxRelayOptions: KafkaOutboxRelayOptions; resolvedRunnerJobDefaults: RunnerJobDefaults; aipodSpecDir?: string }): Promise<JsonValue> {
+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<ManagerServerOptions["runnerJobDefaults"]>; sessionPvcDefaults?: NonNullable<ManagerServerOptions["sessionPvcOptions"]>; providerProfileDefaults?: NonNullable<ManagerServerOptions["providerProfileOptions"]>; toolCredentialDefaults?: NonNullable<ManagerServerOptions["toolCredentialOptions"]>; runnerReconcilerOptions: RunnerReconcilerRuntimeOptions; runnerDispatcherOptions: RunnerDispatcherOptions; kafkaOutboxRelayOptions: KafkaOutboxRelayOptions; resolvedRunnerJobDefaults: RunnerJobDefaults; aipodSpecDir?: string }): Promise<JsonValue> {
   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,10 +932,19 @@ 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 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) {
     const startedAt = Date.now();
@@ -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;

src/mgr/store.ts

@@ -79,6 +79,8 @@ export interface AgentRunStore {
   createRun(input: CreateRunInput, identity?: CreateRunIdentity): MaybePromise<RunRecord>;
   getRun(runId: string): MaybePromise<RunRecord>;
   listEvents(runId: string, afterSeq: number, limit: number): MaybePromise<RunEvent[]>;
+  countEvents(runId: string, afterSeq: number): MaybePromise<number>;
+  waitForRunChange(runId: string, afterSeq: number, timeoutMs: number, signal?: AbortSignal): Promise<"changed" | "timeout" | "aborted">;
   listEventsForCommand(runId: string, commandId: string, limit: number): MaybePromise<RunEvent[]>;
   createCommand(runId: string, input: CreateCommandInput): MaybePromise<CommandRecord>;
   admitSessionSteer(input: SessionSteerAdmissionInput): MaybePromise<SessionSteerAdmission | null>;
@@ -262,6 +264,7 @@ export class MemoryAgentRunStore implements AgentRunStore {
   private readonly queueReadCursors = new Map<string, QueueReadCursorRecord>();
   private readonly runnerCreateFenceTails = new Map<string, Promise<void>>();
   private readonly runnerRetentionFences = new Set<string>();
+  private readonly runChangeWaiters = new Map<string, Set<() => 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<BackendTurnResult, "terminalStatus" | "failureKind" | "failureMessage" | "threadId" | "turnId">): RunRecord {

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<void>((resolve) => server.server.close(() => resolve()));
   }
@@ -48,6 +49,75 @@ async function assertManagerAuthBoundary(): Promise<void> {
   }
 }
 
+async function assertEventsLongPolling(client: ManagerClient, store: MemoryAgentRunStore): Promise<void> {
+  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<JsonRecord>;
+  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<JsonRecord>;
+  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<JsonRecord>;
+  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<JsonRecord>;
+  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<void> {
   const run = store.createRun({
     tenantId: "unidesk",

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<JsonRecord>;
+  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<JsonRecord>;
+  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<void>((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,
+  };
+}