feat(terminal): add opt-in flow control - #67
Conversation
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 424c2da81e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if state.shutdown.load(Ordering::Acquire) { | ||
| return; | ||
| } | ||
| queue.pop() |
There was a problem hiding this comment.
On Unix, once the queue exceeds the reservation threshold, terminal_bridge.rs:110-120 stops polling the terminal descriptor. This writer later pops packets and frees capacity here but never signals the bridge wake pipe, so a connected backpressure session resumes reading terminal output only when an unrelated client or forwarding event occurs—typically the next keepalive, up to five seconds later. Repeating that pause whenever the 64 KiB queue fills can reduce a continuously producing session to roughly one queueful per keepalive rather than the available network rate; signal the bridge whenever a pop makes terminal capacity available.
Useful? React with 👍 / 👎.
| let Ok(mut queue) = state.wake.wait_while(queue, |queue| { | ||
| !state.shutdown.load(Ordering::Acquire) | ||
| && (!state.connected.load(Ordering::Acquire) || queue.bytes() == 0) |
There was a problem hiding this comment.
When an active flow-controlled session queues a valid header-only packet, such as via the public RuntimeHandle::send_packet(id, header, b""), OutputQueue::bytes() remains zero even though the packet deque is nonempty. This predicate therefore keeps the writer asleep and the packet is not delivered until some later nonempty packet is queued; a stream of empty control packets can also grow the deque without limit despite the advertised bounded queue. Expose and test actual queue emptiness instead of using the payload-byte count as the condition.
Useful? React with 👍 / 👎.
| let keep_from = packet_bytes - self.limit; | ||
| packet = Packet::new(packet.header(), &packet.payload()[keep_from..]); |
There was a problem hiding this comment.
In discard mode, a single TERMINAL_BUFFER payload larger than 64 KiB can reach this queue through the public RuntimeHandle::send_packet API, while the encrypted network framing accepts messages far larger than that. The packet payload is an encoded TerminalBuffer protobuf, so removing bytes from its front usually removes the field tag or part of its length prefix; client_terminal.rs:192-197 then fails to decode the output and terminates the client instead of displaying the newest tail. Decode and re-encode the buffer field, or split the message at the terminal-byte level, rather than slicing the serialized protobuf.
Useful? React with 👍 / 👎.
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a4aeb38570
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| pub(crate) fn finish_terminal(&self) -> Result<(), SessionError> { | ||
| self.shutdown.store(true, Ordering::Release); | ||
| self.stop_flow_writer(); |
There was a problem hiding this comment.
When a flow-controlled terminal exits while output remains queued—especially on a slow link—the bridge returns after reading the terminal HUP and runtime_handler.rs calls finish_terminal, which sets the flow writer's shutdown flag here. run_flow_writer then exits without draining the queue, so the client's final command output or shell-exit message is silently lost. Graceful terminal completion needs to let the queued packets drain before stopping the writer and closing the TCP write side.
Useful? React with 👍 / 👎.
| Err(SessionError::Connection(ConnError::Backpressure)) => { | ||
| if let Ok(mut queue) = state.queue.lock() { | ||
| queue.push_front(packet); |
There was a problem hiding this comment.
If the writer pops a packet, the client is marked disconnected before it acquires the connection lock, and another sender refills the newly freed capacity, this unconditional push_front reinserts the in-flight packet on top of a full queue. The byte count can therefore exceed the advertised 64 KiB limit by as much as another full-size packet, breaking the hard per-session bound precisely during disconnect/backpressure races; reinsertion needs to reserve capacity atomically or apply the mode's eviction/backpressure policy.
Useful? React with 👍 / 👎.
| queue | ||
| .push(Packet::new(header, payload)) | ||
| .map_err(|_| SessionError::Connection(ConnError::Backpressure))?; |
There was a problem hiding this comment.
When RuntimeHandle::send_packet fills the flow queue after terminal_bridge.rs:110 checks can_buffer_write but before the terminal packet reaches this push, the push returns Backpressure. The terminal path at terminal_bridge.rs:142-150 treats that error as fatal rather than retaining the packet like send_or_hold does for forwarding traffic, so concurrent server-originated traffic can tear down an otherwise healthy terminal session. Make the capacity check and enqueue atomic or return the terminal packet to the bridge for retry.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 96f511e745
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| Err(error) => { | ||
| crate::diag::info(format!("flow-control writer stopped: {error}")); | ||
| state.stop = StopMode::Hard; |
There was a problem hiding this comment.
When an opted-in session's live write times out or encounters a connection reset, PreparedWrite::send returns ConnError::Io, and this branch permanently changes the flow state to Hard. Although the connection is then soft-disconnected for recovery, RecoverPermit::drop only calls resume and never resets stop, so the writer remains dead and the next terminal packet is rejected with SessionError::Unavailable, ending the recovered session. Treat client transport failures as a disconnected state that waits for recovery rather than a hard writer failure.
Useful? React with 👍 / 👎.
| if let Some(state) = &self.flow_control { | ||
| return state.enqueue(Packet::new(header, payload)); |
There was a problem hiding this comment.
When an opted-in session also has a continuously busy port forward, this routes forwarding packets through the same FIFO as terminal output. Forwarding packets are never evictable, while OutputQueue::push in discard mode may evict terminal buffers to admit them; additionally, terminal_bridge retries a held outbound forwarding packet before checking the terminal reservation. A sustained tunnel can therefore keep the queue above the terminal admission threshold indefinitely, so backpressure stops reading shell output and discard removes prompt output instead of keeping the display current. Give terminal output reserved capacity or scheduling priority separate from lossless forwarding traffic.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 73fb73faac
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| self.disconnect(); | ||
| } | ||
| Ok(()) | ||
| let stream = self.stream.try_clone().map_err(ConnError::Io)?; |
There was a problem hiding this comment.
When try_clone() fails, for example because the process temporarily hits its file-descriptor limit, writer.write_packet() has already advanced the encryption nonce and sequence and retained a frame that was never sent. A later retry on the same connection—such as flush_recover_hold() being called again from RecoverPermit::drop—encrypts the retried payload with the next nonce, so the peer observes a missing nonce and rejects the next frame with a bad MAC. Clone the stream before mutating BackedWriter, or disconnect/restore its state when cloning fails.
Useful? React with 👍 / 👎.
| // installs the candidate (or safely abandons it). Terminal EOF | ||
| // must not bypass that pause and drain queued output onto the old | ||
| // stream; RecoverPermit::drop resumes the writer atomically. | ||
| state.stop = StopMode::Graceful; |
There was a problem hiding this comment.
When terminal EOF races with an explicit session/server shutdown, shutdown() can set StopMode::Hard while finish_terminal() subsequently calls this method and changes it back to Graceful. The writer may then drain queued packets over a slow or blocked transport instead of terminating, delaying the hard shutdown by one or more live-write timeouts while the shutdown caller waits in join(). Make stop transitions monotonic so Graceful is only installed while the state is still Running.
Useful? React with 👍 / 👎.
| return; | ||
| } | ||
| let active = match ActiveSession::new(connection, &terminal) { | ||
| let active = match ActiveSession::new(connection, &terminal, payload.flowcontrol) { |
There was a problem hiding this comment.
For --flow-control backpressure sessions routed through a jumphost, this enables the outer server queue, but terminal_jump::run never calls minimize_terminal_output_buffering after decoding the relayed flowcontrol value. Consequently the jump process can place the destination's output into its default-sized local socket buffer before the outer 64 KiB queue applies backpressure, leaving prompts behind substantially more buffered output and defeating the bounded-latency behavior on the jumphost path. Apply the same terminal-side socket bound used by terminal_pty::run when the relayed mode is non-NONE.
Useful? React with 👍 / 👎.
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com> Co-authored-by: OmO <noreply@openai.com> Co-authored-by: GPT-5.6 <noreply@openai.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ae0e50d900
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if let Some(worker) = self.worker.take() { | ||
| let _ = worker.join(); |
There was a problem hiding this comment.
When an opted-in client writes to a full pipe or an indefinitely stalled terminal, the worker can block inside write_all; if the remote session then ends or stdin closes, dropping ConsoleOutput waits on this join() forever because setting stopping cannot interrupt the blocking write. This leaves the client unable to exit cleanly, so shutdown needs a cancellable/nonblocking output path or must not synchronously join a potentially blocked writer.
Useful? React with 👍 / 👎.
| if let Err(error) = writer.write_all(&bytes).and_then(|()| writer.flush()) { | ||
| if let Ok(mut state) = shared.state.lock() { | ||
| state.error = Some(error); | ||
| state.stopping = true; | ||
| shared.wake.notify_all(); | ||
| } | ||
| #[cfg(unix)] | ||
| signal_capacity(capacity_signal); |
There was a problem hiding this comment.
If the worker gets BrokenPipe or another write error after accepting the last terminal packet, it stores the error and signals capacity, but the pump merely drains that signal and never inspects state.error unless a later terminal buffer is routed through try_write. A quiet shell can therefore leave the client running indefinitely after stdout has failed; the capacity/error wake should also propagate the worker error immediately.
Useful? React with 👍 / 👎.
| let concurrent = std::mem::take(&mut *hold); | ||
| hold.push((header, payload)); | ||
| hold.extend(remaining); |
There was a problem hiding this comment.
For a non-flow-controlled session, if the newly recovered transport fails while flush_recover_hold sends output accumulated during the handshake, Connection::write_packet now returns the live-send error only after BackedWriter has encrypted and retained that packet for replay. Re-inserting the same plaintext here makes RecoverPermit::drop buffer it again on the disconnected writer under a second sequence, so the next recovery delivers duplicate terminal or port-forward data and can corrupt the forwarded stream; only the unwritten tail should be restored when replay already owns the failed packet.
Useful? React with 👍 / 👎.
| if !output(&bytes)? { | ||
| return Ok(DisplayOutcome::Pending(packet)); | ||
| } | ||
| terminal_modes.observe(&bytes); | ||
| io::stdout() | ||
| .lock() | ||
| .write_all(&bytes) | ||
| .and_then(|()| io::stdout().lock().flush()) | ||
| .map_err(|error| terminal_io("writing terminal output", error))?; | ||
| Ok(contains_cursor_report_request(&bytes)) | ||
| Ok(DisplayOutcome::Displayed { | ||
| cursor_report: contains_cursor_report_request(&bytes), |
There was a problem hiding this comment.
In discard mode, output only admits these bytes to a queue, yet observe immediately records their alternate-screen transitions. A later enqueue can evict that queued packet before the worker displays it, leaving alternate_screen inconsistent with the emulator; for example, an evicted CSI ? 1049 h makes shutdown send an unmatched alternate-screen leave, while an evicted leave can strand the user in the alternate screen. Mode tracking must follow bytes actually written or be updated when queued bytes are discarded.
Useful? React with 👍 / 👎.
| Err(packet) => { | ||
| pending = Some(packet); | ||
| state = self | ||
| .wake | ||
| .wait(state) | ||
| .map_err(|_| SessionError::Unavailable)?; |
There was a problem hiding this comment.
When a non-terminal packet's framed cost exceeds the 64 KiB control-lane limit, OutputQueue::push always returns it even when the queue is empty, but this branch treats every rejection as temporary and waits on the condition variable before retrying forever. Thus an opted-in session can permanently block RuntimeHandle::send_packet with an oversized control payload, and a maximum-sized relayed packet can similarly pin the bridge; detect the wanted > limit case and return Backpressure or another size error instead of waiting for capacity that cannot exist.
Useful? React with 👍 / 👎.
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com> Co-authored-by: OmO <noreply@openai.com> Co-authored-by: GPT-5.6 <noreply@openai.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: f586a57665
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| Ok(_) if descriptors[0].revents().contains(PollFlags::OUT) => { | ||
| return rustix::io::write(&self.file, &bytes[..bytes.len().min(4096)]) | ||
| .map_err(io::Error::from); | ||
| } | ||
| Ok(_) => {} |
There was a problem hiding this comment.
When an opted-in Unix client has stdout connected to a pipe whose reader exits, poll reports ERR/HUP on the output descriptor rather than OUT. This loop handles neither event, so it immediately polls again without attempting a write, never records BrokenPipe, and can leave the client spinning at 100% CPU instead of exiting. Treat output error/hangup events as write failures or attempt the write so the terminal loop receives the asynchronous error.
Useful? React with 👍 / 👎.
| } | ||
| let Some(session) = session.upgrade() else { | ||
| return; | ||
| }; |
There was a problem hiding this comment.
When terminal EOF leaves hundreds of small packets queued, the bridge returns and finish_terminal joins this writer without reading the wake socket anymore. This blocking one-byte signal() runs for every drained packet; on Unix a socketpair can exhaust its send queue after only a few hundred tiny writes, causing the writer to block permanently and the graceful join to hang even though network delivery is progressing. Make these notifications nonblocking/coalesced, and signal only when queue capacity actually transitions to available.
Useful? React with 👍 / 👎.
| if std::env::args_os().nth(1).as_deref() == Some(std::ffi::OsStr::new("__et-console-writer")) { | ||
| std::process::exit(crate::client_output::run_windows_helper()); | ||
| } |
There was a problem hiding this comment.
On Windows, a normal invocation whose destination or SSH-config alias is exactly __et-console-writer is intercepted here before Clap or role dispatch runs. Instead of connecting, the process enters the private length-framed helper protocol and may hang reading stdin or emit unexpected bytes. Use a private environment marker or another invocation mechanism that cannot collide with the client's positional host argument.
Useful? React with 👍 / 👎.
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com> Co-authored-by: OmO <noreply@openai.com> Co-authored-by: GPT-5.6 <noreply@openai.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b298f53f37
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| writer::FlowWriteResult::BeforeReplay(_error) => { | ||
| state.queue.restore_front(packet); | ||
| state.connected = false; |
There was a problem hiding this comment.
When TcpStream::try_clone fails transiently, such as during temporary file-descriptor exhaustion, write_packet_with returns BeforeReplay together with connected == true, but this branch unconditionally changes the flow state to disconnected. The live socket remains healthy, so the bridge never initiates recovery and next_packet waits indefinitely, permanently stalling queued terminal and forwarding output. Fresh evidence after moving the clone before encryption is that this pre-replay path now preserves the underlying connection's connected() result; keep that result or retry instead of overriding it.
Useful? React with 👍 / 👎.
| let Some(bytes) = state.queue.pop_front() else { | ||
| return; | ||
| }; | ||
| state.bytes -= bytes.bytes.len(); |
There was a problem hiding this comment.
When stdout stalls while the worker is writing a large entry, this removes the entry from both the deque and state.bytes before the write completes, then immediately signals capacity. The pump can therefore admit another full 64 KiB while the first 64 KiB is still retained by the worker, so the supposedly 64 KiB-bounded console path can hold roughly twice its limit and discard mode cannot evict the stale in-flight output. Keep the entry charged until its write finishes, as the server-side queue does.
Useful? React with 👍 / 👎.
| if bounded_output { | ||
| // Destination output enters the jumphost router through this terminal- | ||
| // side sender. Keep pressure in the server's bounded application lanes. | ||
| et_net::local::minimize_terminal_output_buffering(&router) | ||
| .map_err(|error| format!("could not bound jumphost output buffering: {error}"))?; |
There was a problem hiding this comment.
For a jumphost session using backpressure, the outer server intentionally stops reading this router when its terminal lane fills, and this new small send buffer makes the router reach capacity routinely. However, relay has configured the router as nonblocking and its destination-to-router path treats any write_local_packet error, including WouldBlock after a partial write, as relay termination at lines 228-229 and 282-283. Sustained destination output therefore tears down the jumphost instead of applying lossless backpressure; retain the framed packet and resume it when the router is writable.
Useful? React with 👍 / 👎.
| let Some(removed) = state.queue.pop_front() else { | ||
| break; | ||
| }; | ||
| state.bytes -= removed.bytes.len(); |
There was a problem hiding this comment.
When discard mode evicts an entry containing a device-status request such as ESC[6n, the entry never reaches run_writer, which is the only place that records the request and triggers an automatic cursor response. An interactive emulator also cannot answer because it never sees the evicted bytes, so a remote ConPTY or application waiting for the response can block indefinitely after output saturation. Treat response-requiring terminal queries as non-discardable control data or record and answer them when an entry is evicted.
Useful? React with 👍 / 👎.
| if !recover(connection, &mut reconnect, &mut stream, terminal_enabled)? { | ||
| return Ok(()); | ||
| return console_output | ||
| .complete(ConsoleCompletion::RemoteSessionEnded) | ||
| .map_err(|error| terminal_io("draining terminal output", error)); |
There was a problem hiding this comment.
When backpressure leaves the final server packet in pending_output because the local console queue is full, a simultaneous network HUP sets reconnect_needed; if reconnect then reports SessionEnded, this path drains only packets already admitted to ConsoleOutput and drops the separately held final packet. This loses command output despite backpressure being advertised as lossless, even when the capacity wake arrived in the same poll iteration because the held packet is not retried until the next loop. Admit or synchronously flush pending_output before completing the console.
Useful? React with 👍 / 👎.
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com> Co-authored-by: OmO <noreply@openai.com> Co-authored-by: GPT-5.6 <noreply@openai.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a73be99daa
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if reconnect_needed { | ||
| if !recover(connection, &mut reconnect, &mut stream, terminal_enabled)? { | ||
| return Ok(()); | ||
| return finish_remote_completion( |
There was a problem hiding this comment.
When backpressure leaves pending_output occupied and the server sends additional final packets before closing its write side, POLLHUP sets reconnect_needed while the read loop remains disabled by that pending packet. If reconnect then returns SessionEnded, finish_remote_completion preserves only pending_output and never reads the additional packets already buffered in the socket or BackedReader, silently losing the session tail despite lossless backpressure. The retained-packet fix therefore needs to drain the ordered connection stream after capacity returns, before accepting SessionEnded as completion.
Useful? React with 👍 / 👎.
| if terminal_eof && pending_terminal.is_none() { | ||
| return Ok(()); |
There was a problem hiding this comment.
When the flow-control control lane is full, an outbound forwarding packet is removed from the forwarder and retained only in pending_outbound. If the terminal reaches EOF before queue capacity reopens, this condition returns from the bridge as soon as pending_terminal is empty, without checking pending_outbound; finish_terminal can drain only packets already in the flow queue, so the held forwarding bytes are dropped and the tunnel stream is truncated. Terminal completion should first admit every bridge-owned outbound packet or transfer it into the graceful drain state.
Useful? React with 👍 / 👎.
| } | ||
| } | ||
| let network_flags = if pending_forward.is_none() && pending_output.is_none() { |
There was a problem hiding this comment.
When backpressure fills the local console queue, this removes IN from the network poll set and stops draining the TCP socket, but the client never reduces that socket's receive buffer. On a fast link with slow or blocked stdout, the kernel can therefore ACK and retain a large autotuned receive window after the 64 KiB console queue fills, allowing substantially more old terminal output to accumulate ahead of a later prompt and defeating the intended bounded-latency behavior. Configure a bounded receive buffer for opted-in backpressure sessions before using socket readability as the pressure boundary.
Useful? React with 👍 / 👎.
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com> Co-authored-by: OmO <noreply@openai.com> Co-authored-by: GPT-5.6 <noreply@openai.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 56f33648cc
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| // Popping this packet may have reopened bounded queue capacity. | ||
| // Wake the bridge so it polls terminal output again instead of | ||
| // sleeping indefinitely with terminal readability disabled. | ||
| let _ = session.signal(); |
There was a problem hiding this comment.
In the Unix bridge, when backpressure fills the terminal lane, OutputQueue::take deliberately keeps the in-flight packet charged until flow.complete, so this signal wakes the bridge while can_accept_terminal is still false; the bridge then sleeps again, and complete only notifies the writer condvar rather than the bridge wake socket. Continuous terminal output therefore pauses until an unrelated client/forwarding event—commonly the next keepalive—after every full queue. Fresh evidence for the previously reported wakeup issue is that the newly added signal is placed before the operation that actually frees capacity; move or repeat it after successful completion.
Useful? React with 👍 / 👎.
| if matches!( | ||
| write_cursor_report( | ||
| connection, | ||
| &mut reconnect, | ||
| &mut stream, | ||
| terminal_enabled, | ||
| )?, | ||
| OwnedWriteOutcome::SessionEnded |
There was a problem hiding this comment.
In the Unix pump, if this poll iteration observed HUP on the old network stream, reconnect_needed is already true; an asynchronous cursor report can then recover synchronously here and replace stream, but this branch handles only SessionEnded and leaves that stale flag set. The loop consequently calls recover again at the end of the same iteration, forcing a second returning-client handshake even though the first recovery succeeded (and potentially leaving the user back in the retry loop if that redundant connection attempt fails). Handle OwnedWriteOutcome::Recovered by clearing reconnect_needed and refreshing the keepalive timers; the resize branch has the same outcome-handling gap.
Useful? React with 👍 / 👎.
| let _ = self.session.flush_recover_hold(); | ||
| self.session.recovering.store(false, Ordering::Release); | ||
| // Catch anything that observed `recovering` and queued after the first | ||
| // flush but before the flag cleared (re-check is under the hold lock). | ||
| let _ = self.session.flush_recover_hold(); |
There was a problem hiding this comment.
For a non-flow-controlled session, if output was queued in recover_hold during recovery and transport preparation keeps failing before replay ownership—for example, TcpStream::try_clone repeatedly fails during file-descriptor exhaustion—flush_recover_hold restores that plaintext to the hold, but both errors here are ignored and recovering is cleared anyway. Subsequent send_packet calls then bypass the hold and write newer packets directly, while no normal path retries the stranded entries until another recovery happens, causing output to remain undelivered and later output to overtake it. Do not clear the recovery state without arranging a durable retry, or disconnect the writer so the held packets can be admitted to its replay buffer.
Useful? React with 👍 / 👎.
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com> Co-authored-by: OmO <noreply@openai.com> Co-authored-by: GPT-5.6 <noreply@openai.com>
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com> Co-authored-by: OmO <noreply@openai.com> Co-authored-by: GPT-5.6 <noreply@openai.com>
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com> Co-authored-by: OmO <noreply@openai.com> Co-authored-by: GPT-5.6 <noreply@openai.com>
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com> Co-authored-by: OmO <noreply@openai.com> Co-authored-by: GPT-5.6 <noreply@openai.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: ec57a39b6b
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if !state.connected { | ||
| state.unrecoverable = true; | ||
| state.stop = StopMode::Hard; |
There was a problem hiding this comment.
When terminal EOF races with a returning client after the old transport was marked disconnected, recover_body has already paused the writer but state.connected remains false until RecoverPermit::drop calls resume. This branch therefore switches to Hard and makes finish_terminal terminate the writer; even if the recovery candidate installs successfully moments later, resume cannot undo the hard stop, so queued final output is discarded despite backpressure being lossless. Defer the graceful-stop decision while the writer is paused for recovery.
Useful? React with 👍 / 👎.
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 36e1161c34
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| if client_events.contains(PollFlags::IN) && pending_output.is_none() { | ||
| match destination.try_read_packet() { | ||
| Ok(Some(packet)) => { |
There was a problem hiding this comment.
On Unix jumphosts, when the destination closes after sending multiple packets and poll returns IN together with HUP or ERR, this new single-packet read consumes at most one packet before the branch below immediately exits the relay. Any additional frames already buffered in the socket or BackedReader are dropped, truncating final terminal or forwarded output even in backpressure mode; defer closure until the pending local frame is flushed and reads reach EOF.
Useful? React with 👍 / 👎.
| break; | ||
| } | ||
| Err(error) => return Err(error), | ||
| terminal_closing |= terminal_events.intersects(PollFlags::HUP | PollFlags::ERR); |
There was a problem hiding this comment.
When a disconnected or slow flow-controlled session fills its terminal lane just as the local terminal closes, pending_terminal remains occupied and this records the persistent HUP; the next wait still polls that terminal descriptor for HUP, so it returns immediately despite the retry timeout and the bridge spins retrying the full queue until recovery or capacity arrives. A disconnected ended shell can therefore consume a server core indefinitely; after recording closure, wait only for flow capacity/recovery while a packet is held.
Useful? React with 👍 / 👎.
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>
Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>
Preserve flow-control replay ownership, bounded jumphost relay state, and broadcast cancellation alongside current forwarding security and lifecycle hardening. Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 8d1e1e4234
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| /// be completed and were explicitly abandoned. | ||
| pub fn shutdown_hard(&mut self) -> Result<bool, ForwardError> { | ||
| self.cancel.take(); | ||
| let mut abandoned = !self.commands.is_empty() || !self.outbound.is_empty(); |
There was a problem hiding this comment.
When the remote session ends after a forwarding packet has moved from commands into an active socket writer, a slow local endpoint can leave that data queued or blocked inside write_all. This abandonment check examines only the command and outbound channels, but the joined worker handles cancellation by calling abort_io, which explicitly discards active writers' queued output; shutdown_hard can therefore return false, causing finish_remote_completion to report a clean exit while silently truncating the forwarded stream. Track active writer backlog as abandoned or drain it before reporting successful completion.
Useful? React with 👍 / 👎.
Make per-socket stop admission cancellation-safe, report discarded admitted bytes, and drain retained jumphost output after destination closure. Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>
Preserve startup-under-load hardening while pausing flow writers before recovery status exposure, retaining final forwarding ownership, and keeping Windows test helpers warning-clean. Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>
Promote returning raw sockets before exposing recovery status so terminal disconnect cleanup cannot cancel an admitted recovery candidate. Cover the real HUP boundary with deterministic keyed lifecycle hooks. Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
31 issues found across 64 files
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="crates/et-bin/src/client_terminal_loop_tests.rs">
<violation number="1" location="crates/et-bin/src/client_terminal_loop_tests.rs:73">
P2: This test never runs `pump` or supplies its stdin; it sends Ctrl-C over an unrelated `Connection` pair after calling `route_server_packet` directly. A regression that blocks the real pump while `pending_output` is set would still pass; drive the actual pump with controllable input and assert that the packet reaches the server.</violation>
</file>
<file name="crates/et-bin/src/client_output_tests.rs">
<violation number="1" location="crates/et-bin/src/client_output_tests.rs:5">
P3: `GatedWriter` duplicates `GatedConsole`, including its blocking writer behavior. Extract a shared test helper so future synchronization changes cannot diverge between these tests.</violation>
<violation number="2" location="crates/et-bin/src/client_output_tests.rs:44">
P2: This channel is disconnected from `ConsoleOutput` and the terminal loop, so the send/receive succeeds even if a full queue blocks Ctrl-C handling. Exercise `route_server_packet` with a real connection, as `client_terminal_loop_tests.rs` does.</violation>
</file>
<file name="crates/et-net/src/forward_worker.rs">
<violation number="1" location="crates/et-net/src/forward_worker.rs:347">
P2: When a destination Unix socket connect blocks, hard cancellation still waits indefinitely because this receiver is only observed after the connect returns and `Worker::run` joins the connector thread. Make connector establishment interruptible, or use an abort-aware teardown that does not synchronously join a blocked connector.</violation>
</file>
<file name="crates/et-bin/src/client_terminal_loop.rs">
<violation number="1" location="crates/et-bin/src/client_terminal_loop.rs:289">
P2: When a resize, cursor-report, or keepalive write reconnects successfully, this call ignores `OwnedWriteOutcome::Recovered` and leaves `last_received` stale. The next silence check performs another reconnect; handle `Recovered` like input and forwarding writes by refreshing the liveness timestamps.</violation>
<violation number="2" location="crates/et-bin/src/client_terminal_loop.rs:396">
P2: When backpressure leaves `pending_output` set, this guard disables the silence watchdog indefinitely. A silently dead TCP connection then receives keepalives but never calls `recover`; keep transport liveness detection active while draining the packet.</violation>
</file>
<file name="crates/et-server/src/runtime_recovery_test.rs">
<violation number="1" location="crates/et-server/src/runtime_recovery_test.rs:56">
P2: Running Miri executes this network-dependent test, but Miri cannot support its TCP and Unix socket operations. Add the same Miri ignore attribute used by the adjacent runtime tests.</violation>
</file>
<file name="crates/et-net/src/forward_io.rs">
<violation number="1" location="crates/et-net/src/forward_io.rs:316">
P1: When hard cancellation races `stop_io` while `send(io.writer, Stop)` is ready, `select!` can choose the send branch even though cancellation is ready. `stop_io` then only shuts down the read half, so a blocked writer can strand `shutdown_hard` after `remove` has discarded the `ActiveIo`; retain removed I/O for abort or coordinate cancellation so it always closes both halves.</violation>
</file>
<file name="crates/et-bin/src/client_terminal.rs">
<violation number="1" location="crates/et-bin/src/client_terminal.rs:260">
P1: When discard-mode output is saturated, this callback can lose a terminal `ESC[6n` request before it is classified. Preserve cursor-report/control data losslessly, or detect and handle it before lossy output admission, otherwise ConPTY can remain blocked waiting for a response.</violation>
</file>
<file name="crates/et-server/src/session.rs">
<violation number="1" location="crates/et-server/src/session.rs:200">
P1: When a backpressure queue fills, the bridge stops polling terminal output and can remain asleep after the writer drains the queue. `FlowControl::complete` releases the in-flight reservation without waking the bridge, so subsequent terminal output is not read until an unrelated wake arrives; signal the session wake pipe after completion releases capacity.</violation>
</file>
<file name="crates/et-net/src/forward.rs">
<violation number="1" location="crates/et-net/src/forward.rs:534">
P2: When hard cancellation races with a worker processing a dequeued forwarding command, both queues can be empty even though that command is abandoned. Track in-flight cancellation in the worker and include it in the returned `abandoned` result, otherwise remote completion can be reported successful after losing forwarding data.</violation>
</file>
<file name="crates/et-server/src/session_recovery.rs">
<violation number="1" location="crates/et-server/src/session_recovery.rs:76">
P2: When the connection mutex times out or is poisoned, this `?` drops the batch taken from `recover_hold`; `RecoverPermit::drop` ignores the error, so terminal output is lost and the byte budget remains charged. Restore the batch before returning the lock error.</violation>
</file>
<file name="crates/et-net/src/local_packet.rs">
<violation number="1" location="crates/et-net/src/local_packet.rs:143">
P3: Every packet written through this hot path now allocates a second `frame` Vec and memcpys up to the full 64 KiB `serialize()` output into it, in addition to `serialize()`'s own allocation. Previously `write_local_packet_with` wrote the length prefix and payload in two `write_all_blocking` calls with no combined copy, so terminal output (via `write_local_packet_until_cancelled`), status packets, and the runtime handler each incur an extra per-frame allocation and copy. The combined frame is only needed by the flow-control queue in `terminal_jump.rs`; the direct-write path could keep the two-call form to avoid it.</violation>
</file>
<file name="crates/et-server/src/terminal_bridge.rs">
<violation number="1" location="crates/et-server/src/terminal_bridge.rs:133">
P1: When a terminal packet is pending, Unix `wait` stops polling client readability, so later Ctrl-C/input packets can wait behind a blocked terminal write. Keep client `IN` polling enabled for `pending_terminal`; only forwarding backpressure needs this read suppression.</violation>
</file>
<file name="crates/et-bin/src/terminal_jump.rs">
<violation number="1" location="crates/et-bin/src/terminal_jump.rs:274">
P2: When the jumphost router closes while output is pending on Windows, `relay_with_output_observer` now returns an error instead of normal relay termination. Handle both Windows `write_pending_local` sites like the Unix branch and return `Ok(0)` for router write failures.</violation>
</file>
<file name="crates/et-net/src/local.rs">
<violation number="1" location="crates/et-net/src/local.rs:48">
P3: On Linux this sets the bound to ~128 KiB, not the documented 64 KiB: the kernel doubles SO_SNDBUF to reserve bookkeeping space, so send_buffer_size() reads back 2x the requested value (your own test in terminal_jump_tests.rs:378-380 asserts the doubled bound). The effective kernel queue is twice the constant, so the '64 KiB' doc comment and constant name understate the real bound. Since the bound is an OS behavior, either document the ~2x amplification on Linux in the doc comment or pick the constant accordingly.</violation>
</file>
<file name="crates/et-bin/src/client_terminal_windows.rs">
<violation number="1" location="crates/et-bin/src/client_terminal_windows.rs:146">
P2: When a console, resize, forwarding, or cursor-report write reconnects an otherwise idle Windows session, the loop leaves `last_received` stale and immediately treats the recovered connection as silent again. Refresh the heartbeat timestamps whenever a write returns `OwnedWriteOutcome::Recovered`, as the Unix pump does.</violation>
<violation number="2" location="crates/et-bin/src/client_terminal_windows.rs:408">
P1: In discard mode, evicting a queued terminal packet can discard its `ESC[6n` cursor request before the worker reports it. The async path skips direct cursor replies, so preserve cursor-report control bytes across eviction or detect and answer them before discarding output.</violation>
</file>
<file name="crates/et-bin/tests/reconnect_outage_end_to_end.rs">
<violation number="1" location="crates/et-bin/tests/reconnect_outage_end_to_end.rs:213">
P3: `wait_for_refused` calls `recv_timeout(TIMEOUT).unwrap()`, so a timeout or channel disconnect fails with a generic unwrap panic rather than a message naming the expected refusals. The rest of this file (receive_until/receive_number) reports recv_timeout failures with descriptive panics; follow that style here.</violation>
</file>
<file name="crates/et-bin/src/client_output.rs">
<violation number="1" location="crates/et-bin/src/client_output.rs:331">
P1: When the console stops accepting output while the remote session ends, `finish_gracefully` waits forever on the worker. Setting `stopping` only wakes the queue; it cannot interrupt an in-progress `write_all`, and the cancellation callback runs only after this join. Add a bounded drain timeout with cancellation fallback before joining the worker.</violation>
<violation number="2" location="crates/et-bin/src/client_output.rs:542">
P1: When a redirected stdout pipe closes, `poll` reports `ERR`/`HUP` rather than `OUT`, so this branch spins forever and never surfaces the broken output. Treat terminal-output error/hangup readiness as a write error so the worker can stop and session shutdown can complete.</violation>
</file>
<file name="crates/et-server/tests/terminal_bridge.rs">
<violation number="1" location="crates/et-server/tests/terminal_bridge.rs:165">
P3: The terminal stream in this test never gets a read timeout, so the `read_local_packet(&mut terminal)` for `_init` will block indefinitely if the server fails to deliver TermInit. Every other test in this file sets `terminal.set_read_timeout(Some(TIMEOUT))` before reading; do the same here so a missing TermInit fails the test instead of hanging the `--test-threads=1` run.</violation>
<violation number="2" location="crates/et-server/tests/terminal_bridge.rs:263">
P3: The Discard branch cannot detect that discard dropped any output: it only checks that values are strictly ordered and end in 31, which is also true when all 32 packets are delivered losslessly (queue never drops). Assert an upper bound on the received count (e.g. the 64 KiB lane / 16 KiB packet allows at most 4) so a regression that makes Discard behave like Backpressure fails the test.</violation>
</file>
<file name="crates/et-bin/tests/flow_control_tty_qa.rs">
<violation number="1" location="crates/et-bin/tests/flow_control_tty_qa.rs:114">
P3: The FLOW-INTERRUPTED wait uses the full `prompt_timeout`, but the two later waits budget against `deadline`. Pass the remaining time to the deadline for the interrupt wait too, so all three stages share one hard budget and failures are attributable.</violation>
</file>
<file name="crates/et-server/tests/flow_control_recovery.rs">
<violation number="1" location="crates/et-server/tests/flow_control_recovery.rs:89">
P2: If a test in this file fails before it sends `gate.release` (for example `snapshot.recv_timeout` or `send_packet` panics), `Drop` joins the worker thread while it is still blocked on `listener.accept()` or `release_rx.recv()`. Shutting down the `stop` stream only unblocks the `io::copy` stage, so the `worker.join()` in `Drop` blocks forever, turning a test failure into an unbounded hang. Make the join interruptible (e.g. don't join in `Drop`, or bound the worker with non-blocking select/timeouts on its receive points) so a failed test returns promptly.</violation>
<violation number="2" location="crates/et-server/tests/flow_control_recovery.rs:102">
P3: Both recovery tests only opt into `FlowControlMode::Backpressure`. The PR introduces a second opt-in mode, `Discard`, whose recovery semantics (oldest-output discard while preserving control packets) are materially different and are left untested on this reconnect path. Add a discard-mode test that queues output past the bound, recovers, and asserts the bounded/control delivery behavior.</violation>
</file>
<file name="crates/et-net/src/connection.rs">
<violation number="1" location="crates/et-net/src/connection.rs:166">
P2: Every live `write_packet`/`write_terminal` now clones the TcpStream (`TcpStream::try_clone`, a `dup()` + later fd close) and writes on the clone, even on the default non-flow-control path that the PR claims is unchanged. This adds two syscalls per terminal output packet to all sessions. Clone only where the flow-control `PreparedWrite` design actually needs an owned stream (e.g. a dedicated sender), or reuse the borrowed `&mut self.stream` for the synchronous `write_packet` path.</violation>
</file>
<file name="crates/et-server/src/session_flow_test.rs">
<violation number="1" location="crates/et-server/src/session_flow_test.rs:514">
P3: Pushing into `recover_hold` without incrementing `recover_hold_bytes` makes the ReplayOwned path in `flush_recover_hold_with` call `fetch_sub(payload.len())` on a zero counter, wrapping it to u64::MAX-8. Keep the accounting consistent: increment `recover_hold_bytes` alongside the push (or set `recovering` and use `queue_if_recovering`).</violation>
</file>
<file name="crates/et-net/src/forward_worker_state.rs">
<violation number="1" location="crates/et-net/src/forward_worker_state.rs:281">
P1: Graceful `Forwarder::shutdown()` can deadlock: the worker's new select-based `emit`/`receive_data` only wake on cancel-channel disconnect, but `stop()` never drops the cancel sender. When the worker is blocked in `emit` (outbound queue full) or `receive_data` (writer queue full) at shutdown time, `stop()` sets `self.shutdown` and calls `commands.shutdown()`, neither of which the worker observes while stuck in the select, so `worker.join()` blocks forever. The old `emit` looped on `self.shutdown.load(Acquire)` and returned `Err(ForwardError::Unavailable)` when set, so `stop()` used to wake it; the `shutdown` flag was removed from `Worker` in this change. The pre-existing test `shutdown_completes_with_command_and_outbound_queues_saturated` (forward.rs) saturates both queues and calls `shutdown()`, which now hangs. Re-add the shutdown flag to `Worker` and check it in the `emit`/`receive_data` blocking paths (or otherwise wake the worker on graceful stop without flipping the drain to the hard `abort_io` path).</violation>
</file>
<file name="crates/et-core/src/flow_control.rs">
<violation number="1" location="crates/et-core/src/flow_control.rs:77">
P2: In Discard mode a terminal packet can be refused with `QueuePushError::Full` when an in-flight packet (reserved in `terminal_bytes` by `take()`) already consumes the budget and the deque is empty. `enqueue()` maps that to `ConnError::Backpressure`, so the newest output is dropped with a backpressure/session error instead of the oldest-output-discard behavior this mode is meant to guarantee. `can_accept_terminal` in Discard mode also ignores in-flight bytes and the per-lane packet cap, so the bridge can be told output is acceptable and then have `push` reject it.</violation>
</file>
<file name="crates/et-bin/src/terminal_protocol.rs">
<violation number="1" location="crates/et-bin/src/terminal_protocol.rs:34">
P2: When a client sends an unrecognized `flowcontrol` value (any integer other than 0/1/2), `FlowControlMode::try_from(value).ok()` drops the error and `.unwrap_or(FlowControlMode::None)` silently downgrades the session to `None`. Because `None` keeps the unbounded synchronous output path, a client that negotiates an unsupported or future mode fails open, defeating the output-bounding this feature exists to provide. Return an error for unrecognized modes instead of silently treating them as "no flow control" so the failure is explicit.</violation>
</file>
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
| io.control.shutdown_read(); | ||
| // Keep the control socket alive while Stop waits for queue capacity so | ||
| // hard cancellation can still abort an in-flight write and release it. | ||
| let stop_admitted = channel::select! { |
There was a problem hiding this comment.
P1: When hard cancellation races stop_io while send(io.writer, Stop) is ready, select! can choose the send branch even though cancellation is ready. stop_io then only shuts down the read half, so a blocked writer can strand shutdown_hard after remove has discarded the ActiveIo; retain removed I/O for abort or coordinate cancellation so it always closes both halves.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-net/src/forward_io.rs, line 316:
<comment>When hard cancellation races `stop_io` while `send(io.writer, Stop)` is ready, `select!` can choose the send branch even though cancellation is ready. `stop_io` then only shuts down the read half, so a blocked writer can strand `shutdown_hard` after `remove` has discarded the `ActiveIo`; retain removed I/O for abort or coordinate cancellation so it always closes both halves.</comment>
<file context>
@@ -143,81 +155,269 @@ pub(crate) fn spawn_io(
- io.control.shutdown_read();
+ // Keep the control socket alive while Stop waits for queue capacity so
+ // hard cancellation can still abort an in-flight write and release it.
+ let stop_admitted = channel::select! {
+ send(io.writer, WriteCommand::Stop) -> result => result.is_ok(),
+ recv(io.cancel) -> _ => false,
</file context>
| .and_then(|()| io::stdout().lock().flush()) | ||
| .map_err(|error| terminal_io("writing terminal output", error))?; | ||
| Ok(contains_cursor_report_request(&bytes)) | ||
| if !output(&bytes)? { |
There was a problem hiding this comment.
P1: When discard-mode output is saturated, this callback can lose a terminal ESC[6n request before it is classified. Preserve cursor-report/control data losslessly, or detect and handle it before lossy output admission, otherwise ConPTY can remain blocked waiting for a response.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-bin/src/client_terminal.rs, line 260:
<comment>When discard-mode output is saturated, this callback can lose a terminal `ESC[6n` request before it is classified. Preserve cursor-report/control data losslessly, or detect and handle it before lossy output admission, otherwise ConPTY can remain blocked waiting for a response.</comment>
<file context>
@@ -184,42 +191,97 @@ pub(crate) const CURSOR_REPORT_REPLY: &[u8] = b"\x1b[1;1R";
- .and_then(|()| io::stdout().lock().flush())
- .map_err(|error| terminal_io("writing terminal output", error))?;
- Ok(contains_cursor_report_request(&bytes))
+ if !output(&bytes)? {
+ return Ok(DisplayOutcome::Pending(packet));
+ }
</file context>
| where | ||
| W: FnMut(&mut Connection, u8, &[u8]) -> Result<(), et_net::connection::WritePacketError>, | ||
| { | ||
| if let Some(state) = &self.flow_control { |
There was a problem hiding this comment.
P1: When a backpressure queue fills, the bridge stops polling terminal output and can remain asleep after the writer drains the queue. FlowControl::complete releases the in-flight reservation without waking the bridge, so subsequent terminal output is not read until an unrelated wake arrives; signal the session wake pipe after completion releases capacity.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-server/src/session.rs, line 200:
<comment>When a backpressure queue fills, the bridge stops polling terminal output and can remain asleep after the writer drains the queue. `FlowControl::complete` releases the in-flight reservation without waking the bridge, so subsequent terminal output is not read until an unrelated wake arrives; signal the session wake pipe after completion releases capacity.</comment>
<file context>
@@ -102,38 +137,108 @@ impl ActiveSession {
+ where
+ W: FnMut(&mut Connection, u8, &[u8]) -> Result<(), et_net::connection::WritePacketError>,
+ {
+ if let Some(state) = &self.flow_control {
+ return state
+ .enqueue(Packet::new(header, payload))
</file context>
| || pending_terminal.is_some() | ||
| || !connected, |
There was a problem hiding this comment.
P1: When a terminal packet is pending, Unix wait stops polling client readability, so later Ctrl-C/input packets can wait behind a blocked terminal write. Keep client IN polling enabled for pending_terminal; only forwarding backpressure needs this read suppression.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-server/src/terminal_bridge.rs, line 133:
<comment>When a terminal packet is pending, Unix `wait` stops polling client readability, so later Ctrl-C/input packets can wait behind a blocked terminal write. Keep client `IN` polling enabled for `pending_terminal`; only forwarding backpressure needs this read suppression.</comment>
<file context>
@@ -118,7 +128,10 @@ fn run_mode_poll(
- pending_forward.is_some() || pending_outbound.is_some() || !connected,
+ pending_forward.is_some()
+ || pending_outbound.is_some()
+ || pending_terminal.is_some()
+ || !connected,
)?;
</file context>
| || pending_terminal.is_some() | |
| || !connected, | |
| || !connected, |
| return display_packet(packet, terminal_modes); | ||
| return crate::client_terminal::display_packet_with(packet, |bytes| { | ||
| output | ||
| .try_write(bytes, terminal_modes) |
There was a problem hiding this comment.
P1: In discard mode, evicting a queued terminal packet can discard its ESC[6n cursor request before the worker reports it. The async path skips direct cursor replies, so preserve cursor-report control bytes across eviction or detect and answer them before discarding output.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-bin/src/client_terminal_windows.rs, line 408:
<comment>In discard mode, evicting a queued terminal packet can discard its `ESC[6n` cursor request before the worker reports it. The async path skips direct cursor replies, so preserve cursor-report control bytes across eviction or detect and answer them before discarding output.</comment>
<file context>
@@ -180,17 +331,88 @@ where
- return display_packet(packet, terminal_modes);
+ return crate::client_terminal::display_packet_with(packet, |bytes| {
+ output
+ .try_write(bytes, terminal_modes)
+ .map_err(|error| terminal_io("writing terminal output", error))
+ });
</file context>
| #[test] | ||
| fn flow_control_mode_reaches_terminal_and_relays_output() { | ||
| let mut server = TestRuntime::start(); | ||
| let mut terminal = server.register(ID_A, KEY_A); |
There was a problem hiding this comment.
P3: The terminal stream in this test never gets a read timeout, so the read_local_packet(&mut terminal) for _init will block indefinitely if the server fails to deliver TermInit. Every other test in this file sets terminal.set_read_timeout(Some(TIMEOUT)) before reading; do the same here so a missing TermInit fails the test instead of hanging the --test-threads=1 run.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-server/tests/terminal_bridge.rs, line 165:
<comment>The terminal stream in this test never gets a read timeout, so the `read_local_packet(&mut terminal)` for `_init` will block indefinitely if the server fails to deliver TermInit. Every other test in this file sets `terminal.set_read_timeout(Some(TIMEOUT))` before reading; do the same here so a missing TermInit fails the test instead of hanging the `--test-threads=1` run.</comment>
<file context>
@@ -157,6 +159,115 @@ fn terminal_hup_still_delivers_buffered_final_packet() {
+#[test]
+fn flow_control_mode_reaches_terminal_and_relays_output() {
+ let mut server = TestRuntime::start();
+ let mut terminal = server.register(ID_A, KEY_A);
+ terminal.set_read_timeout(Some(TIMEOUT)).unwrap();
+ let (stream, response) = server.handshake(ID_A);
</file context>
| let mut terminal = server.register(ID_A, KEY_A); | |
| let mut terminal = server.register(ID_A, KEY_A); | |
| terminal.set_read_timeout(Some(TIMEOUT)).unwrap(); |
| values.windows(2).all(|pair| pair[0] < pair[1]), | ||
| "{values:?}" | ||
| ); | ||
| assert_eq!(values.last(), Some(&31)); |
There was a problem hiding this comment.
P3: The Discard branch cannot detect that discard dropped any output: it only checks that values are strictly ordered and end in 31, which is also true when all 32 packets are delivered losslessly (queue never drops). Assert an upper bound on the received count (e.g. the 64 KiB lane / 16 KiB packet allows at most 4) so a regression that makes Discard behave like Backpressure fails the test.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-server/tests/terminal_bridge.rs, line 263:
<comment>The Discard branch cannot detect that discard dropped any output: it only checks that values are strictly ordered and end in 31, which is also true when all 32 packets are delivered losslessly (queue never drops). Assert an upper bound on the received count (e.g. the 64 KiB lane / 16 KiB packet allows at most 4) so a regression that makes Discard behave like Backpressure fails the test.</comment>
<file context>
@@ -157,6 +159,115 @@ fn terminal_hup_still_delivers_buffered_final_packet() {
+ values.windows(2).all(|pair| pair[0] < pair[1]),
+ "{values:?}"
+ );
+ assert_eq!(values.last(), Some(&31));
+ if mode == FlowControlMode::Backpressure {
+ assert_eq!(values, (0u8..32).collect::<Vec<_>>());
</file context>
| let deadline = interrupted + prompt_timeout; | ||
| writer.write_all(b"\x03").unwrap(); | ||
| writer.flush().unwrap(); | ||
| let interrupt = receive_until( |
There was a problem hiding this comment.
P3: The FLOW-INTERRUPTED wait uses the full prompt_timeout, but the two later waits budget against deadline. Pass the remaining time to the deadline for the interrupt wait too, so all three stages share one hard budget and failures are attributable.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-bin/tests/flow_control_tty_qa.rs, line 114:
<comment>The FLOW-INTERRUPTED wait uses the full `prompt_timeout`, but the two later waits budget against `deadline`. Pass the remaining time to the deadline for the interrupt wait too, so all three stages share one hard budget and failures are attributable.</comment>
<file context>
@@ -0,0 +1,177 @@
+ let deadline = interrupted + prompt_timeout;
+ writer.write_all(b"\x03").unwrap();
+ writer.flush().unwrap();
+ let interrupt = receive_until(
+ &receiver,
+ output.clone(),
</file context>
| assert_eq!(response.status, Some(ConnectStatus::NewClient as i32)); | ||
| let key = passkey_to_key(KEY_A).unwrap(); | ||
| let mut payload = default_payload(); | ||
| payload.flowcontrol = Some(FlowControlMode::Backpressure as i32); |
There was a problem hiding this comment.
P3: Both recovery tests only opt into FlowControlMode::Backpressure. The PR introduces a second opt-in mode, Discard, whose recovery semantics (oldest-output discard while preserving control packets) are materially different and are left untested on this reconnect path. Add a discard-mode test that queues output past the bound, recovers, and asserts the bounded/control delivery behavior.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-server/tests/flow_control_recovery.rs, line 102:
<comment>Both recovery tests only opt into `FlowControlMode::Backpressure`. The PR introduces a second opt-in mode, `Discard`, whose recovery semantics (oldest-output discard while preserving control packets) are materially different and are left untested on this reconnect path. Add a discard-mode test that queues output past the bound, recovers, and asserts the bounded/control delivery behavior.</comment>
<file context>
@@ -0,0 +1,249 @@
+ assert_eq!(response.status, Some(ConnectStatus::NewClient as i32));
+ let key = passkey_to_key(KEY_A).unwrap();
+ let mut payload = default_payload();
+ payload.flowcontrol = Some(FlowControlMode::Backpressure as i32);
+ let (mut client, initial) = initialize(stream, &key, payload);
+ assert_eq!(initial.error, None);
</file context>
| .recover_hold | ||
| .lock() | ||
| .unwrap() | ||
| .push((54, b"held-once".to_vec())); |
There was a problem hiding this comment.
P3: Pushing into recover_hold without incrementing recover_hold_bytes makes the ReplayOwned path in flush_recover_hold_with call fetch_sub(payload.len()) on a zero counter, wrapping it to u64::MAX-8. Keep the accounting consistent: increment recover_hold_bytes alongside the push (or set recovering and use queue_if_recovering).
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At crates/et-server/src/session_flow_test.rs, line 514:
<comment>Pushing into `recover_hold` without incrementing `recover_hold_bytes` makes the ReplayOwned path in `flush_recover_hold_with` call `fetch_sub(payload.len())` on a zero counter, wrapping it to u64::MAX-8. Keep the accounting consistent: increment `recover_hold_bytes` alongside the push (or set `recovering` and use `queue_if_recovering`).</comment>
<file context>
@@ -0,0 +1,632 @@
+ .recover_hold
+ .lock()
+ .unwrap()
+ .push((54, b"held-once".to_vec()));
+ let result = session.flush_recover_hold_with(|connection, header, payload| {
+ let prepared = connection.prepare_write_packet(header, payload)?;
</file context>
1 participant
Summary
--flow-control {none,backpressure,discard}with unchangednonedefaultInitialPayloadandTermInitPorts EternalTerminal issue #631 and the design from upstream PR #730.
Evidence
cargo fmt --all --checkcargo clippy --workspace --all-targets -- -D warningscargo test --workspace -- --test-threads=1Compatibility
Protocol version remains 6. Unset flow-control fields serialize to no bytes; legacy/default sessions keep the existing synchronous path.
Fixes upstream MisterTea/EternalTerminal#631.
Summary by cubic
Adds opt-in flow control for terminal sessions so a slow network no longer makes Ctrl-C and prompt responses unresponsive. Default sessions keep today's unbounded replay behavior; clients that pass
--flow-controlget lossless backpressure or oldest-output discard, both bounded at 64 KiB of queued terminal output.New Features
--flow-control {none,backpressure,discard}client flag onet, defaulting tonone.InitialPayloadandTermInitprotobuf fields, so protocol version stays 6 and unchanged sessions stay wire-compatible.Written for commit 7da329f. Summary will update on new commits.