feat(mt#1853): Cockpit server-side SSE broker: Postgres NOTIFY to SSE clients

[minsky-ai]

Summary

Implements the cockpit server-side SSE broker (mt#1853) that bridges Postgres NOTIFY events to connected SSE clients. Three pieces shipped in one PR:

• Piece A — src/cockpit/topic-filter.ts: pure matchesTopic() function for glob-style prefix matching
• Piece B — src/cockpit/sse-broker.ts: SseBroker class wrapping ChannelListener (mt#1852) with ring buffer, monotonic event IDs, and topic-filter dispatch
• Piece C — src/cockpit/server.ts: new GET /api/events SSE endpoint + eager broker init + comprehensive ADR-010 channel pre-subscription
• Piece D — src/cockpit/widgets/attention.ts: defaultDepsFactory reads active window key from broker's ring buffer (with same-windowKey close-cancellation semantics)

Execution evidence:

Final test run (post-R3 fixes, commit 8ba07bd28):

$ bun test --preload ./tests/setup.ts --timeout=15000 ./src/cockpit ./src/domain/mesh

src/cockpit/topic-filter.test.ts: 19 pass, 0 fail
src/cockpit/sse-broker.test.ts:   19 pass, 0 fail (incl. 4-test NOTIFY→SSE integration block)
src/cockpit/cockpit.test.ts:     (existing tests; no regressions)
src/domain/mesh/postgres-channel-listener.test.ts: 22 pass, 0 fail

73 pass cockpit, 22 pass mesh = 95 pass total, 0 fail
Ran 95 tests across 4 files. [248.00ms]

New test coverage:

• topic-filter.test.ts (19): empty patterns, * wildcard, exact match, glob prefix (attention.*, minsky.*), dotted-boundary prefix, multi-pattern OR, mismatch scenarios
• sse-broker.test.ts (19): ring buffer sizing/eviction, monotonic IDs, Last-Event-ID replay with/without buffer hit, topic-filtered replay, two-client dispatch isolation, wildcard client, closed/detached client guards, ensureChannel deduplication, close() teardown, +4-test NOTIFY→SSE integration block covering same-tick delivery, topic-filter exclusion, two-channel dispatch, and Last-Event-ID reconnect

CI: build + bundle-boot-smoke + Prevent Placeholder Tests all green on 8ba07bd28.

Key Changes

Topic filter (src/cockpit/topic-filter.ts)

• matchesTopic(channel, patterns) with explicit dotted-boundary semantics: * wildcard, exact match, glob prefix attention.* matches minsky.attention_window_opened (strict prefix, NOT substring — minsky.noattention does NOT match), multi-pattern OR

SSE broker (src/cockpit/sse-broker.ts)

• SseBroker holds one ChannelListener instance; ensureChannel() is idempotent (deduplicates postgres-js LISTEN registrations)
• Ring buffer (SseEvent[]) capped at ringBufferSize (default 100); evicts oldest on overflow
• attachClient(client, lastEventId?) registers a client and returns buffered events to replay after lastEventId; if lastEventId was evicted, returns full buffer (best-effort replay)
• detachClient(clientId) removes a client; close() unsubscribes all channels and clears clients
• latestForChannel(channel) scans the ring buffer backward for the most recent event on a channel
• All dispatch to clients checks client.closed before calling send() (defensive against close-vs-dispatch race)

Cockpit server (src/cockpit/server.ts)

• CockpitServerOptions.overrideSseBroker for test injection
• Eager init via initServerSseBroker() called from CLI entrypoint src/commands/cockpit/start-command.ts — broker is ready before first SSE client connects (R1 BLOCKING build(deps-dev): bump eslint from 8.57.1 to 9.26.0 #1 fix)
• Comprehensive pre-subscription — COCKPIT_SSE_CHANNELS enumerates all 6 ADR-010 §3 canonical channels (2 attention currently emitting; 4 session/task pending mt#1854). postgres-js sql.listen() doesn't accept wildcards; pre-subscribing all spec-defined channels at startup is the architecturally-correct alternative to dynamic subscription (R2 BLOCKING fix).
• Falls back to noop listener for non-Postgres backends (200 endpoint, silent stream); 503 only on unexpected init failure
• GET /api/events: parses ?topics= (defaults to ["*"]), reads Last-Event-ID header, writes text/event-stream headers, replays ring-buffer events, sends 30s : keep-alive heartbeats, detaches client on req.on("close"); returns 400 on malformed topics

Attention widget (src/cockpit/widgets/attention.ts)

• defaultDepsFactory lazy-loads the shared broker and queries broker.latestForChannel("minsky.attention_window_opened") to derive activeWindowKey
• Same-windowKey close-cancellation (R3 NON-BLOCKING fix): cancellation requires both same windowKey AND newer-than-open. open(A) → close(B) does NOT clear A's active state.
• Replaced the v0 "no active window key" comment block with a one-line doc
• AttentionDeps interface shape unchanged (keeps activeWindowKey: string | null) so all existing tests pass without modification

R1/R2/R3 reviewer convergence

• R1 (2x CHANGES_REQUESTED, both at create-time): channel-registration gap + missing NOTIFY→SSE integration test + 3 NON-BLOCKING refinements. Fixed in commit 49cf1fbfc — eager broker init, COCKPIT_SSE_CHANNELS constant, 4 new NOTIFY→SSE integration tests, plus the 3 nits.
• R2 (CHANGES_REQUESTED): reiterated channel-registration concern — clients requesting session.* topics weren't covered by the 2-channel pre-subscription. Fixed in commit d1a020844 — comprehensive pre-subscription of all 6 ADR-010 §3 channels (4 channels pending mt#1854 producers).
• R3 (APPROVED with one NON-BLOCKING + full spec-verification table): close-event-clearing-different-window edge case. Fixed in commit 8ba07bd28 — AND semantics on close-cancellation.

Concurrency analysis

The broker runs in a JS single-threaded event loop so no concurrent mutation races exist between attachClient, detachClient, and _dispatch. However, the req.on("close") handler fires asynchronously from the HTTP layer, so a client may receive one final send() call after the connection closes. This is defended by:

• client.closed checked at the top of send() — the stub returns early
• client.closed checked in _dispatch before calling send() — broker skips closed clients
• req.on("close") sets closed = true before calling broker.detachClient() so any in-flight dispatch sees the closed flag

The ring-buffer append (push + conditional shift) is synchronous and not interleaved with other operations given JS's event loop guarantee.

Out of scope

• mt#1148 Stage 2 contract changes
• Event taxonomy expansion (mt#1854) — the 4 future channels in COCKPIT_SSE_CHANNELS get producers when mt#1854 ships
• Cockpit web client UI changes — the JS that consumes the SSE stream from a browser

Live verification

(N/A — not a structural change requiring live verification per /implement-task §7a. The integration test covers the NOTIFY→broker→client dispatch path using createRecordingChannelListener from mt#1852. End-to-end HTTP-layer testing of /api/events against a live Postgres requires a running DB with minsky.attention_window_opened producer firing; deferred to mt#1148 Stage 2 integration test or operator validation when the cockpit is brought up against the production Postgres.)

Context

• Parent: mt#1001 (mesh signal channel ADR)
• ADR-010 §2 — defines the SSE-broker shape implemented here
• ADR-010 §3 — defines the 6 canonical channel names this PR pre-subscribes to
• mt#1147 — Attention widget; this PR closes its deferred activeWindowKey: null v0 fallback
• mt#1148 — Cockpit push transport SSE adapter (Stage 2); unblocked by this PR
• mt#1852 — PostgresChannelListener library this PR consumes; DONE
• mt#1854 — Event taxonomy expansion; producers for the 4 pre-subscribed but not-yet-emitting channels

[minsky-reviewer]

Independent adversarial review (Chinese-wall)
Reviewer: minsky-reviewer[bot] via openai:gpt-5
Tier: 3

---

Solid broker and topic-filter implementations with thorough unit tests, but a critical functional gap remains in the SSE endpoint. /api/events parses the topics query yet never subscribes the broker to those topics; only attention channels are pre-subscribed during lazy init, so requests like session.* won’t receive events. This violates the streaming semantics in the spec and the requirement to register listeners for all relevant channels. Additionally, no integration test verifies NOTIFY→SSE delivery. Please wire broker.ensureChannel() for client-requested topics (or all channels used by enabled widgets) and add/plan an integration test before merging.

Findings

• [BLOCKING] src/cockpit/server.ts:165 — /api/events never subscribes the broker to client-requested topics — only attention channels are pre-subscribed
  The SSE endpoint parses ?topics= but does not call broker.ensureChannel() for those patterns. The only subscriptions happen in getServerSseBroker() where it pre-subscribes minsky.attention_window_opened and minsky.attention_window_closed (see src/cockpit/server.ts:~120-135). As a result, clients requesting session.* (or any other topic not in the pre-list) will never receive events because the broker is not listening to those channels. This is a silent functional gap relative to the endpoint's contract and the task spec. Fix by deriving concrete channel names from topic patterns (at least the known top-level namespaces) and invoking broker.ensureChannel() accordingly, or by establishing subscriptions for all channels referenced by enabled widgets at server init.

[minsky-reviewer]

Independent adversarial review (Chinese-wall)
Reviewer: minsky-reviewer[bot] via openai:gpt-5
Tier: 3

---

Good progress on the SSE broker, topic filter, and SSE endpoint with strong unit coverage. Two spec-aligned issues block merge: the broker is lazily initialized and only pre-subscribes attention channels rather than initializing at server startup and registering all channels referenced by enabled widgets; and the required NOTIFY→SSE integration test is missing. I also flagged minor nits: 503 vs noop-listener behavior mismatch, potential stale activeWindowKey when the latest event is a close, and wording/semantics clarity in topic-filter docs. Please address the startup/registration behavior and add the integration test; the rest are non-blocking.

[minsky-reviewer]

Independent adversarial review (Chinese-wall)
Reviewer: minsky-reviewer[bot] via openai:gpt-5
Tier: 3

---

One blocking issue remains: the /api/events endpoint parses ?topics= and attaches clients with those filters but does not subscribe the broker to client-requested channels, relying only on pre-subscribed attention channels. This means topics outside the canonical set (e.g., session.*) won’t deliver events, failing the spec’s streaming requirement. The broker, topic-filter, ring buffer, replay, and eager init are well-implemented with solid unit coverage, and the attention widget now derives activeWindowKey from broker state. Address dynamic or comprehensive channel subscription so requested topics receive events, and this should be ready to merge.

[minsky-reviewer]

Independent adversarial review (Chinese-wall)
Reviewer: minsky-reviewer[bot] via openai:gpt-5
Tier: 3

---

The prior blocking finding about not subscribing to client-requested topics has been addressed by eager initialization and comprehensive pre-subscriptions via COCKPIT_SSE_CHANNELS; the CLI now calls initServerSseBroker() on startup. The /api/events endpoint correctly parses topics, respects Last-Event-ID, and streams SSE with heartbeats. Unit tests cover topic filtering, ring buffer, replay, and delivery. I noted one non-blocking concern in the attention widget: a newer close event for a different window may clear the active key; consider tightening that logic or documenting the intent. With no remaining blocking issues, this PR is approved.

Findings

• [NON-BLOCKING] src/cockpit/widgets/attention.ts:156 — Active-window derivation may clear a still-open window when an unrelated close event is newer
  In defaultDepsFactory(), the logic sets activeWindowKey = null when the latest close event is newer than the latest open event, even if the close targets a different windowKey:

const closeIsNewer = parseInt(latestCloseEvent.id, 10) > parseInt(latestOpenEvent.id, 10);
if (closeWindowKey === openWindowKey || closeIsNewer) {
  windowStillOpen = false;
}

This treats any newer close as cancelling the most recent open regardless of key. If multiple windows can exist sequentially (or concurrently), a close for a different window would erroneously clear the active key. Consider constraining the cancellation to the same windowKey, or additionally validating temporal ordering per key. If the “newer regardless of key” rule is intentional, a short code comment explaining why would help future maintainers.

Spec verification

Criterion	Status	Evidence
/api/events SSE endpoint on the cockpit HTTP server (src/cockpit/server.ts) accepts a topics query param (?topics=session.*,attention.*) and streams events matching the topic filter.	Met	Implemented at src/cockpit/server.ts:228-341. Parses ?topics= into topicPatterns, builds an SseClient with topics, replays filtered ring-buffer events, and streams live ones via writeSseEvent().
Cockpit server initializes one PostgresChannelListener at startup and registers listeners for every channel referenced by an enabled widget.	Met	initServerSseBroker() is called from CLI entrypoint src/commands/cockpit/start-command.ts:28-40 for eager init. getServerSseBroker() pre-subscribes to COCKPIT_SSE_CHANNELS (attention/session/task) at src/cockpit/server.ts:138-170 via broker.ensureChannel(channel).
Topic-filtering supports glob-style prefix patterns (attention.* matches minsky.attention_window_opened, minsky.attention_window_closed).	Met	matchesTopic() implemented at src/cockpit/topic-filter.ts with tests in src/cockpit/topic-filter.test.ts covering wildcard, exact, dotted-boundary prefix, and OR semantics.
Reconnect semantics on the SSE side use the standard EventSource Last-Event-ID header for resume; broker maintains a small ring buffer (size TBD in spec, default 100) for replay.	Met	Ring buffer and replay in src/cockpit/sse-broker.ts:77-121,144-174; DEFAULT_RING_BUFFER_SIZE=100. Last-Event-ID parsed and passed to broker.attachClient() at src/cockpit/server.ts:297-314, 324-331.
Attention widget consumes events to maintain activeWindowKey in production (drops the v0 null fallback documented in src/cockpit/widgets/attention.ts:188-192).	Met	defaultDepsFactory() loads the shared broker and calls latestForChannel("minsky.attention_window_opened") to derive activeWindowKey, with close-cancellation logic at src/cockpit/widgets/attention.ts:130-171.
Integration test: emit NOTIFY → SSE client receives the payload within 1s.	Met	Unit-level integration in src/cockpit/sse-broker.test.ts (e.g., "NOTIFY → SSE delivery integration" at ~lines 319-410) uses createRecordingChannelListener() to simulate pg_notify and verify dispatch reaches an attached client synchronously. While not hitting HTTP, it verifies the broker path per mt#1853.

[minsky-ai]

Supplementary main-agent review — documentation impact assessment

Posted to unblock the docs-impact merge gate alongside reviewer-bot R3 (review 4299754480, APPROVED with full spec-verification table).

Documentation impact

Doc surface	Impact	Action taken
docs/architecture/adr-010-mesh-signal-channel-push-subscription.md	This PR makes ADR-010 §2's SSE-broker architecture concrete: the /api/events endpoint, ring buffer, topic filter, comprehensive pre-subscription of all 6 ADR-010 §3 channels. No ADR amendment needed — ADR-010's §Substrate-constraint correction is tracked separately as mt#1857 (out of mt#1853 scope).	No change required in this PR.
docs/architecture/cockpit.md	References mt#1148 SSE transport (downstream) and mt#1147 (Attention widget — the consumer this PR updates). No required changes — the cockpit doc's existing description of "polling layer with per-widget intervals" is unchanged; the SSE broker layer is additive infrastructure that mt#1148 Stage 2 will adopt formally.	No change required.
CLAUDE.md / AGENTS.md	No new operator-facing rules or process. The new env var MINSKY_POSTGRES_SESSION_URL from mt#1852 is unchanged.	No change required.
.minsky/rules/*.mdc	No new rules added.	No change required.
src/cockpit/CLAUDE.md	The cockpit-domain guide (per the handoff memory). New COCKPIT_SSE_CHANNELS constant is the documented integration point for future widgets consuming mesh signals; the constant's JSDoc IS the documentation surface. No CLAUDE.md changes needed for this PR — when mt#1854 ships the producer side, the CLAUDE.md note can be updated alongside that work.	Deferred to mt#1854 (informational).
Memory entries	feedback_memory_snippet_conflation_at_artifact_write_time (de54bd12-...) from this session — no further updates needed.	No change.

Spec-verification coverage

Cross-referenced with reviewer-bot R3 (4299754480): all 6 success criteria are verified Met with file-line evidence in R3's body. The integration test was added in R1 fix commit; eager init + comprehensive subscription was added in R2 fix commit; the close-event-mis-clearing edge case was fixed in R3 fix commit (this push).

Out-of-scope items already filed as separate tasks

• mt#1854 — Event taxonomy expansion (the 4 future channels this PR pre-subscribes to; producers ship there).
• mt#1857 — ADR-010 §Substrate-constraint inline correction.
• mt#1858 — | null cleanup on persistence interface.
• mt#1859 — Pre-existing persistence test-isolation flake.

Verdict

Doc impact for THIS PR is bounded: no required doc surface changes; the new COCKPIT_SSE_CHANNELS constant's JSDoc is self-documenting for future channel additions. The 5 follow-up items are already filed and tracked.

[minsky-reviewer]

Independent adversarial review (Chinese-wall)
Reviewer: minsky-reviewer[bot] via openai:gpt-5
Tier: 3

---

All prior BLOCKING findings have been addressed. The server now eagerly initializes a shared SseBroker, pre-subscribes to a canonical set of channels, and exposes a robust /api/events SSE endpoint with topic filtering and Last-Event-ID replay. The attention widget correctly derives activeWindowKey from broker state with improved close-cancellation semantics. I found no new critical defects introduced by these changes. Verdict: APPROVE.

Spec verification

Criterion	Status	Evidence
/api/events SSE endpoint on the cockpit HTTP server (src/cockpit/server.ts) accepts a topics query param (?topics=session.*,attention.*) and streams events matching the topic filter.	Met	Implemented at src/cockpit/server.ts:338-460. Parses ?topics= into topicPatterns, builds an SseClient with topics, replays filtered ring-buffer events, and streams live ones via writeSseEvent().
Cockpit server initializes one PostgresChannelListener at startup and registers listeners for every channel referenced by an enabled widget.	Met	Eager init via initServerSseBroker() called in src/commands/cockpit/start-command.ts:43-49. Broker constructed in getServerSseBroker() and pre-subscribes to COCKPIT_SSE_CHANNELS at src/cockpit/server.ts:167-232 using broker.ensureChannel(channel).
Topic-filtering supports glob-style prefix patterns (attention.* matches minsky.attention_window_opened, minsky.attention_window_closed).	Met	matchesTopic() implemented at src/cockpit/topic-filter.ts with tests in src/cockpit/topic-filter.test.ts covering wildcard, exact, dotted-boundary prefix, and OR semantics.
Reconnect semantics on the SSE side use the standard EventSource Last-Event-ID header for resume; broker maintains a small ring buffer (size TBD in spec, default 100) for replay.	Met	Ring buffer and replay in src/cockpit/sse-broker.ts:88-126, 200-247; DEFAULT_RING_BUFFER_SIZE=100. Header parsed and passed to broker.attachClient() at src/cockpit/server.ts:413-420, 455-460.
Attention widget consumes events to maintain activeWindowKey in production (drops the v0 null fallback documented in src/cockpit/widgets/attention.ts:188-192).	Met	defaultDepsFactory() now loads the shared broker and derives activeWindowKey from latestForChannel("minsky.attention_window_opened") with close-cancellation keyed by windowKey at src/cockpit/widgets/attention.ts:216-271.