perf(api): replace SSE polling with shared EventEmitter (ENG-1669)

[limaronaldo]

Problema

O endpoint SSE GET /api/logs/stream fazia polling no banco a cada 2s por conexão (getRecentTaskEvents), escalando linearmente o custo de DB com o número de clientes conectados.

Solução (ENG-1669)

• core/task-event-bus.ts (novo): EventEmitter compartilhado in-process; setMaxListeners(0) para suportar muitas conexões SSE.
• db.createTaskEvent: após o INSERT, emite o evento no bus — choke point único que cobre todos os ~20 call sites (orchestrator, swarm, user-proxy, router, etc.). Faz lookup leve de tasks.status (RML-716) apenas quando há listeners; broadcast é best-effort e nunca falha o write.
• Endpoint SSE: removido o setInterval de polling. Agora: subscribe no bus → catch-up único via Last-Event-ID/cursor → push em tempo real. Guards de dedupe cobrem a janela de sobreposição entre catch-up e eventos live. Keep-alive (30s) e cleanup no abort preservados.

Limitações

In-process only — se a API escalar para múltiplos processos, trocar por Postgres LISTEN/NOTIFY ou Redis pub/sub (documentado no módulo).

Verificação

• bunx tsc --noEmit -p packages/api → exit 0
• bun test packages/api/src/core/task-event-bus.test.ts → 3 pass
• bun test packages/api → 443 pass / 36 fail (falhas idênticas na árvore limpa: Visual Regression/BrowserManager, dependem de browser — pré-existentes)

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 3cee88b272

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

Drain every page during catch-up

getRecentTaskEvents defaults to 50 rows, but the new implementation performs this catch-up only once. When a new or reconnecting client is more than 50 events behind, it receives only the first page; rows 51 onward were persisted before the bus subscription and therefore will never be emitted live. The previous polling loop advanced the cursor through successive pages, so the catch-up must continue until fewer than 50 rows are returned.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Buffer live events until catch-up completes

If a live event is emitted while the catch-up query is pending, sendEvent advances lastCursor to that new event. The catch-up loop then treats every older returned row as already delivered and skips it, even though those rows were never sent on this connection. This loses backlog whenever work completes during connection setup; live events need to be buffered and merged with the catch-up results rather than advancing the shared cursor immediately.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Preserve live events emitted out of database order

For concurrent createTaskEvent calls, each insert is followed by an independent awaited status query before emission, so a later database row can be emitted before an earlier one. Once the later event advances lastCursor, this comparison rejects the earlier event as already delivered even though it was never sent. This can silently omit events from the unfiltered stream under normal concurrent task processing; deduplication should track delivered IDs or the bus must guarantee database ordering.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Register abort cleanup before awaiting catch-up

The abort handler is attached only after the catch-up database request completes. If the client disconnects while that request is pending, the abort event has already fired by the time this listener is registered, so the bus subscription remains installed and the subsequently created keep-alive timer continues targeting a dead stream. Register cleanup before the await and also handle an already-aborted signal.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Route every task-event insert through the event bus

Not every persisted event uses db.createTaskEvent: for example, the stale-task cleanup endpoint inserts STALE_CLEANUP directly in router.ts:1249-1262, and the prompt optimizer also inserts directly. Because polling has been removed, active SSE clients will never receive those events. Either centralize all inserts through the broadcasting method or use a database-level notification mechanism that observes every insert.

Useful? React with 👍 / 👎.

[chatgpt-codex-connector]

Emit the persisted event when status enrichment fails

If the post-insert status lookup fails transiently, control jumps to the catch block without calling emitTaskEvent. The event is already committed, and connected clients no longer poll after initial catch-up, so they permanently miss it. Treat only the status as best-effort by emitting mapped without taskStatus when enrichment fails, or provide another delivery/retry path.

Useful? React with 👍 / 👎.

[limaronaldo]

Review: ENG-1669 — replace SSE polling with shared EventEmitter

Verdict: REQUEST CHANGES (FAIL) — a confirmed BLOCKER data-loss race plus a second confirmed BLOCKER (backlog >50 events silently dropped on reconnect) make this unsafe to ship as-is. Findings below combine manual analysis with an independent codex exec (gpt-5.6-terra, medium reasoning, read-only) cross-check on the same diff; both reviewers independently converged on the core race.

Process note: codex exec was invoked with --skip-git-repo-check (required outside a trusted git root) and completed within budget (~9 min wall including one retry after a missing-flag failure; findings below are from the successful run, corroborated point-by-point against the diff before inclusion).

Findings

#	Severity	File:Line	Finding
1	BLOCKER	packages/api/src/router.ts:4340-4368 (live subscriber vs. catch-up loop, diff hunk ~198-268)	Silent permanent event loss via shared mutable cursor race. lastCursor is written by both the live taskEventBus.onTaskEvent handler and the catch-up loop over db.getRecentTaskEvents(lastCursor, taskId). If a live event with a later timestamp is delivered while the catch-up loop still has earlier, not-yet-sent backlog items pending, the live handler advances lastCursor past them; when the catch-up loop resumes, its own dedupe check (event.createdAt < lastCursor.createdAt, hunk lines ~254-259) now evaluates true for those earlier events and continues past them without sending. There is no requeue/backfill — the event is gone forever, and no future catch-up can recover it because the cursor has already passed its timestamp. This is a genuine violation of the "no lost events" requirement the polling-replacement was supposed to preserve; polling was not perfectly ordered either, but it never permanently dropped a persisted row. Traced concretely: lastCursor=T0 → catch-up query in flight (WHERE created_at > T0) captures E_a(T1), E_b(T2) → live event E_c(T2.5) arrives and is delivered first (sendEvent sets lastCursor=T2.5) → catch-up loop resumes, sends E_a, then evaluates E_b(T2) against lastCursor=T2.5 → T2 < T2.5 → skipped. E_b is never delivered.
2	BLOCKER	packages/api/src/router.ts:4350 (catch-up call site) + packages/api/src/integrations/db.ts:415-421 (getRecentTaskEvents default limit: number = 50)	Catch-up silently truncates backlogs >50 events. The router calls db.getRecentTaskEvents(lastCursor, taskId) with no limit argument, so it always uses the default of 50 and does not page. The old setInterval polling loop ran this same query every 2s, so a large backlog simply spread across multiple polls and eventually drained. With the one-shot catch-up model, any client that reconnects (or first connects) with more than 50 missed/backlogged events permanently loses everything past the first 50 — there is no follow-up fetch, and the live subscription only delivers events emitted after subscribe, not the truncated remainder. This is a straightforward, reproducible regression versus old polling semantics, not just a race-timing edge case.
3	HIGH	packages/api/src/integrations/db.ts:326-336 (catch (err) { console.error(...) } around the status-lookup + emit)	Best-effort catch swallows delivery, not just enrichment. The comment says "Broadcasting is best-effort; never fail the write because of it" — correct not to fail the write. But the try block wraps both the SELECT status lookup and taskEventBus.emitTaskEvent(...). If the status query throws (e.g. transient pool/timeout), the event is never emitted at all — under polling, the client would still see this event on the next poll since it's already persisted; under push-only delivery, a transient status-lookup failure now means silent, permanent non-delivery to all currently-connected clients (recoverable only if they later reconnect within the 50-row/backlog window, per finding #2). The status enrichment should be best-effort; the broadcast itself should not be coupled to it — e.g. emit with taskStatus: undefined on lookup failure rather than skipping emit entirely.
4	HIGH	packages/api/src/router.ts:4340 (taskEventBus.onTaskEvent subscribe) vs. packages/api/src/router.ts:4379-4384 (abort listener registration)	Listener/keep-alive registered before cleanup handler exists; disconnect during the await catch-up window leaks. The live listener is subscribed synchronously, then the code awaits db.getRecentTaskEvents(...) for the one-shot catch-up, and only after that resolves does execution reach the req.signal.addEventListener("abort", ...) registration that wires unsubscribe() + clearInterval(keepAlive) + controller.close(). If the client disconnects while the catch-up query is in flight, the abort event fires on req.signal before the listener is attached (or is simply missed if it already fired) — isActive never flips to false via the intended path, the EventEmitter listener is never removed, and later taskEventBus.emitTaskEvent(...) calls will keep invoking a callback bound to a dead connection, whose sendEvent() calls controller.enqueue() on an already-closed/errored controller (unhandled per-call error, only caught by the inner try/catch around sendEvent, which logs and moves on rather than actually cleaning up). Net effect: a slow catch-up query + a fast disconnect is a reliable way to leak a listener that fires (and errors) for the lifetime of the process. taskEventBus.setMaxListeners(0) (task-event-bus.ts:37) removes Node's default-10 warning that would otherwise have surfaced exactly this class of leak in logs/tests.
5	MEDIUM	packages/api/src/router.ts:4295-4309 (controller.enqueue in sendEvent)	No backpressure/queue bound on the shared push path. Every broadcast event is synchronously enqueue()'d into every subscribed connection's stream with no check of controller.desiredSize and no bound on a slow/stalled client's internal queue. Under the old polling model, a slow client just meant the next poll's batch grew (bounded partly by wall-clock interval); under push, a burst of task events (many agents completing concurrently) enqueues into every open SSE connection unconditionally, and a client that isn't reading fast enough (or is on a bad network) has no backpressure signal — memory grows unbounded until the socket actually closes/errors. Not a data-loss bug on its own, but a resource-exhaustion risk under load that polling structurally avoided.
6	MEDIUM	packages/api/src/router.ts (GET /api/logs/stream, whole handler) — no cancel() on the ReadableStream	No ReadableStream.cancel() handler; cleanup depends entirely on req.signal's abort event. This is unchanged from the pre-PR polling code (same gap existed there), but the failure mode is materially worse now: previously a missed abort left a bounded setInterval polling loop running (self-limiting, one DB query per 2s, and per-connection). Now a missed abort leaves a permanent EventEmitter listener registered on a shared, process-wide bus that fires on every future task event system-wide, for the life of the process, invoking controller.enqueue() against a closed controller each time (see finding #4). Given the PR explicitly changes the leak's blast radius (per-connection bounded timer → shared unbounded emitter), it should also add the cancel() handler as part of this change, not defer it.
7	LOW	packages/api/src/core/task-event-bus.test.ts (all 3 tests)	Test coverage gap: only the standalone TaskEventBus class is unit-tested; the SSE route handler itself (where every finding above lives) has zero test coverage. The 3 tests confirm subscribe/unsubscribe/50-concurrent-listeners work in isolation, which is useful but does not exercise: subscribe-then-catch-up ordering/race (finding #1), backlog >50 (finding #2), status-lookup failure during broadcast (finding #3), abort-during-await (finding #4), or backpressure (finding #5). Given the bulk of the risk is in router.ts's integration of the bus with the stream lifecycle, not in the bus primitive itself, this test suite provides limited confidence for the actual regression surface.

Confirmed non-issues (checked, ruled out)

• db.ts:315 listenerCountTaskEvent > 0 gate racing with task completion — the listener-count check is synchronous (no await between the check and either taking or skipping the broadcast path), so there's no window for a listener to appear/disappear mid-check that would cause incorrect behavior beyond "one extra/one fewer status query," which is cosmetic, not a correctness bug.
• UUID tiebreak correctness (event.id <= lastCursor.id[on equal timestamps]) — confirmed the DB query (db.ts:415-436) orders by created_at ASC, id ASC and the JS-side dedupe uses the same field/ordering convention, so the tiebreak itself is internally consistent. It's the shared-cursor race (finding #1) that breaks correctness, not the tiebreak comparator.
• taskEventBus.setMaxListeners(0) as a leak in itself — 0 means "unlimited," which is the documented, intentional choice given one listener per SSE connection is expected. Correct choice given the design; it only becomes a liability in combination with finding #4/#6 (no cancel(), so leaked listeners accumulate silently without Node's warning as a tripwire).

Summary

• BLOCKER: 2 — data-loss race between catch-up and live delivery (#1); catch-up truncates backlogs beyond 50 events with no pagination (#2). Either is sufficient to block merge.
• HIGH: 2 — best-effort catch swallowing broadcast, not just enrichment (#3); listener/keep-alive leak on disconnect-during-catch-up (#4).
• MEDIUM: 2 — no backpressure bound on push path (#5); no ReadableStream.cancel() handler, worse blast radius than before (#6).
• LOW: 1 — route-handler-level test coverage gap (#7).

Recommendation: Do not merge as-is. At minimum, fix #1 (replace the shared-cursor merge with an ordering-safe approach — e.g. buffer live events received during the catch-up await and replay/dedupe them against the final catch-up result before switching to live-only, or serialize catch-up-then-subscribe so there is no concurrent window) and #2 (page the catch-up query until exhausted, or explicitly cap and document a "too far behind, resync" path) before this can safely replace polling. #3, #4, and #6 should be addressed in the same PR given they compound the blast radius of any missed cleanup path.

[limaronaldo]

Rework pushed in fbe318d.

BLOCKER 1 (race on shared lastCursor) — fixed in fbe318d. Extracted the catch-up/live-merge logic into core/sse-log-stream.ts implementing buffer-then-drain: the live listener subscribes first but buffers events (does not send) while catch-up runs; catch-up drains fully; the buffer is then flushed deduped by event id; only then does live delivery go direct. There is a single dedup/advancement point (sentIds set keyed by event id), so the two independently-advancing timestamp cursors that caused the race no longer exist. Covered by the "BLOCKER 1" test in sse-log-stream.test.ts (injects a live event with a later timestamp mid-catch-up and asserts no backlog event is skipped).

BLOCKER 2 (truncating LIMIT 50, no pagination) — fixed in fbe318d. runSseLogStream now loops the catch-up query, paginating by cursor (pageSize defaults to 50, matching the prior LIMIT) until a short/empty page signals full drain, with a maxPages safety valve (200). Covered by the "BLOCKER 2" test: a 137-event backlog is delivered in full, in order.

HIGH 1 (SELECT + emit sharing one try/catch) — fixed in fbe318d. db.ts's createTaskEvent now wraps the status SELECT and the taskEventBus.emitTaskEvent call in separate try/catch blocks. A failing SELECT logs and degrades to emitting the event without taskStatus enrichment; it can no longer suppress the broadcast.

• Follow-up (not closed in this PR): no dedicated unit test exists for this path in db.ts — the module instantiates its neon() client at import time, so testing it in isolation needs either a mock.module on @neondatabase/serverless or a DI refactor of the sql client, which felt out of scope for this rework. Flagging as a follow-up rather than skipping silently.

HIGH 2 (listener leak on disconnect during catch-up) — fixed in fbe318d. In router.ts, req.signal.addEventListener("abort", cleanup) is now registered synchronously in start(), before any await, and cleanup() unsubscribes the live listener, clears the keep-alive interval, and closes the controller. Covered by the "HIGH 2" test: asserts listenerCount is 0 before start, 1 immediately after starting (pre-await), and back to 0 after stop() (idempotent).

MED (no backpressure check, no cancel() handling) — fixed in fbe318d. sink.send() in router.ts checks controller.desiredSize before every enqueue() and closes the connection via cleanup() if the client is at/below the threshold instead of growing the buffer unbounded. The ReadableStream's cancel() handler also invokes the same cleanup(). Covered by the "MED" test: simulates backpressure kicking in after 3 of 10 deliveries and asserts sending stops there.

Verification: bun test (451 pass, 36 pre-existing failures unrelated to this change — Playwright/Chromium browser binary missing in this sandbox, confined to ActionExecutor/BrowserManager/Visual Regression suites, nothing touching router/db/SSE), bun run typecheck clean. No lint script exists in this repo, so that step was not applicable.