refactor(agent): route all SSE streams through a shared seam with a required heartbeat - #368
Merged
Merged
Conversation
The agent had three different SSE implementations: containers-events and logs each hand-rolled a ReadableStream with their own closed flag, teardown path and copy of the 5s heartbeat interval, while stats and zfs had no heartbeat at all. A Docker host with zero running containers therefore produced a completely silent stats stream, and Bun's default HTTP idleTimeout of 10s dropped the socket, putting the worker into a reconnect loop. zfs has the same exposure: `zpool iostat -v 1` normally ticks every second, but a degraded pool with a hung disk can stall output for long stretches, which is exactly when the dashboard should not be reconnecting. Port the web app's createSseStream seam into agent/src/lib/sse-stream.ts and build all four routes on it. The heartbeat is armed unconditionally inside stream construction, so there is no call for a future route to forget. Frames go out through an SseEmitter (data/event/raw/close) rather than hand-written `data: ` strings, one teardown owns the closed flag, and abort is registered before onStart with a recheck after it resolves so a subscription cannot leak past a disconnect. This duplicates src/lib/sse/create-sse-stream.ts. The agent is deliberately not a workspace member and web/worker import only types from it, so it cannot import web code; the seam has to exist once on each side of that split. Routes that own a subprocess or subscription (stats, zfs) now return their cleanup from onStart before starting their read/poll loop, so a teardown mid-loop kills the subprocess and stops the loop instead of leaving it running against a dead consumer. Heartbeat stays at 5s and keeps the bare `:\n\n` comment frame. Consumers are unaffected: the worker's shared SSE reader skips messages with no `data: ` line, the docker-logs route pipes agent bytes through verbatim, and browser EventSource ignores comment frames natively. sse-utils.ts had no remaining consumer after the conversion and is removed. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: ASSERTIVE Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
💤 Files with no reviewable changes (1)
📝 WalkthroughWalkthroughAgent SSE routes now use shared ChangesAgent SSE streaming
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related issues
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@agent/src/__tests__/containers-events.test.ts`:
- Around line 54-64: Move parseDataFrames from containers-events.test.ts into
the shared test-utility module under src/lib/test/, then import and reuse it in
both containers-events.test.ts and zfs.test.ts. Also consolidate the duplicated
readUntil helper there and update all agent SSE tests to use the shared
implementation, removing their local copies while preserving behavior.
- Around line 651-669: Update the test “a quiet host still gets a comment
heartbeat on the 5s cadence” to capture the original globalThis.setInterval
before mocking it, then wrap the test assertions and response handling in
try/finally and restore the captured function in finally, matching the cleanup
pattern used by the equivalent stats and zfs tests.
In `@agent/src/lib/sse-stream.ts`:
- Around line 67-79: Update the write function so build() serialization errors
are caught separately, logged as unexpected payload failures, and return without
calling teardown; keep controller.enqueue errors on the existing error-handling
and teardown path.
In `@agent/src/routes/zfs.ts`:
- Around line 57-60: Update the subprocess options in the zpool iostat spawn
flow to prevent an undrained stderr pipe: set stderr to ignore, or explicitly
drain proc.stderr if ZFS diagnostics must be exposed. Keep stdout piped for SSE
frame processing.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 93d5c45e-bfe6-41ef-b2f3-bf67b6d061ab
📒 Files selected for processing (13)
CLAUDE.mdagent/src/__tests__/containers-events.test.tsagent/src/__tests__/logs.test.tsagent/src/__tests__/stats.test.tsagent/src/__tests__/zfs.test.tsagent/src/lib/__tests__/sse-stream.test.tsagent/src/lib/__tests__/sse-utils.test.tsagent/src/lib/sse-stream.tsagent/src/lib/sse-utils.tsagent/src/routes/containers-events.tsagent/src/routes/logs.tsagent/src/routes/stats.tsagent/src/routes/zfs.ts
💤 Files with no reviewable changes (2)
- agent/src/lib/tests/sse-utils.test.ts
- agent/src/lib/sse-utils.ts
`zpool iostat -v 1` runs for the whole SSE session with `stderr: 'pipe'`
and nothing ever reads it. Sustained stderr fills the OS pipe buffer and
blocks the child, stalling stdout frames while the new heartbeat keeps
the socket alive, which hides exactly the wedged-pool case the heartbeat
was added to surface. Switch that spawn to `stderr: 'ignore'`; the frame
protocol carries only `{ line, timestamp }`, so there is no channel for
ZFS diagnostics and nothing to drain. The short-lived `zpool list -Hp`
spawn keeps its pipe, since it reads stderr on non-zero exit.
Move `readUntil` and `parseDataFrames` into agent/src/lib/test/, mirroring
the web package's src/lib/test/ convention, and drop the three local
copies. The consolidated `readUntil` is the containers-events variant,
which clears its race timer instead of leaking one per read.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
The streaming `zpool iostat -v 1` subprocess piped stderr and never read it, so a chatty pool fills the OS pipe buffer and blocks zpool's stdout. The previous commit fixed the blocking with `stderr: 'ignore'`, which also threw the diagnostics away. Read the pipe instead and log each line, matching the short-lived `zpool list -Hp` spawn in the same file, which already surfaces stderr on failure. The stdout pump and the stderr drain share a `readLines` helper rather than repeating the chunk-buffering loop; both observe the same `stopped` flag the SSE cleanup flips. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
agent/src/__tests__/containers-events.test.ts (1)
424-445: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winSynchronize the subscription race without sleeping.
Line 443 can race on slow CI. Resolve a promise inside the
listContainersmock, then await that promise before aborting.Proposed fix
+ let markSubscribeStarted!: () => void; + const subscribeStarted = new Promise<void>((resolve) => { + markSubscribeStarted = resolve; + }); const listHolder: { resolve: ((v: unknown[]) => void) | null } = { resolve: null }; const { stream, destroyed } = makeDestroyableEventsStream(); const docker = { listContainers: mock(() => new Promise<unknown[]>((resolve) => { listHolder.resolve = resolve; + markSubscribeStarted(); })), @@ - await new Promise((r) => setTimeout(r, 10)); + await subscribeStarted; ac.abort();🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@agent/src/__tests__/containers-events.test.ts` around lines 424 - 445, Replace the timing-based setTimeout wait in the abort-during-broadcasterSubscribe test with an explicit synchronization promise resolved by the listContainers mock. Await that promise to confirm subscription is blocked in listContainers before calling ac.abort(), then resolve listHolder so the late subscriber teardown path remains tested deterministically.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@agent/src/__tests__/containers-events.test.ts`:
- Line 6: Replace the relative sse-test-utils imports with the
`@/lib/test/sse-test-utils` alias in agent/src/__tests__/containers-events.test.ts
lines 6-6, agent/src/__tests__/stats.test.ts lines 4-4, and
agent/src/__tests__/zfs.test.ts lines 4-4; no other changes are needed.
In `@agent/src/lib/test/sse-test-utils.ts`:
- Around line 28-30: Update the finally path in the SSE test helper to await
reader.cancel() and then release the reader lock, ensuring asynchronous teardown
completes and any errors are propagated through the helper’s existing async
flow.
- Around line 11-20: Update readUntil to treat timeoutMs as a total deadline by
creating the timeout promise once before the while loop or calculating remaining
time from a fixed deadline. Remove per-iteration timer recreation and ensure
timer cleanup remains correct while preserving the existing predicate and
reader.read handling.
In `@agent/src/routes/zfs.ts`:
- Line 28: Remove the redundant JSDoc immediately above pumpZpoolOutput, while
retaining the separate comment documenting the lifecycle constraint below.
---
Outside diff comments:
In `@agent/src/__tests__/containers-events.test.ts`:
- Around line 424-445: Replace the timing-based setTimeout wait in the
abort-during-broadcasterSubscribe test with an explicit synchronization promise
resolved by the listContainers mock. Await that promise to confirm subscription
is blocked in listContainers before calling ac.abort(), then resolve listHolder
so the late subscriber teardown path remains tested deterministically.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: de41bf06-20fa-4a10-8164-1a87e61f5b5d
📒 Files selected for processing (5)
agent/src/__tests__/containers-events.test.tsagent/src/__tests__/stats.test.tsagent/src/__tests__/zfs.test.tsagent/src/lib/test/sse-test-utils.tsagent/src/routes/zfs.ts
Comment on lines
+28
to
+30
| } finally { | ||
| reader.cancel(); | ||
| } |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== locate file =="
git ls-files | rg '(^|/)sse-test-utils\.ts$' || true
echo "== file excerpt =="
file="$(git ls-files | rg '(^|/)sse-test-utils\.ts$' | head -n1 || true)"
if [ -n "${file:-}" ]; then
wc -l "$file"
cat -n "$file"
fi
echo "== stream reader cancel/releaseLock usage in codebase =="
rg -n '\b(reader|readable|ReadableStream)\.(cancel|releaseLock)\(' -g '*.ts' -g '*.tsx' . || true
echo "== runtime semantics probe if Node is available =="
node - <<'JS'
console.log("node", process.version);
const { TransformStream } = require("stream/web");
let reader;
let cancelled = false;
let released = false;
let cancelCallCount = 0;
let releaseCallCount = 0;
const ts = new TransformStream({
transform(chunk, controller) {
controller.enqueue(chunk);
controller.terminate();
},
});
reader = ts.readable.getReader();
reader.closed
.catch(() => {})
.then(() => {})
.then(async () => {
console.log("readUntil resolved before cancel await?", { cancelled, released, cancelCallCount, releaseCallCount });
});
const promise = reader.cancel();
console.log("cancel returns promise?", promise instanceof Promise);
promise.then(
(value) => {
cancelled = true;
reader.releaseLock();
released = true;
cancelCallCount = 1;
},
(err) => {
console.log("cancel rejected:", String(err));
cancelled = true;
reader.releaseLock();
released = true;
cancelCallCount = 1;
}
);
releaseCallCount = 1;
reader.releaseLock();
console.log("final cancelled released cancel release calls?", { cancelled, released, cancelCallCount, releaseCallCount });
JSRepository: jaredglaser/homelab-manager
Length of output: 4638
🌐 Web query:
MDN ReadableStreamDefaultReader cancel promise releaseLock
💡 Result:
In the context of the Streams API, both ReadableStreamDefaultReader.cancel() and ReadableStreamDefaultReader.releaseLock() are methods used to manage the lifecycle and ownership of a stream, but they serve distinct purposes [1][2][3]. ReadableStreamDefaultReader.cancel The cancel method is used to signal that the consumer is no longer interested in the stream [1][2]. Calling it performs a terminal action: it cancels the stream entirely, meaning no further data can be read, even if chunks remain in the internal queue [2][4]. It returns a promise that resolves when the stream is successfully canceled [2]. Because cancellation is terminal, the stream cannot be reused afterward [4]. ReadableStreamDefaultReader.releaseLock The releaseLock method releases the lock the reader holds on the stream, allowing the stream to be read by a different reader in the future [3][5]. Importantly, releasing the lock does not cancel the stream [3]. If there are any pending read requests when releaseLock is called, those requests are immediately rejected with a TypeError [3][6]. The stream remains open, and any unread chunks stay in the internal queue [3]. Key Differences and Interaction 1. Terminal vs. Non-Terminal: cancel is a terminal operation that ends the stream [4]. releaseLock is a non-terminal operation that simply relinquishes control [3]. 2. Order of Operations: You cannot call any reader methods (including cancel) after calling releaseLock, as the reader no longer has access to the stream [4]. Therefore, if you intend to both cancel and release the lock, you must call cancel first [4]. 3. Pending Reads: While both methods can affect pending reads, they do so differently: cancel resolves them (signaling the stream is done), whereas releaseLock rejects them with a TypeError [3][4][6]. In practice, if you have a reader and wish to stop consuming a stream, you should use reader.cancel [2][4]. If you simply need to transfer control of the stream to another part of your application, you should use reader.releaseLock [3][5].
Citations:
- 1: https://developer.mozilla.org/en-US/docs/Web/API/ReadableStreamDefaultReader
- 2: https://developer.mozilla.org/en-US/docs/Web/API/ReadableStreamDefaultReader/cancel
- 3: https://developer.mozilla.org/en-US/docs/Web/API/ReadableStreamDefaultReader/releaseLock
- 4: The difference between ReadableStream.cancel() and ReadableStreamDefaultReader.cancel()? whatwg/streams#1033
- 5: https://developer.mozilla.org/en-US/docs/Web/API/Streams_API/Using_readable_streams
- 6: https://developer.mozilla.org/en-US/docs/Web/API/ReadableStreamDefaultReader/read
Await stream cancellation, then release the reader lock.
reader.cancel() returns a promise and reader.releaseLock() only relinquishes ownership. In the timeout/canonical exit path, this helper can resolve before asynchronous teardown completes, and any cancellation errors/release errors can become unhandled. Await cancellation and release the lock from the finally path.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@agent/src/lib/test/sse-test-utils.ts` around lines 28 - 30, Update the
finally path in the SSE test helper to await reader.cancel() and then release
the reader lock, ensuring asynchronous teardown completes and any errors are
propagated through the helper’s existing async flow.
The timer was rebuilt each loop iteration, so every chunk reset it. On a stream that keeps emitting (a 5s heartbeat is enough) a predicate that never matches would never trip the timeout and the helper would hang until bun's own per-test limit, reporting a generic timeout instead of the helper's message. Build the deadline once outside the loop. Also await reader.cancel() so a rejection is not unhandled. It rejects when the stream has already errored, which the reader-throws tests induce deliberately, so the await is guarded. Drops the pumpZpoolOutput JSDoc, which restated a body that is now four lines of readLines call. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This was referenced Jul 25, 2026
jaredglaser
added a commit
that referenced
this pull request
Jul 25, 2026
…374) ## Why this touches .gitignore `.gitignore` excluded everything under `.claude/` except `scripts/` and `settings.json`, so skills could not be committed at all, and none were. Cloud sessions start from a fresh clone and see only what is tracked, so a skill that lives on one machine is a skill the cloud agents do not have. This adds `!/.claude/skills/`. ## The skill `.claude/skills/pr-queue/SKILL.md` covers orchestrating the open PR queue, written from what this repo actually does rather than generic advice. **Where state lives.** Not in the repo. A per-session GitHub issue holds queue-level state (PR table, decisions waiting, parked questions) and each PR carries one status comment edited in place. That way an orchestrator's context can be compacted without losing track of what is in flight, what was already declined and why, and what is waiting on a human. **Session tagging.** Several orchestrator sessions can run at once, so each owns a tag derived from its session id and stamps it on every artifact it creates. One search resolves a whole session's work. **Triage rules.** Verify every CodeRabbit finding against current code before acting, and reply with evidence on every decline. Recorded because the false-positive rate is real: of 8 findings on #368, 3 were wrong or out of scope, and one that was correct came with a suggested diff that would have disabled the timeout it was fixing. **Traps that have cost time here.** Agent worktrees branch from `main` rather than the feature branch. `bun run setup` drifts the lockfile under bun 1.3.x. `get_status` shows only CodeRabbit and has reported green over two failing jobs that `get_check_runs` showed. Migrations collide silently because the filenames differ. ## Scope Documentation and one `.gitignore` line. No source, build, or CI changes. `pr-queue session pq-db6c36c3` --- _Generated by [Claude Code](https://claude.ai/code/session_01E72xKQhMceYgcw4WrtjZau)_ --------- Co-authored-by: Claude <noreply@anthropic.com>
This was referenced Jul 25, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Two of the agent's four SSE routes had no heartbeat. Rather than adding one to each, this ports the web app's
createSseStreamseam into the agent package and converts all four routes onto it, so the heartbeat is armed by stream construction and there is no call left to forget.Replaces #277, which was closed as overtaken:
src/lib/sse/create-sse-stream.tsalready owns the heartbeat on the web side, and both web factories delegate to it. Only the agent half was genuinely outstanding.The seam
agent/src/lib/sse-stream.tsis a port ofsrc/lib/sse/create-sse-stream.ts.createSseStream(request, { onStart, heartbeatMs = 5000 })returns the finishedResponse. The heartbeat is armed unconditionally insidestart()right after the': ok\n\n'flush.onStart(emit, signal)receives only anSseEmitter(data/event/raw/close) and may return a cleanup that runs exactly once at teardown. Oneteardownclears the interval, runs cleanup and closes the controller; every write goes through the deferred-buildwrite()which tears down on failure; abort is registered beforeonStartwith anabortedrecheck after it resolves.It is a deliberate copy, not an import. The agent is not a workspace member and web/worker take only types from it via
@homelab-manager/agent/*with no runtime dependency, so it cannot import web code. The duplication is what the package split forces.The web seam's test file was ported verbatim (import path only) to
agent/src/lib/__tests__/sse-stream.test.ts; all 18 pass unmodified, which is the evidence that the port is faithful.Routes converted
containers-events.tsandlogs.tslose their hand-rolledReadableStream,closedflags, abort handlers and inline heartbeat blocks.stats.tsandzfs.tsgain a heartbeat as a consequence of the conversion.stats.tspreviously had none at all, so a host with zero running containers produced a silent stream.Removed in total: 2 inline heartbeat implementations, 4 hand-rolled
ReadableStreamteardown paths and 2 localsendSSEhelpers.agent/src/lib/sse-utils.tswas deleted; no consumer remained.Why
zfs.tsneeded one after allThe earlier reasoning was that
zpool iostat -v 1emits every second so the stream cannot idle. That is true of the healthy case and insufficient. The route has no timeout, no tick of its own and no output of its own: every frame originates from a line read off zpool's stdout, andreader.read()blocks identically whether zpool is quiet or wedged. A degraded pool with a hung disk stalls output, Bun's 10s defaultidleTimeoutthen drops the socket, and that is exactly when the dashboard should stay connected.Lifecycle subtlety worth reviewing
stats.tsandzfs.tsown long-running loops. IfonStartawaited the loop, the returned cleanup would never be registered and a mid-loop teardown would never kill the subprocess or stop the poll. Both start the loop withvoidand return cleanup synchronously; the loop observes astoppedflag the cleanup flips.Constraints
5000ms, not 25000. Bun's default HTTP
idleTimeoutis 10s, whichcontainers-events.tsalready documented and is why the existing routes chose 5s. Frames use the bare':\n\n'comment, matching both existing implementations.Consumers verified, not assumed
All three streaming worker collectors go through
connectAgentSseStream, whosereadFramesdoesconst dataLine = lines.find((line) => line.startsWith('data: ')); if (!dataLine) continue;so comment frames are skipped before any JSON parse.src/routes/api/docker-logs.$containerId.tsforwards agent chunks verbatim throughemit.raw, so pings pass through and it needs no second heartbeat. BrowserEventSourceignores:comments per spec.Testing
bun run typecheck:allclean.':\n\n'frame on the wire, driven by aglobalThis.setIntervalspy per CLAUDE.md rule 7.bun run test:coverage:agentpasses;sse-stream.ts,containers-events.ts,logs.tsandzfs.tsall at 100% lines.src/lib/sse/__tests__/andsrc/worker/collectors/__tests__/: 134 pass. No web source files changed.containers-eventsis one test lower because its two inline heartbeat tests collapsed into one idle-heartbeat test; the clears-on-abort case is now covered at the seam.Note for the ESLint PR
This PR removes 22 of the 23 findings the proposed SSE lint rules produce. If this merges first, those five
no-restricted-syntaxselectors can land aserrorwith one remaining fix insrc/routes/api/settings.ts:17, rather than aswarn.Generated by Claude Code
Summary by CodeRabbit
: okstartup and regular 5s heartbeat comment frames.