fix(broker): serialize terminal worker writes - #1488
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughThe broker now serializes worker writes through bounded queues, applies finite model-write deadlines, routes terminal PTY requests through non-blocking submission, and releases resize ownership during cleanup. Current-generation writer failures close affected sessions with preserved error details. ChangesWorker writer and terminal runtime
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant TerminalClient
participant BrokerRuntime
participant WorkerWriter
participant WorkerProcess
participant WorkerEvents
TerminalClient->>BrokerRuntime: send terminal or model request
BrokerRuntime->>WorkerWriter: enqueue bounded worker frame
WorkerWriter->>WorkerProcess: write and flush frame
WorkerWriter-->>WorkerEvents: report WriterFailed on write error
WorkerEvents->>BrokerRuntime: fail attached terminal sessions
BrokerRuntime-->>TerminalClient: send error and close notification
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 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 |
7f15f19 to
5f29b55
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 12
🧹 Nitpick comments (8)
packages/cli/src/cli/commands/local-agent.test.ts (2)
140-149: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd coverage for the already-prefixed message and the nonzero exit code.
Two branches in
local-agent.tsare untested:
- Line 616
message.startsWith('Error:').FleetNodeAttachErrormessages fromattach-fleet-node.tsLines 114 and 159 already carry the prefix. No test proves the command avoids emittingError: Error: ….- Line 613
if (code !== 0) deps.exit(code);. No test proves a nonzero attach code propagates.♻️ Proposed additional tests
it('attach --node does not double-prefix an already-prefixed error', async () => { const attachNode = vi.fn(async () => { throw new Error('Error: no agent named "lead"'); }); const { program, error, exit } = harness({ attachNode }); await program.parseAsync(['local', 'agent', 'attach', 'lead', '--node', 'finn'], { from: 'user' }); expect(error).toHaveBeenCalledWith('Error: no agent named "lead"'); expect(exit).toHaveBeenCalledWith(1); }); it('attach --node propagates a nonzero attach code', async () => { const attachNode = vi.fn(async () => 7); const { program, error, exit } = harness({ attachNode }); await program.parseAsync(['local', 'agent', 'attach', 'lead', '--node', 'finn'], { from: 'user' }); expect(error).not.toHaveBeenCalled(); expect(exit).toHaveBeenCalledWith(7); });🤖 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 `@packages/cli/src/cli/commands/local-agent.test.ts` around lines 140 - 149, Add tests alongside the existing attach --node coverage for both branches in the local-agent command: verify an already-prefixed attach error is emitted once without an additional “Error:” prefix, and verify a nonzero return code from attachNode is propagated to exit with no error output. Reuse the existing harness and command arguments.
107-112: 🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick winAssert that no broker credentials reach the node attach runner.
expect.objectContaining({ json: true })passes even if the command forwardsbrokerUrl,apiKey, orstateDir. Not forwarding those is the stated contract of--node, perCHANGELOG.mdLine 13. Assert the exact option object so a regression that leaks a broker credential fails the test.♻️ Proposed change
expect(attachNode).toHaveBeenCalledWith('lead', 'passthrough', 'daytona-live', { json: true, reasoning: undefined, diagnostics: undefined, });🤖 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 `@packages/cli/src/cli/commands/local-agent.test.ts` around lines 107 - 112, Update the attachNode assertion in the --node test to compare the complete expected options object rather than using expect.objectContaining. Include only the intended option, json: true, so the test fails if brokerUrl, apiKey, stateDir, or any other broker credentials are forwarded.packages/cli/src/cli/lib/attach.test.ts (1)
963-988: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider asserting that
flush()detaches the drain listener.The test verifies ordering. It does not verify the detach step at
attach.tsLine 1324. That step is what prevents a later'drain'emission from re-enteringflushQueueagainst an already-emptied queue. Track theoffcall and assert it, as the sibling test at Lines 1004-1008 does.♻️ Proposed addition
let drain: (() => void) | null = null; + let offCalled = false; const stdout: BackpressureWritable = { @@ off: (_event, listener) => { - if (listener === drain) drain = null; + if (listener === drain) { + offCalled = true; + drain = null; + } return undefined; }, }; const w = createBackpressureAwareWriter(stdout); w.write('first'); w.write('second'); w.write('third'); w.flush(); + expect(offCalled).toBe(true); w.dispose(); expect(written).toEqual(['first', 'second', 'third']);🤖 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 `@packages/cli/src/cli/lib/attach.test.ts` around lines 963 - 988, Extend the test “flushes buffered records in order before disposal” to track the writable’s off call and assert that flush() detaches the registered drain listener. Preserve the existing ordering assertions and verify the listener passed to off matches the one captured by once, consistent with the sibling test.packages/cli/src/cli/lib/attach-fleet-node.ts (1)
146-151: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winAdd a timeout to the ticket request.
Node's
fetchapplies no default request timeout. If Relaycast accepts the connection and never responds,startFleetNodeAttachProxynever resolves andagent attach --nodehangs with no output. Pass anAbortSignalso the command fails with the structuredFleetNodeAttachErrorpath instead.♻️ Proposed change
const nodePath = safeNodePath(options.node); - const ticketResponse = await fetchFn(`${baseUrl}/v1/nodes/${nodePath}/terminal/sessions`, { - method: 'POST', - headers: { Authorization: `Bearer ${workspaceKey}`, 'Content-Type': 'application/json' }, - body: JSON.stringify({ agent: options.agent, mode: options.mode }), - }); + const ticketResponse = await fetchFn(`${baseUrl}/v1/nodes/${nodePath}/terminal/sessions`, { + method: 'POST', + headers: { Authorization: `Bearer ${workspaceKey}`, 'Content-Type': 'application/json' }, + body: JSON.stringify({ agent: options.agent, mode: options.mode }), + signal: AbortSignal.timeout(SNAPSHOT_WAIT_MS), + });Note that an
AbortErrorpropagates as a rawDOMException, not aFleetNodeAttachError. Wrap the call intry/catchand rethrow asnew FleetNodeAttachError('Error: terminal session request timed out', 'node_unreachable')to keep the structured contract thatlocal-agent.tsLines 614-618 expects.🤖 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 `@packages/cli/src/cli/lib/attach-fleet-node.ts` around lines 146 - 151, Update the ticket request in startFleetNodeAttachProxy around ticketResponse to use an AbortController/AbortSignal with a finite timeout, and clear the timer after completion. Wrap the fetch call and response parsing in try/catch; convert an AbortError into FleetNodeAttachError with the timeout message and node_unreachable code, while preserving the existing structured error handling for other failures.crates/broker/src/runtime/worker_events.rs (1)
76-102: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider delegating
end_terminal_sessiontofail_terminal_session.
end_terminal_sessionrepeats the three-map cleanup already implemented byfail_terminal_sessionincrates/broker/src/runtime/fleet.rs(Lines 41-77). The only difference is that this variant omits the non-finalTerminalToCloud::Errorframe. If the session-cleanup contract changes later, the two copies can drift.An optional consolidation is to give
fleet::fail_terminal_sessiona flag that suppresses theErrorframe, then call it from both sites.🤖 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 `@crates/broker/src/runtime/worker_events.rs` around lines 76 - 102, Consolidate the cleanup in end_terminal_session with fleet::fail_terminal_session so the three terminal-request maps are maintained in one place. Extend fail_terminal_session with an option to suppress the non-final TerminalToCloud::Error frame, then invoke it from end_terminal_session while preserving the existing Closed notification and failure behavior.crates/broker/src/worker.rs (1)
1290-1311: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winReuse
encode_worker_framefor the shutdown frame.This block re-implements serialize-then-append-newline.
encode_worker_framealready does exactly that and keeps the newline invariant in one place.♻️ Proposed refactor
- let shutdown_frame = ProtocolEnvelope { - v: PROTOCOL_VERSION, - msg_type: "shutdown_worker".to_string(), - request_id: None, - payload: json!({"reason":"release","grace_ms": release_grace.as_millis() as u64}), - }; - let encoded = serde_json::to_vec(&shutdown_frame)?; - let mut frame = encoded; - frame.push(b'\n'); + let frame = encode_worker_frame( + "shutdown_worker", + None, + json!({"reason":"release","grace_ms": release_grace.as_millis() as u64}), + )?;🤖 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 `@crates/broker/src/worker.rs` around lines 1290 - 1311, Update the shutdown-frame encoding in the release flow to call the existing encode_worker_frame helper instead of manually serializing ProtocolEnvelope and appending a newline. Preserve the current shutdown payload and error propagation, then continue sending the resulting frame through WorkerWriteCommand.crates/broker/src/runtime/init.rs (1)
312-321: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the duplicated rationale in the channel comment.
The two comment paragraphs state the same bounding rationale twice. Keep one.
♻️ Proposed change
- // The terminal queue is deliberately bounded. A wedged remote attach must - // fail its session rather than accumulating unbounded PTY output in the - // broker or starving node control. - // Terminal bytes have a burstier profile than control actions. Keep this - // lane bounded, but give a short PTY burst room without impacting the - // independent control queue. + // Terminal bytes are burstier than control actions. Keep this lane bounded + // so a wedged remote attach fails its own session instead of accumulating + // unbounded PTY output or starving the independent node-control queue.🤖 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 `@crates/broker/src/runtime/init.rs` around lines 312 - 321, In the channel initialization block, remove the duplicated bounding rationale from the comments above terminal_control_tx and terminal_event_tx, retaining only one concise explanation of the intentional queue bounds and burst capacity.crates/broker/src/terminal_control.rs (1)
130-140: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueLog commands that are dropped while the token is missing.
The token-wait arm consumes a
TerminalControlCommand::Sendand discards it without any record. The caller'stry_sendalready returned success, so the runtime believes the frame reached the lane. A droppedterminal.closedorterminal.errorleaves the cloud session without a final frame until its own timeout fires.The window is small because sessions only exist after a connected
terminal.open, but a token rotation between a disconnect and a reconnect can hit it. Add atracing::debug!(orwarn!) so the drop is observable.♻️ Proposed change
tokio::select! { command = command_rx.recv() => { if matches!(command, Some(TerminalControlCommand::Shutdown) | None) { return; } - // Preserve bounded backpressure by dropping commands only - // when the caller itself chose a non-blocking try_send. + // Preserve bounded backpressure by dropping commands only + // when the caller itself chose a non-blocking try_send. + tracing::debug!( + target = "relay_broker::terminal", + "dropped terminal frame while waiting for a node token" + ); } _ = tokio::time::sleep(TOKEN_WAIT_DELAY) => {} }🤖 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 `@crates/broker/src/terminal_control.rs` around lines 130 - 140, In the token-wait branch of the terminal control loop, detect when the received command is a TerminalControlCommand::Send and emit a tracing::debug! or warn! record before discarding it. Keep Shutdown and channel-closed handling unchanged, and preserve the existing token-wait behavior for all other commands.
🤖 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 `@CHANGELOG.md`:
- Line 13: Move the agent-relay SSH fallback entry from the released [11.5.2]
section into [Unreleased], or remove it if it is not pending for the next
release; leave the released section describing only changes included in version
11.5.2.
In `@crates/broker/src/runtime/api.rs`:
- Around line 821-842: Add fail_terminal_sessions_for_worker next to
fail_terminal_session in crates/broker/src/runtime/fleet.rs, centralizing
session collection, removal, pending-request cleanup, Closed message delivery,
and rejection warnings. Replace the inline block in
crates/broker/src/runtime/api.rs#L821-L842 with a call using "agent_released"
and "terminal worker was released"; replace the corresponding block in
crates/broker/src/runtime/maintenance.rs#L288-L307 with a call using
"agent_exited" and "terminal worker exited".
- Around line 739-754: Update handle_api_request’s set_model path to enforce
timeout_ms around the send_raw_to_worker await, using the runtime’s existing
timeout mechanism and preserving successful/error results. Remove the advisory
tracing::info block, and ensure an expired deadline returns the established
timeout error without allowing the actor task to remain blocked on the writer.
In `@crates/broker/src/runtime/fleet.rs`:
- Around line 306-343: Update the TerminalControlEvent::Message Resize arm to
route resizing through the existing plan_resize and commit_resize_ownership
flow, keyed by session_id, before sending resize_pty to the worker. Reuse the
HTTP ResizePty path’s ownership and rejection behavior so remote and local drive
clients share the same single-resizer lease; only send the worker command after
ownership is successfully committed.
- Around line 180-201: The terminal-open handling around the terminal_sessions
insertion must clear all pending snapshot and input requests for the given
session_id before replacing the existing TerminalSession. Update the relevant
request collections before self.terminal_sessions.insert, preserving the new
session and request initialization behavior.
In `@crates/broker/src/runtime/maintenance.rs`:
- Around line 43-108: Replace the duplicated timeout cleanup and notification
logic in the expired snapshot and input sweeps with calls to the existing
fail_terminal_session helper. Pass the corresponding session ID, timeout code
("snapshot_timeout" or "input_timeout"), and timeout message, while preserving
removal of each expired request before invoking the helper.
In `@crates/broker/src/runtime/worker_events.rs`:
- Around line 1115-1117: Update the snapshot_failed branch in the terminal
session event handling to enqueue TerminalToCloud::Closed before removing the
session. Also remove or otherwise clean up the session’s pending entries in
terminal_snapshot_requests and terminal_input_requests, matching the cleanup and
close-frame behavior used by end_terminal_session and the maintenance timeout
paths.
- Around line 576-583: Update the command-writer failure handling around
workers.release so it terminates the affected worker without unregistering it
first, allowing reap_exited to invoke supervisor.on_exit and apply the
configured restart_policy. Use the existing failure-specific termination path if
available, and retain warning logging for termination errors.
In `@crates/broker/src/terminal_control.rs`:
- Around line 187-218: Add WebSocket liveness handling to the terminal
connection loop around connect_async and the tokio::select! in the terminal
control task. Configure a keepalive/read-timeout mechanism and ensure missed
liveness responses or expired reads set connected to false and close the socket,
rather than leaving the session active indefinitely; preserve the existing
reconnect and Disconnected-event flow.
In `@packages/cli/src/cli/lib/attach-fleet-node.ts`:
- Around line 438-448: Update the startup error handling in attachFleetNode so
every failure after server.listen, including the address guard and new
URL(terminalUrl), closes the listener before throwing. Convert malformed
terminal_url errors into the existing structured FleetNodeAttachError contract
with the appropriate code, and preserve TypeScript narrowing around abortStartup
using a non-null assertion or equivalent restructure.
- Around line 413-420: Replace the rawDataToString-based encoding in the
terminal.input handling with a rawDataToBuffer helper that preserves WebSocket
frame bytes across Buffer, ArrayBuffer, array, and fallback inputs, then
base64-encode that Buffer directly for remote.send.
- Around line 575-592: Update the cleanup flow around activeRemote so
termination waits for the terminal.close send callback or a short timeout,
ensuring the close payload is flushed before activeRemote.terminate(). Preserve
best-effort behavior and existing socket/server cleanup; if adding
server.closeAllConnections() for active HTTP requests, invoke it only after
server.close().
---
Nitpick comments:
In `@crates/broker/src/runtime/init.rs`:
- Around line 312-321: In the channel initialization block, remove the
duplicated bounding rationale from the comments above terminal_control_tx and
terminal_event_tx, retaining only one concise explanation of the intentional
queue bounds and burst capacity.
In `@crates/broker/src/runtime/worker_events.rs`:
- Around line 76-102: Consolidate the cleanup in end_terminal_session with
fleet::fail_terminal_session so the three terminal-request maps are maintained
in one place. Extend fail_terminal_session with an option to suppress the
non-final TerminalToCloud::Error frame, then invoke it from end_terminal_session
while preserving the existing Closed notification and failure behavior.
In `@crates/broker/src/terminal_control.rs`:
- Around line 130-140: In the token-wait branch of the terminal control loop,
detect when the received command is a TerminalControlCommand::Send and emit a
tracing::debug! or warn! record before discarding it. Keep Shutdown and
channel-closed handling unchanged, and preserve the existing token-wait behavior
for all other commands.
In `@crates/broker/src/worker.rs`:
- Around line 1290-1311: Update the shutdown-frame encoding in the release flow
to call the existing encode_worker_frame helper instead of manually serializing
ProtocolEnvelope and appending a newline. Preserve the current shutdown payload
and error propagation, then continue sending the resulting frame through
WorkerWriteCommand.
In `@packages/cli/src/cli/commands/local-agent.test.ts`:
- Around line 140-149: Add tests alongside the existing attach --node coverage
for both branches in the local-agent command: verify an already-prefixed attach
error is emitted once without an additional “Error:” prefix, and verify a
nonzero return code from attachNode is propagated to exit with no error output.
Reuse the existing harness and command arguments.
- Around line 107-112: Update the attachNode assertion in the --node test to
compare the complete expected options object rather than using
expect.objectContaining. Include only the intended option, json: true, so the
test fails if brokerUrl, apiKey, stateDir, or any other broker credentials are
forwarded.
In `@packages/cli/src/cli/lib/attach-fleet-node.ts`:
- Around line 146-151: Update the ticket request in startFleetNodeAttachProxy
around ticketResponse to use an AbortController/AbortSignal with a finite
timeout, and clear the timer after completion. Wrap the fetch call and response
parsing in try/catch; convert an AbortError into FleetNodeAttachError with the
timeout message and node_unreachable code, while preserving the existing
structured error handling for other failures.
In `@packages/cli/src/cli/lib/attach.test.ts`:
- Around line 963-988: Extend the test “flushes buffered records in order before
disposal” to track the writable’s off call and assert that flush() detaches the
registered drain listener. Preserve the existing ordering assertions and verify
the listener passed to off matches the one captured by once, consistent with the
sibling test.
🪄 Autofix
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: CHILL
Plan: Pro Plus
Run ID: 051b0bb2-1b96-4c89-ae2d-2222d313c3c4
📒 Files selected for processing (17)
CHANGELOG.mdcrates/broker/src/lib.rscrates/broker/src/runtime/api.rscrates/broker/src/runtime/event_loop.rscrates/broker/src/runtime/fleet.rscrates/broker/src/runtime/init.rscrates/broker/src/runtime/maintenance.rscrates/broker/src/runtime/mod.rscrates/broker/src/runtime/tests.rscrates/broker/src/runtime/worker_events.rscrates/broker/src/terminal_control.rscrates/broker/src/worker.rspackages/cli/src/cli/commands/local-agent.test.tspackages/cli/src/cli/commands/local-agent.tspackages/cli/src/cli/lib/attach-fleet-node.tspackages/cli/src/cli/lib/attach.test.tspackages/cli/src/cli/lib/attach.ts
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's internal server error or limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
🛑 Comments failed to post (9)
CHANGELOG.md (1)
13-13: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash # Description: Determine whether --ssh-host is new in the unreleased cycle. set -euo pipefail # Find the commit that introduced the --ssh-host option. rg -n --iglob '*local-agent*' -- '--ssh-host' || true echo "--- commits touching the --ssh-host option ---" fd -t f 'local-agent.ts' -x git log --oneline -S'--ssh-host' -- {} \; echo "--- released CHANGELOG headings ---" rg -n '^## \[' CHANGELOG.md | head -20 echo "--- prior CHANGELOG mentions of ssh-host ---" rg -n 'ssh-host' CHANGELOG.mdRepository: AgentWorkforce/relay
Length of output: 286
🏁 Script executed:
#!/bin/bash set -euo pipefail echo "--- matching source files ---" git ls-files | grep -E '(^|/)local-agent\.ts$' || true echo "--- current option references ---" rg -n -- '--ssh-host' . || true echo "--- commits introducing the option ---" while IFS= read -r file; do git log --all --oneline --reverse -S'--ssh-host' -- "$file" done < <(git ls-files | grep -E '(^|/)local-agent\.ts$') echo "--- changelog context ---" cat -n CHANGELOG.md | sed -n '1,45p' echo "--- changelog history for ssh-host ---" git log --all --oneline -S'ssh-host' -- CHANGELOG.mdRepository: AgentWorkforce/relay
Length of output: 2565
🏁 Script executed:
#!/bin/bash set -u file='packages/cli/src/cli/commands/local-agent.ts' echo "--- commits introducing the option ---" git log --all --oneline --reverse -S'--ssh-host' -- "$file" || true echo "--- recent commits for the source file ---" git log --all --oneline -12 -- "$file" || true echo "--- changelog headings and entry ---" grep -nE '^## \[' CHANGELOG.md | head -20 || true cat -n CHANGELOG.md | sed -n '1,35p' echo "--- changelog history for ssh-host ---" git log --all --oneline -S'ssh-host' -- CHANGELOG.md || true echo "--- release tags containing the introducing commit ---" intro=$(git log --all --format='%H' --reverse -S'--ssh-host' -- "$file" | head -1) if [ -n "$intro" ]; then git tag --contains "$intro" | head -30 echo "introducing commit: $intro" fiRepository: AgentWorkforce/relay
Length of output: 150
🏁 Script executed:
#!/bin/bash set -u file='packages/cli/src/cli/commands/local-agent.ts' echo "--- changelog context ---" grep -nE '^## \[' CHANGELOG.md | head -20 || true cat -n CHANGELOG.md | sed -n '1,35p' echo "--- current branch and recent history ---" git branch --show-current git log --oneline --decorate -20 echo "--- option history on the current branch ---" git log --oneline -S'--ssh-host' -- "$file" | head -20 || true echo "--- changelog history on the current branch ---" git log --oneline -S'ssh-host' -- CHANGELOG.md | head -20 || true echo "--- tags ---" git tag --sort=-version:refname | head -20Repository: AgentWorkforce/relay
Length of output: 150
🏁 Script executed:
#!/bin/bash set -u echo "--- changelog headings and entry ---" grep -nE '^## \[' CHANGELOG.md || true cat -n CHANGELOG.md | sed -n '1,35p' echo "--- source declaration and package versions ---" sed -n '550,640p' packages/cli/src/cli/commands/local-agent.ts grep -R -nE '"version":|version:' package.json packages/cli/package.json 2>/dev/null | head -20 || true echo "--- repository metadata ---" git rev-parse --show-toplevel git branch --show-current git show-ref --tags 2>/dev/null | head -20 || trueRepository: AgentWorkforce/relay
Length of output: 11776
Move this entry to
[Unreleased]or remove it.[11.5.2]is a released section, and both packages already declare version11.5.2; the entry does not describe the pending release.🤖 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 `@CHANGELOG.md` at line 13, Move the agent-relay SSH fallback entry from the released [11.5.2] section into [Unreleased], or remove it if it is not pending for the next release; leave the released section describing only changes included in version 11.5.2.Source: Coding guidelines
crates/broker/src/runtime/api.rs (1)
821-842: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Add one helper that closes every terminal session for a worker. Both sites inline the same sequence: collect the session ids whose
session.agentmatches the worker, remove each session, filterterminal_snapshot_requestsandterminal_input_requests, queue oneTerminalToCloud::Closed, and warn when the queue rejects it. Only the close code and message differ. A future change to the teardown order or to the reserve-capacity handling must currently be applied twice.Add
fail_terminal_sessions_for_worker(terminal_control_tx, terminal_sessions, terminal_snapshot_requests, terminal_input_requests, &name, code, message)next tofail_terminal_sessionincrates/broker/src/runtime/fleet.rs, then call it from both sites.
crates/broker/src/runtime/api.rs#L821-L842: replace the inline block with a call passing"agent_released"and"terminal worker was released".crates/broker/src/runtime/maintenance.rs#L288-L307: replace the inline block with a call passing"agent_exited"and"terminal worker exited".📍 Affects 2 files
crates/broker/src/runtime/api.rs#L821-L842(this comment)crates/broker/src/runtime/maintenance.rs#L288-L307🤖 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 `@crates/broker/src/runtime/api.rs` around lines 821 - 842, Add fail_terminal_sessions_for_worker next to fail_terminal_session in crates/broker/src/runtime/fleet.rs, centralizing session collection, removal, pending-request cleanup, Closed message delivery, and rejection warnings. Replace the inline block in crates/broker/src/runtime/api.rs#L821-L842 with a call using "agent_released" and "terminal worker was released"; replace the corresponding block in crates/broker/src/runtime/maintenance.rs#L288-L307 with a call using "agent_exited" and "terminal worker exited".crates/broker/src/runtime/fleet.rs (1)
180-201: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash # Description: Check whether the CLI attach proxy can re-send terminal.open with the same session_id. set -euo pipefail fd -e ts . packages/cli/src --exec rg -n -C6 'terminal\.open|sessionId|resume_token|reconnect' {} \;Repository: AgentWorkforce/relay
Length of output: 27594
🏁 Script executed:
#!/bin/bash set -euo pipefail printf '%s\n' '--- fleet.rs structure and target region ---' ast-grep outline crates/broker/src/runtime/fleet.rs sed -n '130,245p' crates/broker/src/runtime/fleet.rs printf '%s\n' '--- snapshot/request maintenance and session removal paths ---' rg -n -C8 'terminal_snapshot_requests|terminal_input_requests|handle_maintenance_tick|snapshot_timeout|terminal\.open|TerminalSnapshotRequest|terminal_sessions\.remove' crates/broker/src/runtime/fleet.rs crates/broker/src printf '%s\n' '--- all terminal.open producers and protocol definitions ---' rg -n -C5 'terminal\.open|TerminalOpen|session_id' packages crates -g '*.ts' -g '*.rs' | head -n 500Repository: AgentWorkforce/relay
Length of output: 50377
🏁 Script executed:
#!/bin/bash set -euo pipefail printf '%s\n' '--- terminal event handling ---' rg -n -C12 'Snapshot|snapshot_pty|terminal\.ready|terminal\.output|terminal_snapshot_requests|TerminalSnapshotRequest' \ crates/broker/src/runtime crates/broker/src/worker* crates -g '*.rs' | head -n 450 printf '%s\n' '--- maintenance timeout implementation ---' sed -n '35,75p' crates/broker/src/runtime/maintenance.rs printf '%s\n' '--- terminal control protocol and tests ---' sed -n '1,125p' crates/broker/src/terminal_control.rs rg -n -C8 'snapshot_timeout|terminal.open|duplicate|re-?send|same session|snapshot_pty' crates/broker/src -g '*.rs'Repository: AgentWorkforce/relay
Length of output: 50377
🏁 Script executed:
#!/bin/bash set -euo pipefail printf '%s\n' '--- worker event snapshot branches ---' rg -n -C18 'snapshot_response|request_id.*snapshot|terminal_snapshot_requests|TerminalToCloud::Ready|terminal.ready' \ crates/broker/src/runtime/worker_events.rs printf '%s\n' '--- related worker-event range ---' sed -n '535,760p' crates/broker/src/runtime/worker_events.rsRepository: AgentWorkforce/relay
Length of output: 33801
🏁 Script executed:
#!/bin/bash set -euo pipefail python3 - <<'PY' from dataclasses import dataclass `@dataclass` class Session: ready: bool `@dataclass` class Pending: session_id: str deadline: int # Model the exact HashMap::insert and maintenance behavior in the source. sessions = {} snapshots = {} inputs = {} sessions["s1"] = Session(ready=False) snapshots["R1"] = Pending("s1", deadline=10) # A repeated terminal.open replaces only the session entry. sessions["s1"] = Session(ready=False) snapshots["R2"] = Pending("s1", deadline=20) inputs["I1"] = Pending("s1", deadline=20) # Model handle_maintenance_tick at R1's deadline. for request_id, pending in list(snapshots.items()): if pending.deadline <= 10: del snapshots[request_id] if sessions.pop(pending.session_id, None) is not None: inputs = { key: value for key, value in inputs.items() if value.session_id != pending.session_id } assert "s1" not in sessions assert "R2" in snapshots print("stale snapshot request removes the replacement session:", True) print("replacement snapshot request remains orphaned:", "R2" in snapshots) print("stale input requests are also removed by the timeout path:", not inputs) PYRepository: AgentWorkforce/relay
Length of output: 335
Clear pending requests before replacing a terminal session
A repeated
terminal.openwith the samesession_idreplaces the session but leaves old snapshot and input requests. When the old snapshot expires, maintenance removes the replacement session and emitssnapshot_timeout. Remove all pending requests forsession_idbefore inserting the replacement session.🤖 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 `@crates/broker/src/runtime/fleet.rs` around lines 180 - 201, The terminal-open handling around the terminal_sessions insertion must clear all pending snapshot and input requests for the given session_id before replacing the existing TerminalSession. Update the relevant request collections before self.terminal_sessions.insert, preserving the new session and request initialization behavior.crates/broker/src/runtime/maintenance.rs (1)
43-108: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Reuse
fail_terminal_sessionfor the two timeout sweeps.Both sweeps reproduce the body of
super::fleet::fail_terminal_session: remove the session, filter both request maps, queue a non-finalTerminalToCloud::Error, then queue oneTerminalToCloud::Closed, and warn if the close cannot be queued. The helper is already imported into scope throughsuper::*and ispub(super).Calling it removes about 50 lines and keeps the reserve-capacity ordering in one place.
♻️ Proposed change for the snapshot sweep
for (request_id, session_id) in expired_terminal_snapshots { terminal_snapshot_requests.remove(&request_id); - if terminal_sessions.remove(&session_id).is_some() { - terminal_input_requests.retain(|_, pending| pending.session_id != session_id); - if !try_send_terminal( - terminal_control_tx, - TerminalToCloud::Error { - session_id: session_id.clone(), - code: "snapshot_timeout".into(), - message: "terminal snapshot timed out".into(), - }, - ) { - tracing::warn!(target = "relay_broker::terminal", session_id = %session_id, "terminal queue full or closed while reporting snapshot timeout"); - } - if !try_send_terminal( - terminal_control_tx, - TerminalToCloud::Closed { - session_id: session_id.clone(), - code: Some("snapshot_timeout".into()), - message: Some("terminal snapshot timed out".into()), - }, - ) { - tracing::warn!(target = "relay_broker::terminal", session_id = %session_id, "terminal close could not be queued after snapshot timeout"); - } - } + if terminal_sessions.contains_key(&session_id) { + super::fleet::fail_terminal_session( + terminal_control_tx, + terminal_sessions, + terminal_snapshot_requests, + terminal_input_requests, + session_id, + "snapshot_timeout", + "terminal snapshot timed out".to_string(), + ); + } }Apply the same substitution to the input sweep with
"input_timeout".🤖 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 `@crates/broker/src/runtime/maintenance.rs` around lines 43 - 108, Replace the duplicated timeout cleanup and notification logic in the expired snapshot and input sweeps with calls to the existing fail_terminal_session helper. Pass the corresponding session ID, timeout code ("snapshot_timeout" or "input_timeout"), and timeout message, while preserving removal of each expired request before invoking the helper.crates/broker/src/runtime/worker_events.rs (1)
1115-1117: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Send a close frame when a snapshot fails.
This branch removes the session from
terminal_sessionsbut queues noTerminalToCloud::Closed. Every sibling failure path in this PR emits a close:end_terminal_session(Lines 88-95),fleet::fail_terminal_session, and the snapshot/input timeout sweeps incrates/broker/src/runtime/maintenance.rs. The client therefore receivesterminal.errorbut neverterminal.closed, and it cannot distinguish a recoverable error from a dead session.The pending snapshot and input request entries for this session also remain in
terminal_snapshot_requestsandterminal_input_requests; only the session map entry is removed here.🐛 Proposed fix
} else if snapshot_failed { - terminal_sessions.remove(&session_id); + end_terminal_session( + terminal_control_tx, + terminal_sessions, + terminal_snapshot_requests, + terminal_input_requests, + &session_id, + "snapshot_failed", + "terminal snapshot failed", + ); }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.} else if snapshot_failed { end_terminal_session( terminal_control_tx, terminal_sessions, terminal_snapshot_requests, terminal_input_requests, &session_id, "snapshot_failed", "terminal snapshot failed", ); }🤖 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 `@crates/broker/src/runtime/worker_events.rs` around lines 1115 - 1117, Update the snapshot_failed branch in the terminal session event handling to enqueue TerminalToCloud::Closed before removing the session. Also remove or otherwise clean up the session’s pending entries in terminal_snapshot_requests and terminal_input_requests, matching the cleanup and close-frame behavior used by end_terminal_session and the maintenance timeout paths.crates/broker/src/terminal_control.rs (1)
187-218: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash # Description: Check whether other broker websocket clients apply keepalive pings or read timeouts. set -euo pipefail fd -e rs . crates/broker/src --exec rg -n -C4 'Message::Ping|Message::Pong|tungstenite|WebSocketConfig|read_timeout|keepalive|ping_interval' {} \;Repository: AgentWorkforce/relay
Length of output: 6630
🏁 Script executed:
#!/bin/bash set -euo pipefail printf '%s\n' '--- terminal_control.rs outline ---' ast-grep outline crates/broker/src/terminal_control.rs printf '%s\n' '--- terminal control implementation ---' sed -n '1,245p' crates/broker/src/terminal_control.rs printf '%s\n' '--- terminal session lifecycle references ---' rg -n -C5 'terminal_sessions|TerminalControlEvent::Disconnected|TerminalControlEvent::Message|TerminalControlCommand' crates/broker/src printf '%s\n' '--- websocket construction and configuration ---' rg -n -C8 'connect_async|WebSocketConfig|set_config|accept_async|terminal' crates/broker/src/terminal_control.rs crates/broker/srcRepository: AgentWorkforce/relay
Length of output: 50377
🏁 Script executed:
#!/bin/bash set -euo pipefail printf '%s\n' '--- terminal_control.rs ---' cat -n crates/broker/src/terminal_control.rs | sed -n '1,235p' printf '%s\n' '--- terminal lifecycle handlers ---' rg -n -C8 'fn handle_terminal_control_event|TerminalControlEvent::Disconnected|terminal_sessions\.remove|terminal_sessions\.retain|try_send_terminal|fail_terminal_session' crates/broker/src/runtime printf '%s\n' '--- terminal websocket server and heartbeat references ---' rg -n -C6 'node/terminal/ws|terminal.*ws|ping_interval|Message::Ping|WebSocketConfig|read_timeout|keepalive' --glob '*.rs' --glob '!crates/broker/src/terminal_control.rs' .Repository: AgentWorkforce/relay
Length of output: 50376
🏁 Script executed:
#!/bin/bash set -euo pipefail printf '%s\n' '--- terminal transport disconnect handling ---' cat -n crates/broker/src/runtime/fleet.rs | sed -n '135,180p' printf '%s\n' '--- terminal endpoint files ---' rg -l 'node/terminal/ws|terminal.*websocket|terminal.*WebSocket|terminal.*ping|ping.*terminal' --glob '*.rs' . printf '%s\n' '--- all explicit websocket heartbeat and timeout code ---' rg -n 'ping_interval|Message::Ping|Message::Pong|WebSocketConfig|read_timeout|write_timeout|keepalive|set_config' --glob '*.rs' . | grep -E 'terminal|node_control|fleet|ws|socket' | head -n 250 printf '%s\n' '--- dependency versions ---' rg -n -C2 'tokio-tungstenite|tungstenite' Cargo.toml Cargo.lock crates/*/Cargo.tomlRepository: AgentWorkforce/relay
Length of output: 5513
🏁 Script executed:
#!/bin/bash set -euo pipefail python3 - <<'PY' from pathlib import Path import re terminal = Path("crates/broker/src/terminal_control.rs").read_text() fleet = Path("crates/broker/src/runtime/fleet.rs").read_text() connected = terminal[terminal.index("while connected"):terminal.index("\n }\n let _ = event_tx.send(TerminalControlEvent::Disconnected)", terminal.index("while connected"))] checks = { "uses_default_connect_async": "connect_async(request)" in terminal, "has_ping_interval": bool(re.search(r"ping_interval|interval\s*\(", connected)), "has_read_or_write_timeout": bool(re.search(r"read_timeout|write_timeout|timeout\s*\(", connected)), "handles_inbound_ping_with_no_send": bool(re.search( r"Some\(Ok\(Message::Ping\(_\)\)\)\s*=>\s*\{\s*\}", connected )), "emits_disconnected_after_loop": "TerminalControlEvent::Disconnected" in terminal, "clears_terminal_sessions_on_disconnect": bool(re.search( r"TerminalControlEvent::Disconnected.*?terminal_sessions\.clear\(\)", fleet, re.S )), } for name, value in checks.items(): print(f"{name}={value}") assert checks["uses_default_connect_async"] assert not checks["has_ping_interval"] assert not checks["has_read_or_write_timeout"] assert checks["handles_inbound_ping_with_no_send"] assert checks["emits_disconnected_after_loop"] assert checks["clears_terminal_sessions_on_disconnect"] PYRepository: AgentWorkforce/relay
Length of output: 364
🏁 Script executed:
#!/bin/bash set -euo pipefail python3 - <<'PY' from pathlib import Path import re terminal = Path("crates/broker/src/terminal_control.rs").read_text() fleet = Path("crates/broker/src/runtime/fleet.rs").read_text() connected = terminal[terminal.index("while connected"):terminal.index( "\n }\n let _ = event_tx.send(TerminalControlEvent::Disconnected)", terminal.index("while connected"), )] checks = { "uses_default_connect_async": "connect_async(request)" in terminal, "has_ping_interval": bool(re.search(r"ping_interval|interval\s*\(", connected)), "has_read_or_write_timeout": bool(re.search(r"read_timeout|write_timeout|timeout\s*\(", connected)), "handles_inbound_ping_with_no_send": bool(re.search( r"Some\(Ok\(Message::Ping\(_\)\)\)\s*=>\s*\{\s*\}", connected )), "emits_disconnected_after_loop": "TerminalControlEvent::Disconnected" in terminal, "clears_terminal_sessions_on_disconnect": bool(re.search( r"TerminalControlEvent::Disconnected.*?terminal_sessions\.clear\(\)", fleet, re.S )), } for name, value in checks.items(): print(f"{name}={value}") assert checks["uses_default_connect_async"] assert not checks["has_ping_interval"] assert not checks["has_read_or_write_timeout"] assert checks["handles_inbound_ping_with_no_send"] assert checks["emits_disconnected_after_loop"] assert checks["clears_terminal_sessions_on_disconnect"] PYRepository: AgentWorkforce/relay
Length of output: 364
Add terminal WebSocket liveness detection.
connect_asyncuses no ping interval or read timeout, and theMessage::Ping(_)arm performs no liveness check. A half-open socket can keepconnectedtrue indefinitely, soBrokerRuntimeretainsterminal_sessionsand queues outbound frames until the bounded lane fills. Add a keepalive or read deadline that closes the socket when liveness fails.🤖 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 `@crates/broker/src/terminal_control.rs` around lines 187 - 218, Add WebSocket liveness handling to the terminal connection loop around connect_async and the tokio::select! in the terminal control task. Configure a keepalive/read-timeout mechanism and ensure missed liveness responses or expired reads set connected to false and close the socket, rather than leaving the session active indefinitely; preserve the existing reconnect and Disconnected-event flow.packages/cli/src/cli/lib/attach-fleet-node.ts (3)
413-420: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Encode input bytes directly instead of through a UTF-8 string round-trip.
rawDataToStringdecodes the frame as UTF-8, andBuffer.from(raw, 'utf8')re-encodes it. Any byte sequence that is not valid UTF-8 is replaced with U+FFFD during the decode step and cannot be recovered. PTY input can carry raw bytes, so keystrokes in a non-UTF-8 locale reach the remote terminal corrupted.Convert the raw frame to a
Bufferonce and base64-encode that buffer.🐛 Proposed fix
- const raw = rawDataToString(data); remote.send( JSON.stringify({ type: 'terminal.input', session_id: sessionId, - data_base64: Buffer.from(raw, 'utf8').toString('base64'), + data_base64: rawDataToBuffer(data).toString('base64'), }) );Add the helper next to
rawDataToString:function rawDataToBuffer(data: WebSocket.RawData): Buffer { if (Buffer.isBuffer(data)) return data; if (data instanceof ArrayBuffer) return Buffer.from(data); if (Array.isArray(data)) return Buffer.concat(data); return Buffer.from(String(data), 'utf8'); }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.remote.send( JSON.stringify({ type: 'terminal.input', session_id: sessionId, data_base64: rawDataToBuffer(data).toString('base64'), }) );🤖 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 `@packages/cli/src/cli/lib/attach-fleet-node.ts` around lines 413 - 420, Replace the rawDataToString-based encoding in the terminal.input handling with a rawDataToBuffer helper that preserves WebSocket frame bytes across Buffer, ArrayBuffer, array, and fallback inputs, then base64-encode that Buffer directly for remote.send.
438-448: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Close the listener when startup fails after
server.listen.The HTTP server is already listening at Line 436. Two later paths throw without closing it:
- Line 439: the
addressguard throwsFleetNodeAttachError.- Line 445:
new URL(terminalUrl)throws aTypeErrorif the broker returns a malformedterminal_url.In both cases the caller never receives a
FleetNodeAttachProxy, soproxy.close()inattachFleetNode(packages/cli/src/cli/commands/local-agent.tsLine 147) never runs. The listening socket keeps the event loop alive and the CLI does not exit.The
TypeErrorat Line 445 also escapes the structured error contract.local-agent.tsLines 614-618 prefixes it withError:but loses the code.🐛 Proposed fix
const address = server.address(); - if (!address || typeof address === 'string') - throw new FleetNodeAttachError( - 'Error: could not allocate loopback terminal listener.', - 'loopback_unavailable' - ); - - const resumeUrl = new URL(terminalUrl); + const abortStartup = async (error: FleetNodeAttachError): Promise<never> => { + websocketServer.close(); + await new Promise<void>((resolve) => server.close(() => resolve())); + throw error; + }; + if (!address || typeof address === 'string') + await abortStartup( + new FleetNodeAttachError( + 'Error: could not allocate loopback terminal listener.', + 'loopback_unavailable' + ) + ); + + let resumeUrl: URL; + try { + resumeUrl = new URL(terminalUrl); + } catch { + await abortStartup( + new FleetNodeAttachError('Error: the fleet node returned an invalid terminal URL.', 'node_unreachable') + ); + }The
addressnarrowing afterabortStartupneeds a non-null assertion or a restructure, because TypeScript does not inferneverthrough the awaited call.🤖 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 `@packages/cli/src/cli/lib/attach-fleet-node.ts` around lines 438 - 448, Update the startup error handling in attachFleetNode so every failure after server.listen, including the address guard and new URL(terminalUrl), closes the listener before throwing. Convert malformed terminal_url errors into the existing structured FleetNodeAttachError contract with the appropriate code, and preserve TypeScript narrowing around abortStartup using a non-null assertion or equivalent restructure.
575-592: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
ws WebSocket terminate discards queued send payload documentation💡 Result:
In the ws library, the websocket.terminate method does not automatically discard queued send payloads [1]. When terminate is called, it forcibly destroys the underlying Node.js socket [2][3][4]. However, because the library operates on top of Node.js streams, any data already buffered in the stream or the internal send queue is not immediately discarded by design [1]. Instead, the underlying socket destroys the connection, and the buffered data is typically processed or read until the 'close' event is emitted [1]. If you have pending send operations, their associated callbacks may still be invoked (often with an error once the connection is closed) as the library cleans up the remaining state [5]. Developers seeking to prevent the processing of queued data must manage this manually, for example by ensuring no further logic executes or by removing event listeners if you want to ignore subsequent events after the termination call [1].
Citations:
- 1: websockets/ws#2209
- 2: https://github.com/websockets/ws/blob/HEAD/doc/ws.md
- 3: https://github.com/websockets/ws/blob/8.8.1/doc/ws.md
- 4: https://stackoverflow.com/questions/41074052/how-to-terminate-a-websocket-connection
- 5: websockets/ws#1226
🏁 Script executed:
#!/bin/bash set -eu printf '%s\n' '--- relevant source ---' sed -n '540,610p' packages/cli/src/cli/lib/attach-fleet-node.ts printf '%s\n' '--- minimum Node declarations and package metadata ---' rg -n '"engines"|node:|`@types/node`|closeAllConnections|closeIdleConnections|NODE_VERSION|node-version' package.json packages/cli package-lock.json pnpm-lock.yaml yarn.lock .nvmrc .node-version 2>/dev/null || true printf '%s\n' '--- server lifecycle context ---' sed -n '110,165p' packages/cli/src/cli/commands/local-agent.ts printf '%s\n' '--- closeSocket definition and call sites ---' rg -n -A35 -B10 'function closeSocket|const closeSocket|closeSocket\(' packages/cli/src/cli/lib/attach-fleet-node.tsRepository: AgentWorkforce/relay
Length of output: 50376
🏁 Script executed:
#!/bin/bash set -eu printf '%s\n' '--- root package engines ---' sed -n '136,152p' package.json printf '%s\n' '--- CLI package engines ---' sed -n '62,78p' packages/cli/package.json printf '%s\n' '--- supported Node version ---' sed -n '1,120p' packages/cli/src/cli/lib/node-version.ts printf '%s\n' '--- closeSocket implementation ---' rg -n -A28 -B8 '^(async )?(function|const) closeSocket|closeSocket\s*=' packages/cli/src/cli/lib/attach-fleet-node.ts printf '%s\n' '--- HTTP server and socket tracking ---' rg -n -A12 -B8 'eventSockets|inputSockets|websocketServer|server\.on|upgrade|request' packages/cli/src/cli/lib/attach-fleet-node.tsRepository: AgentWorkforce/relay
Length of output: 18018
🌐 Web query:
Node.js http.Server closeAllConnections added version server.close callback active connections documentation💡 Result:
The method
server.closeAllConnections()was added to the Node.jshttp.Serverclass in version v18.2.0 [1][2][3]. ### Overview ofserver.closeAllConnections()This method is a forceful way to terminate all established HTTP(S) connections connected to the server, including those that are currently active (e.g., sending a request or waiting for a response) [1][4]. It does not destroy sockets that have been upgraded to other protocols, such as WebSockets or HTTP/2 [1][5]. ### Relationship withserver.close()and Callbacks Theserver.close([callback])method is responsible for stopping the server from accepting new connections [6][7]. Historically,server.close()would wait for existing connections to finish naturally before the server would fully close and fire itscloseevent (and the associatedcallback) [8][7]. Becauseserver.closeAllConnections()is a forceful, synchronous termination of connections, it is often used in conjunction withserver.close()to ensure a server shuts down in a timely manner [1][9]. Official documentation recommends the following: * If you useserver.closeAllConnections()withserver.close(), it is recommended to callcloseAllConnections()afterserver.close()to avoid race conditions where new connections might be established between the two calls [1][4]. * Theserver.close()callback will fire once the server has stopped accepting new connections and all remaining connections have been terminated (whether they closed naturally or were forcibly closed bycloseAllConnections) [4][6][8]. ### Additional Context *server.closeIdleConnections(): Also added in v18.2.0, this method closes only those connections that are not actively sending or waiting for a request [10][4]. As of Node.js v19.0.0,server.close()automatically handles the closing of idlekeep-aliveconnections, makingserver.closeIdleConnections()less necessary for standard graceful shutdowns [4][6]. * Best Practice: For a graceful shutdown, many developers stop accepting new connections withserver.close(), allow a grace period for active requests to finish, and then useserver.closeAllConnections()to forcibly close any remaining stubborn connections [9].Citations:
- 1: https://nodejs.org/docs/latest-v24.x/api/http.html
- 2: https://nodejs.org/download/release/v18.10.0/docs/api/http.html
- 3: https://nodejs.org/download/release/v23.6.1/docs/api/http.html
- 4: https://github.com/nodejs/node/blob/main/doc/api/http.md
- 5: https://beta.docs.nodejs.org/http.html
- 6: https://nodejs.org/api/http.html
- 7: https://beta.docs.nodejs.org/net.html
- 8: https://nodejs.org/dist/latest/docs/api/net.html
- 9: https://stackoverflow.com/questions/78144749/how-to-stop-accepting-new-requests-before-closing-existing-ones
- 10: nodejs/node#42812
🌐 Web query:
ws 8.18.3 WebSocket terminate source _socket destroy Sender queue send callback💡 Result:
In the ws library, including version 8.18.3, the websocket.terminate method is designed to forcibly close the connection by internally calling socket.destroy on the underlying Node.js network socket [1][2]. Regarding the behavior of the sender queue and send callbacks when terminate is invoked: 1. Immediate Destruction: Calling terminate immediately initiates the destruction of the underlying socket [1][3][4]. Because this does not perform a graceful closing handshake, the connection is dropped abruptly [4]. 2. Sender Queue and Callbacks: When the socket is destroyed, any data remaining in the sender's queue or currently being processed (e.g., compressed or framed) may not be successfully written to the network [5][6]. If a send operation is in progress or queued, the associated callback may not be executed as expected because the underlying stream is destroyed [5]. Specifically, if the socket is "corked" or a write is pending when the socket is destroyed, the write callback will often not be invoked, or it may receive an error depending on the exact timing and Node.js stream state (e.g., ERR_STREAM_DESTROYED) [5]. 3. Historical Context: The library has evolved to address issues where send callbacks were not properly handled or where resources were wasted attempting to process data after a closure [6]. While the library has been improved to clear the sender queue more effectively in various scenarios [6], terminate remains an "immediate" operation that prioritizes the cessation of the connection over the completion of pending send callbacks [3][7][4]. If you are experiencing issues where callbacks are not firing after a terminate call, this is expected behavior due to the nature of forcing a socket destruction [5][7]. For graceful handling of pending data, the.close method should be used instead, as it allows for the completion of the closing handshake [3][4].
Citations:
- 1: https://github.com/websockets/ws/blob/HEAD/doc/ws.md
- 2: websockets/ws#1033
- 3: https://www.npmjs.com/package/ws
- 4: websockets/ws#1142
- 5: websockets/ws#2214
- 6: websockets/ws#1226
- 7: websockets/ws#2209
🏁 Script executed:
#!/bin/bash set -eu printf '%s\n' '--- ws dependency version ---' rg -n -m5 '"ws"|"node_modules/ws"' package.json packages/cli/package.json package-lock.json printf '%s\n' '--- ws 8.18.3 terminate and send implementation ---' curl -fsSL https://raw.githubusercontent.com/websockets/ws/8.18.3/lib/websocket.js | awk 'NR>=300 && NR<=360 {print NR ":" $0} NR>=440 && NR<=470 {print NR ":" $0}' curl -fsSL https://raw.githubusercontent.com/websockets/ws/8.18.3/lib/sender.js | awk 'NR>=100 && NR<=180 {print NR ":" $0} NR>=560 && NR<=640 {print NR ":" $0}' printf '%s\n' '--- local-agent cleanup context ---' sed -n '130,155p' packages/cli/src/cli/commands/local-agent.tsRepository: AgentWorkforce/relay
Length of output: 7588
🏁 Script executed:
#!/bin/bash set -eu curl -fsSL https://raw.githubusercontent.com/websockets/ws/8.18.3/lib/websocket.js | nl -ba | rg -A18 -B8 'terminate\(\)|function sendAfterClose|function setCloseTimer' curl -fsSL https://raw.githubusercontent.com/websockets/ws/8.18.3/lib/sender.js | nl -ba | rg -A35 -B8 'send\(data|close\(code|enqueue\(|dequeue\(' printf '%s\n' '--- close behavior in the installed Node API declarations, if present ---' rg -n -A8 -B8 'closeAllConnections' node_modules/@types/node/http.d.ts 2>/dev/null || trueRepository: AgentWorkforce/relay
Length of output: 268
🏁 Script executed:
#!/bin/bash set -eu printf '%s\n' '--- ws terminate and sendAfterClose ---' curl -fsSL https://raw.githubusercontent.com/websockets/ws/8.18.3/lib/websocket.js | awk '/terminate\(\)/ || /function sendAfterClose/ { for (i=NR-8; i<=NR+18; i++) wanted[i]=1 } { lines[NR]=$0 } END { for (i=1; i<=NR; i++) if (wanted[i]) print i ":" lines[i] }' printf '%s\n' '--- ws Sender queue and write callbacks ---' curl -fsSL https://raw.githubusercontent.com/websockets/ws/8.18.3/lib/sender.js | awk '/send\(data/ || /enqueue\(/ || /dequeue\(/ || /close\(code/ { for (i=NR-5; i<=NR+32; i++) wanted[i]=1 } { lines[NR]=$0 } END { for (i=1; i<=NR; i++) if (wanted[i]) print i ":" lines[i] }'Repository: AgentWorkforce/relay
Length of output: 13620
🏁 Script executed:
#!/bin/bash set -eu printf '%s\n' '--- Node HTTP close documentation ---' curl -fsSL https://raw.githubusercontent.com/nodejs/node/v22.0.0/doc/api/http.md | awk '/### server.closeAllConnections\(\)/ || /### server.close\(\[callback\]\)/ { for (i=NR-2; i<=NR+35; i++) wanted[i]=1 } { lines[NR]=$0 } END { for (i=1; i<=NR; i++) if (wanted[i]) print i ":" lines[i] }' printf '%s\n' '--- Node minimum-version enforcement and relevant tests ---' rg -n -A20 -B10 'MINIMUM_NODE_MAJOR|assertSupportedNodeVersion|close\(\)' packages/cli/src/cli/lib/node-version.test.ts packages/cli/src/cli/lib/attach-fleet-node.test.ts packages/cli/src/cli/commands/local-agent.test.ts 2>/dev/null || trueRepository: AgentWorkforce/relay
Length of output: 1702
Wait for the close payload before terminating the remote socket.
ws@8.18.3implementsterminate()with_socket.destroy(). Theterminal.closeframe can therefore be lost whenterminate()followssend()immediately. Wait for the send callback or a short timeout before terminating.
server.closeAllConnections()is available under the Node.js 22 minimum, but it does not close upgraded WebSocket connections. If used for active HTTP requests, callserver.close()beforeserver.closeAllConnections().🤖 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 `@packages/cli/src/cli/lib/attach-fleet-node.ts` around lines 575 - 592, Update the cleanup flow around activeRemote so termination waits for the terminal.close send callback or a short timeout, ensuring the close payload is flushed before activeRemote.terminate(). Preserve best-effort behavior and existing socket/server cleanup; if adding server.closeAllConnections() for active HTTP requests, invoke it only after server.close().
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
All reported issues were addressed
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/broker/src/runtime/worker_events.rs (1)
1114-1116: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winClose the session after a snapshot error.
When
snapshot_failedis true, callend_terminal_sessioninstead of removing onlyterminal_sessions. TheErrorframe was already queued. The helper also clears both pending request maps and queuesTerminalToCloud::Closed. Without it, clients do not receiveClosed, and pending input requests remain until timeout.🤖 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 `@crates/broker/src/runtime/worker_events.rs` around lines 1114 - 1116, Update the snapshot_failed branch in the worker event handling flow to call end_terminal_session instead of removing the session directly from terminal_sessions. Preserve the existing session_id and queued Error frame, and ensure the helper performs cleanup and queues TerminalToCloud::Closed.
🤖 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.
Outside diff comments:
In `@crates/broker/src/runtime/worker_events.rs`:
- Around line 1114-1116: Update the snapshot_failed branch in the worker event
handling flow to call end_terminal_session instead of removing the session
directly from terminal_sessions. Preserve the existing session_id and queued
Error frame, and ensure the helper performs cleanup and queues
TerminalToCloud::Closed.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 3cdeb749-db95-4eee-884a-811691bbe680
📒 Files selected for processing (1)
crates/broker/src/runtime/worker_events.rs
|
Follow-up review fixes are in |
There was a problem hiding this comment.
All reported issues were addressed across 2 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
|
Final Cubic follow-up is in |
|
Final verification is green at |
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
crates/broker/src/runtime/api.rs (1)
2599-2617: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftAdd a regression test for timeout after queue admission.
These tests verify only timeout conversion. Add a test that admits a model command, delays writer completion beyond the requested deadline, and asserts the API result and writer state. This test should lock the accepted-command contract.
🤖 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 `@crates/broker/src/runtime/api.rs` around lines 2599 - 2617, Add a regression test in set_model_timeout_tests that admits a model command, delays writer completion beyond the requested timeout, and verifies both the API result and resulting writer state. Exercise the full command path rather than only set_model_write_timeout conversion, preserving the accepted-command contract when completion exceeds the deadline.
🤖 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 `@crates/broker/src/runtime/api.rs`:
- Around line 759-770: Update the set_model timeout handling around
send_raw_to_worker so a timeout after WorkerWriteCommand admission does not
return an HTTP 500 or cause clients to retry the command. Propagate cancellation
through the writer so its write_all and flush stop safely, or return an explicit
accepted/pending response once admission has occurred; preserve failure
reporting only when admission itself fails.
---
Nitpick comments:
In `@crates/broker/src/runtime/api.rs`:
- Around line 2599-2617: Add a regression test in set_model_timeout_tests that
admits a model command, delays writer completion beyond the requested timeout,
and verifies both the API result and resulting writer state. Exercise the full
command path rather than only set_model_write_timeout conversion, preserving the
accepted-command contract when completion exceeds the deadline.
🪄 Autofix
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: CHILL
Plan: Pro Plus
Run ID: 57fd7c12-ca4b-4d8a-966c-25846dfac7db
📒 Files selected for processing (3)
crates/broker/src/runtime/api.rscrates/broker/src/runtime/fleet.rscrates/broker/src/runtime/maintenance.rs
🚧 Files skipped from review as they are similar to previous changes (1)
- crates/broker/src/runtime/fleet.rs
There was a problem hiding this comment.
All reported issues were addressed across 3 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic
2 participants
Summary
Verification
cargo test -p agent-relay-broker --no-fail-fastcargo clippy -p agent-relay-broker -- -D warningscargo fmt --checkgit diff --check