Struct HostPipe 

pub struct HostPipe {
    inner: Arc<Mutex<HostPipeInner>>,
    events_tx: Sender<Event>,
    state: Arc<Mutex<State>>,
}

Public wrapper around the launcher → host pipe.

Cheap to clone (one Arc); shared by the IPC server’s per- connection task (set/clear writer) and the saga coordinator (send_command, CPD-3) and the per-host fanout task (send_event, refactored in this PR).

Manual Debug (rather than derive) so we don’t have to require Debug on the boxed writer trait object.

Fields

inner: Arc<Mutex<HostPipeInner>>

events_tx: Sender<Event>

Broadcast bus reference — used to emit Event::SagaFailed when a buffered Command is dropped (overflow or 30s timeout).

state: Arc<Mutex<State>>

Reference to the launcher’s reducer state for bump_version when emitting saga lifecycle events. Mirrors the saga coordinator’s pattern so emitted events get a fresh, monotonic global version.

Implementations

impl HostPipe

pub fn new(events_tx: Sender<Event>, state: Arc<Mutex<State>>) -> Self

pub async fn set_writer( &self, writer: Arc<Mutex<Box<dyn AsyncWrite + Unpin + Send>>>, ) -> u64

Hand the host’s writer half to the pipe. Called by the IPC server’s per-connection task once the connecting client registers as ClientKind::Host. Drains any pending frames in FIFO order before returning.

If a writer is already set (a prior host registered and this is a second), the prior writer is replaced. The drain logic still runs against the new writer.

Locking: the inner mutex is held across the per-frame write_all().await. That’s intentional — concurrent send_frame calls during drain would race FIFO ordering against the freshly-installed writer. Holding the lock keeps the drain-then-install transition atomic from the caller’s perspective. Hold time is bounded by pending_buffer size (cap 64) × per-frame write latency (microseconds on a healthy pipe), well under the existing state.lock() discipline. Returns the new host_session_id.

pub async fn clear_writer(&self)

Mark the host as disconnected. Arms the 30s timer; subsequent send_command calls buffer up to PENDING_BUFFER_CAP.

pub async fn cancel_saga(&self, saga_id: u64) -> usize

Remove every buffered frame whose saga_id matches saga_id. Returns the count of removed frames.

Called from SagaCoordinator::claim_terminal when a saga terminates (Done / Failed / timeout / eviction / etc.) — the terminal event closes the saga bracket, so any frames the saga left in the pending buffer would, if drained on reconnect, produce orphan side effects after the bracket is already closed (and risk a second SagaFailed via the 30s expiry path’s emit_drop_failure). Purging here keeps the saga’s terminal slot atomic across both the bus and the wire. (codex + reagent P1 PR #644 round 7.)

async fn drain_pending_to( &self, writer: &Arc<Mutex<Box<dyn AsyncWrite + Unpin + Send>>>, )

Drain pending_buffer through the given writer in FIFO order. Used by send_frame’s ptr_eq-mismatch path to flush a frame that landed in the buffer after a writer-replacement race (codex P1 PR #642 round 4). Symmetric with set_writer’s drain logic but operates on an existing writer rather than installing a new one. On per-frame failure, re-buffers the remaining frames at the front and arms the disconnect timer.

pub async fn send_command(&self, cmd: &Command) -> Result<(), HostPipeError>

Push a Command down to the host. On disconnect, buffers up to PENDING_BUFFER_CAP; on overflow or 30s timeout, drops + emits Event::SagaFailed for the dropped Command’s saga.

CPD-3 — called from the saga coordinator’s apply_action when dispatching IssueCmd::Host actions.

pub async fn try_send_command_no_buffer( &self, cmd: &Command, ) -> Result<(), HostPipeError>

Fail-fast variant of send_command: when no host writer is installed, returns HostNotConnected immediately instead of buffering (the variant the HostPipeError docs reserved for exactly this).

For frames where DELAYED delivery is meaningless or misleading — the UI-liveness probe (SPEC_LAUNCHER_TEARDOWN_BACKSTOP Phase 1) is the first caller: a probe buffered while the host is down would either expire silently in the pending buffer (probes carry no saga_id, so the drop paths can’t even report the loss) or be delivered to a FUTURE host, and in both cases the caller’s outstanding-probe bookkeeping would age into a false “UI thread did not pump” miss (reagent P1, PR #2106 round 2 — the routine disconnected-send during every crash-restart gap hit this).

Residual race, accepted and bounded: a disconnect landing between this check and the delegated write can still buffer the frame — that window is hair-width (same lock, adjacent awaits) versus the routine seconds-long disconnected spans this closes, and if it ever fires the resulting one-tick miss coincides with the pipe supervision’s own loud disconnect logs for the same instant, so the telemetry stays attributable.

async fn purge_pending_probe_frames(&self)

Remove every buffered ProbeUiThread frame. See try_send_command_no_buffer’s error branch for why these can appear despite the fail-fast contract, and why any such frame is stale by construction.

async fn send_frame( &self, frame: HostFrame, saga_id: Option<u64>, ) -> Result<(), HostPipeError>

async fn expire_pending_if_timed_out(&self)

If the disconnect timer has elapsed, flush the pending buffer and emit Event::SagaFailed for every buffered Command.

async fn emit_drop_failure(&self, dropped: PendingFrame, reason: &str)

Trait Implementations

impl Clone for HostPipe

fn clone(&self) -> HostPipe

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for HostPipe

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.