agentmux_launcher\reducer/
mod.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3//
4// Phase B.3 — pure reducer.
5//
6// Per `specs/SPEC_WINDOW_PROCESS_STATE_MACHINE_2026_04_27.md` §5.1:
7//
8//   pub fn update(state: &State, cmd: Command) -> (State, Vec<Event>);
9//
10// We deviate slightly: the function takes `&mut State` (mutating in
11// place rather than returning a new value) because cloning a HashMap
12// per-command in the IPC hot path is wasteful and not needed for
13// the testability properties we want — `proptest` works equally well
14// against a mutating-reducer (apply commands one by one, assert
15// invariants hold after each).
16//
17// Strict properties of `update`:
18//   1. **Total**. Every (cmd, state) combination produces a result;
19//      never panics on input. (Panics ARE used to enforce internal
20//      invariants — those are bugs in the reducer or upstream code,
21//      not user input. See spec §7 / §8.)
22//   2. **Deterministic**. Same (state, cmd, conn) → same (state',
23//      events). No clocks, no UUIDs, no env reads inside update —
24//      injected via the `Reducer` context if needed (B.4 will need
25//      it for spawned_at timestamps; for B.3 the conn carries one).
26//   3. **No I/O**. update never blocks or awaits. Mutex is held for
27//      the duration of update — must stay sub-millisecond.
28//
29// Connection context: every command arrives over a specific
30// connection and the resulting events that are *replies* (Registered,
31// Pong) belong to that connection only, while *broadcasts*
32// (ProcessSpawned, LifecyclePhaseChanged) belong to all subscribers.
33// Phase B.3 ships in a simplified model where every event goes over
34// the originating connection; B.4 splits the routing.
35//
36// Invariants checked:
37//   * Register on a PID that's already registered → AlreadyRegistered
38//     error. (Connection-level enforcement that "Register sent twice"
39//     also lives in the server; this is the cross-connection guard.)
40//   * Lifecycle transitions: Starting → Running on first Host
41//     register. Running → Quitting on Quit (B.3 placeholder).
42//     Quitting → Dead on cleanup-done (B.3 placeholder). No skipping.
43
44use agentmux_common::ipc::{Command, DriftKind, Event};
45
46use crate::state::State;
47
48/// Context the reducer needs but can't read from State (clocks,
49/// connection identity). Passed in per-call so update remains pure.
50#[derive(Debug, Clone)]
51pub struct Ctx {
52    /// RFC3339 timestamp to stamp on new ProcessRecords. Injected
53    /// (rather than read from chrono::Utc::now() inside update) to
54    /// keep the function deterministic for tests.
55    pub now_rfc3339: String,
56    /// The connection on which this command arrived. Currently
57    /// just used for log correlation; B.4+ will use it to route
58    /// per-connection replies.
59    pub conn_id: u64,
60    /// PID the connection has Registered under, if any. The server
61    /// tracks this server-side and passes it on every command after
62    /// the initial Register so the reducer can mark the right
63    /// process Exited on Goodbye. None for the very first command
64    /// on a connection (which MUST be Register, enforced server-
65    /// side). (codex P1 + gemini HIGH PR #574 round-1.)
66    pub registered_pid: Option<u32>,
67    /// Phase B.9.1 — monotonic milliseconds since some reference
68    /// epoch (the IPC server uses launcher start as the epoch).
69    /// Used by the WRR arm for per-window observability timestamps
70    /// (`last_foreground_at_ms`, `pending_hwnds[hwnd].arrived_at_ms`).
71    /// Distinct from `now_rfc3339` (wall clock) because operators
72    /// reading `--diag wrr` want age, not absolute time.
73    pub now_ms: u64,
74}
75
76mod pool;
77mod window;
78mod saga;
79mod connection;
80
81/// Apply one Command to State, returning the resulting Events. State
82/// is mutated in place. Total function — never panics on input
83/// (panics are reserved for internal invariant violations).
84pub fn update(state: &mut State, cmd: Command, ctx: &Ctx) -> Vec<Event> {
85    let _ = ctx.conn_id; // reserved for B.4 routing
86
87    let mut cmd_events = match cmd {
88        Command::Register {
89            kind,
90            pid,
91            version,
92        } => connection::handle_register(state, ctx, kind, pid, version),
93        Command::Ping { nonce } => {
94            let v = state.bump_version();
95            vec![Event::Pong { nonce, version: v }]
96        }
97        Command::Goodbye => connection::handle_goodbye(state, ctx.registered_pid.unwrap_or(0)),
98        Command::ReportWindowOpened {
99            label,
100            kind,
101            parent_label,
102        } => {
103            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportWindowOpened") {
104                return vec![err];
105            }
106            window::handle_report_window_opened(state, ctx, label, kind, parent_label)
107        }
108        Command::ReportWindowClosed { label } => {
109            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportWindowClosed") {
110                return vec![err];
111            }
112            window::handle_report_window_closed(state, label)
113        }
114        Command::ReportPoolWindowAdded { label, saga_id } => {
115            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportPoolWindowAdded") {
116                return vec![err];
117            }
118            pool::handle_report_pool_window_added(state, label, saga_id)
119        }
120        Command::ReportPoolWindowRemoved { label } => {
121            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportPoolWindowRemoved") {
122                return vec![err];
123            }
124            pool::handle_report_pool_window_removed(state, label)
125        }
126        // Phase F.5 — host-only signal that a pool→window promote is
127        // happening. Sent BETWEEN the matching `ReportPoolWindowRemoved`
128        // and `ReportWindowOpened` so the launcher can disambiguate
129        // promote from pre-promote destroy. Pure-reducer arm: emits
130        // the typed event; saga side-effect (start the pool-respawn
131        // saga) lives in the saga coordinator's bus subscription.
132        Command::ReportPoolWindowPromoted { label } => {
133            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportPoolWindowPromoted") {
134                return vec![err];
135            }
136            pool::handle_report_pool_window_promoted(state, label)
137        }
138        // Phase F.5 — `SpawnPoolWindow` is a launcher→host direction
139        // command, NOT a host→launcher report. If a registered client
140        // sends it to the launcher pipe by mistake, return a non-fatal
141        // error so the client knows the dispatch was wrong (vs silently
142        // appearing successful with no reply). Same misrouted-error
143        // pattern as the srv-pipe commands below.
144        Command::SpawnPoolWindow { .. } => {
145            let v = state.bump_version();
146            vec![Event::Error {
147                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
148                message: "SpawnPoolWindow is a launcher→host command; sent to launcher pipe by mistake".into(),
149                fatal: false,
150                version: v,
151            }]
152        }
153        // Phase F.6 — host-only signal that all browser-pane HWNDs
154        // belonging to a closing top-level window have been reaped.
155        // Pure-reducer arm: emits the typed event so the
156        // window-cleanup-cascade saga can advance from Step 1
157        // (ReapingPanes) to Step 2 (DrainingPool). State is
158        // unchanged — pane bookkeeping lives in the host's session
159        // structures (lifecycle entries, pane HWND map), not the
160        // launcher's mirror; the launcher just narrates the
161        // transition for subscribers.
162        Command::ReportPanesReaped { label, saga_id } => {
163            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportPanesReaped") {
164                return vec![err];
165            }
166            saga::handle_report_panes_reaped(state, label, saga_id)
167        }
168        // Phase F.6 — host reports the result of the post-close
169        // drain-pool-if-last decision. `was_last == true` →
170        // `Event::PoolDrained`; `was_last == false` →
171        // `Event::PoolNotLast`. Both are terminal alternatives for
172        // the window-cleanup-cascade saga's Step 2.
173        Command::ReportPoolDrainDecision {
174            label,
175            was_last,
176            saga_id,
177        } => {
178            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportPoolDrainDecision") {
179                return vec![err];
180            }
181            pool::handle_report_pool_drain_decision(state, label, was_last, saga_id)
182        }
183        // Phase F.6 — `ReapPanes` and `DrainPoolIfLast` are
184        // launcher→host direction commands, NOT host→launcher
185        // reports. Same misrouted-error pattern as `SpawnPoolWindow`
186        // above.
187        Command::ReapPanes { .. } => {
188            let v = state.bump_version();
189            vec![Event::Error {
190                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
191                message: "ReapPanes is a launcher→host command; sent to launcher pipe by mistake".into(),
192                fatal: false,
193                version: v,
194            }]
195        }
196        Command::DrainPoolIfLast { .. } => {
197            let v = state.bump_version();
198            vec![Event::Error {
199                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
200                message: "DrainPoolIfLast is a launcher→host command; sent to launcher pipe by mistake".into(),
201                fatal: false,
202                version: v,
203            }]
204        }
205        // Phase CPD-1 — host-emitted report that a saga-issued action
206        // failed. Pure pass-through arm: translates the wire command
207        // into `Event::SagaActionFailed`. The saga coordinator's bus
208        // loop (CPD-3) will treat this as a terminal signal for the
209        // matching `saga_id` and emit `Event::SagaFailed`. Host-only
210        // gate same as other Report* arms.
211        Command::ReportSagaActionFailed { saga_id, reason } => {
212            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportSagaActionFailed") {
213                return vec![err];
214            }
215            saga::handle_report_saga_action_failed(state, saga_id, reason)
216        }
217        Command::ReportHostCounts { windows, pool } => {
218            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportHostCounts") {
219                return vec![err];
220            }
221            handle_report_host_counts(state, windows, pool)
222        }
223        Command::ReportHostPoolCount { count } => {
224            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportHostPoolCount") {
225                return vec![err];
226            }
227            pool::handle_report_host_pool_count(state, count)
228        }
229        Command::ReportBackendWindowIdRegistered { label, window_id } => {
230            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportBackendWindowIdRegistered") {
231                return vec![err];
232            }
233            window::handle_report_backend_window_id_registered(state, label, window_id)
234        }
235        Command::ReportBackendWindowIdUnregistered { label } => {
236            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportBackendWindowIdUnregistered") {
237                return vec![err];
238            }
239            window::handle_report_backend_window_id_unregistered(state, label)
240        }
241        // Phase B.9.1 (WRR) — Win32 reality events. Host-only
242        // because only the host installs the SetWinEventHook /
243        // wndproc wrapper. Same enforce-host pattern as the other
244        // observability reports.
245        Command::ReportHwndOpened { hwnd, class_name, title, label_hint } => {
246            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportHwndOpened") {
247                return vec![err];
248            }
249            crate::wrr::apply_hwnd_opened(state, hwnd, class_name, title, label_hint, ctx.now_ms)
250        }
251        Command::ReportHwndDestroyed { hwnd } => {
252            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportHwndDestroyed") {
253                return vec![err];
254            }
255            let host_running = connection::host_is_running(state);
256            crate::wrr::apply_hwnd_destroyed(state, hwnd, host_running)
257        }
258        Command::ReportHwndVisibilityChanged { hwnd, visible } => {
259            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportHwndVisibilityChanged") {
260                return vec![err];
261            }
262            crate::wrr::apply_hwnd_visibility_changed(state, hwnd, visible, ctx.now_ms)
263        }
264        Command::ReportHwndForegroundChanged { hwnd } => {
265            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportHwndForegroundChanged") {
266                return vec![err];
267            }
268            crate::wrr::apply_hwnd_foreground_changed(state, hwnd, ctx.now_ms)
269        }
270        Command::ReportHwndIconicChanged { hwnd, iconic } => {
271            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportHwndIconicChanged") {
272                return vec![err];
273            }
274            crate::wrr::apply_hwnd_iconic_changed(state, hwnd, iconic)
275        }
276        Command::ReportHwndPositionChanged { hwnd, rect } => {
277            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportHwndPositionChanged") {
278                return vec![err];
279            }
280            crate::wrr::apply_hwnd_position_changed(state, hwnd, rect)
281        }
282        Command::ReportMonitorTopologyChanged { rects } => {
283            if let Some(err) = connection::enforce_host_only(state, ctx, "ReportMonitorTopologyChanged") {
284                return vec![err];
285            }
286            crate::wrr::apply_monitor_topology_changed(state, rects)
287        }
288        // Phase D.1 — read-only snapshot of canonical state. Any
289        // registered client can ask; no host-only gate. Reducer state
290        // is unchanged by this command (it's a query, not a mutation),
291        // but we still bump `event_version` so the snapshot's version
292        // is monotonically distinct from prior events — a subscriber
293        // applying snapshot + delta events knows the snapshot's
294        // version is the "as-of" point.
295        Command::GetSnapshot => connection::handle_get_snapshot(state),
296        // Phase D.3 — `GetEvents` is intercepted by the IPC server's
297        // dispatch path BEFORE reaching the reducer (it's a non-
298        // mutating read against the event log, which is I/O-adjacent
299        // — keeping it out of the pure reducer preserves the
300        // "reducer never blocks" invariant). This arm exists only
301        // to satisfy the exhaustive match; in practice it's
302        // unreachable. Returning empty Vec is the safe no-op.
303        Command::GetEvents { .. } => Vec::new(),
304        // Phase E.2 — srv-pipe domain commands routed to launcher
305        // pipe by mistake.
306        Command::CreateWorkspace { .. } => {
307            let v = state.bump_version();
308            vec![Event::Error {
309                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
310                message: "CreateWorkspace is a srv-pipe command; sent to launcher pipe by mistake".into(),
311                fatal: false,
312                version: v,
313            }]
314        }
315        Command::DeleteWorkspace { .. } => {
316            let v = state.bump_version();
317            vec![Event::Error {
318                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
319                message: "DeleteWorkspace is a srv-pipe command; sent to launcher pipe by mistake".into(),
320                fatal: false,
321                version: v,
322            }]
323        }
324        // Phase E.2b — Tab arms are also srv-pipe commands.
325        Command::CreateTab { .. } => {
326            let v = state.bump_version();
327            vec![Event::Error {
328                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
329                message: "CreateTab is a srv-pipe command; sent to launcher pipe by mistake".into(),
330                fatal: false,
331                version: v,
332            }]
333        }
334        Command::DeleteTab { .. } => {
335            let v = state.bump_version();
336            vec![Event::Error {
337                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
338                message: "DeleteTab is a srv-pipe command; sent to launcher pipe by mistake".into(),
339                fatal: false,
340                version: v,
341            }]
342        }
343        Command::SetActiveTab { .. } => {
344            let v = state.bump_version();
345            vec![Event::Error {
346                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
347                message: "SetActiveTab is a srv-pipe command; sent to launcher pipe by mistake".into(),
348                fatal: false,
349                version: v,
350            }]
351        }
352        Command::ReorderTab { .. } => {
353            let v = state.bump_version();
354            vec![Event::Error {
355                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
356                message: "ReorderTab is a srv-pipe command; sent to launcher pipe by mistake".into(),
357                fatal: false,
358                version: v,
359            }]
360        }
361        // Phase E.5 — window↔workspace mapping commands are srv-pipe.
362        Command::CreateWindow { .. } => {
363            let v = state.bump_version();
364            vec![Event::Error {
365                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
366                message: "CreateWindow is a srv-pipe command; sent to launcher pipe by mistake".into(),
367                fatal: false,
368                version: v,
369            }]
370        }
371        Command::CloseWindowInternal { .. } => {
372            let v = state.bump_version();
373            vec![Event::Error {
374                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
375                message: "CloseWindowInternal is a srv-pipe command; sent to launcher pipe by mistake".into(),
376                fatal: false,
377                version: v,
378            }]
379        }
380        Command::SwitchWorkspace { .. } => {
381            let v = state.bump_version();
382            vec![Event::Error {
383                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
384                message: "SwitchWorkspace is a srv-pipe command; sent to launcher pipe by mistake".into(),
385                fatal: false,
386                version: v,
387            }]
388        }
389        // Phase E.5.3 — atomic single-step domain commands are srv-pipe.
390        Command::ReorderTabsBulk { .. } => {
391            let v = state.bump_version();
392            vec![Event::Error {
393                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
394                message: "ReorderTabsBulk is a srv-pipe command; sent to launcher pipe by mistake".into(),
395                fatal: false,
396                version: v,
397            }]
398        }
399        Command::RenameWorkspace { .. } => {
400            let v = state.bump_version();
401            vec![Event::Error {
402                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
403                message: "RenameWorkspace is a srv-pipe command; sent to launcher pipe by mistake".into(),
404                fatal: false,
405                version: v,
406            }]
407        }
408        Command::RenameTab { .. } => {
409            let v = state.bump_version();
410            vec![Event::Error {
411                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
412                message: "RenameTab is a srv-pipe command; sent to launcher pipe by mistake".into(),
413                fatal: false,
414                version: v,
415            }]
416        }
417        Command::UpdateWorkspaceMeta { .. } => {
418            let v = state.bump_version();
419            vec![Event::Error {
420                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
421                message: "UpdateWorkspaceMeta is a srv-pipe command; sent to launcher pipe by mistake".into(),
422                fatal: false,
423                version: v,
424            }]
425        }
426        Command::UpdateTabMeta { .. } => {
427            let v = state.bump_version();
428            vec![Event::Error {
429                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
430                message: "UpdateTabMeta is a srv-pipe command; sent to launcher pipe by mistake".into(),
431                fatal: false,
432                version: v,
433            }]
434        }
435        Command::UpdateBlockMeta { .. } => {
436            let v = state.bump_version();
437            vec![Event::Error {
438                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
439                message: "UpdateBlockMeta is a srv-pipe command; sent to launcher pipe by mistake".into(),
440                fatal: false,
441                version: v,
442            }]
443        }
444        // Phase E.3 — Block arms are also srv-pipe commands.
445        Command::CreateBlock { .. } => {
446            let v = state.bump_version();
447            vec![Event::Error {
448                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
449                message: "CreateBlock is a srv-pipe command; sent to launcher pipe by mistake".into(),
450                fatal: false,
451                version: v,
452            }]
453        }
454        Command::DeleteBlock { .. } => {
455            let v = state.bump_version();
456            vec![Event::Error {
457                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
458                message: "DeleteBlock is a srv-pipe command; sent to launcher pipe by mistake".into(),
459                fatal: false,
460                version: v,
461            }]
462        }
463        // Phase E.5.5 — saga-driven move commands are srv-pipe.
464        Command::MoveTab { .. } => {
465            let v = state.bump_version();
466            vec![Event::Error {
467                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
468                message: "MoveTab is a srv-pipe command; sent to launcher pipe by mistake".into(),
469                fatal: false,
470                version: v,
471            }]
472        }
473        Command::MoveBlock { .. } => {
474            let v = state.bump_version();
475            vec![Event::Error {
476                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
477                message: "MoveBlock is a srv-pipe command; sent to launcher pipe by mistake".into(),
478                fatal: false,
479                version: v,
480            }]
481        }
482        // Phase E.4 (Option A) — layout-focused/magnified setters are
483        // srv-pipe commands. Misrouted to the launcher pipe, return
484        // a non-fatal error so the client knows the dispatch was wrong.
485        Command::SetFocusedNode { .. } => {
486            let v = state.bump_version();
487            vec![Event::Error {
488                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
489                message: "SetFocusedNode is a srv-pipe command; sent to launcher pipe by mistake".into(),
490                fatal: false,
491                version: v,
492            }]
493        }
494        Command::SetMagnifiedNode { .. } => {
495            let v = state.bump_version();
496            vec![Event::Error {
497                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
498                message: "SetMagnifiedNode is a srv-pipe command; sent to launcher pipe by mistake".into(),
499                fatal: false,
500                version: v,
501            }]
502        }
503        // Phase E.1b — `GetSrvSnapshot` is a srv-pipe command. If a
504        // registered client misroutes it to the launcher pipe, return
505        // an explicit error so the client knows the dispatch was
506        // wrong (vs silently appearing successful with no reply).
507        // Pre-Register, `enforce_register_first` already returns a
508        // soft error. (codex P2 #610.)
509        Command::GetSrvSnapshot => {
510            let v = state.bump_version();
511            vec![Event::Error {
512                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
513                message: "GetSrvSnapshot is a srv-pipe command; sent to launcher pipe by mistake".to_string(),
514                fatal: false,
515                version: v,
516            }]
517        }
518        // Phase E.4.B — layout-tree commands are srv-pipe only; same
519        // soft-error treatment as GetSrvSnapshot above.
520        Command::LayoutInsertNode { .. }
521        | Command::LayoutInsertNodeAtIndex { .. }
522        | Command::LayoutDeleteNode { .. }
523        | Command::LayoutMoveNode { .. }
524        | Command::LayoutSwapNodes { .. }
525        | Command::LayoutResizeNodes { .. }
526        | Command::LayoutReplaceNode { .. }
527        | Command::LayoutSplitHorizontal { .. }
528        | Command::LayoutSplitVertical { .. }
529        | Command::LayoutClear { .. }
530        | Command::LayoutSetTree { .. }
531        | Command::UpdateWindowMeta { .. } => {
532            let v = state.bump_version();
533            vec![Event::Error {
534                code: agentmux_common::ipc::ErrorCode::InvalidCommand,
535                message: "Srv-pipe command (Layout/UpdateWindowMeta) sent to launcher pipe by mistake".to_string(),
536                fatal: false,
537                version: v,
538            }]
539        }
540    };
541
542    // Heartbeat-via-traffic: drain any deferred hidden-since-open
543    // drifts AFTER the command processes. Running this AFTER (not
544    // before) is critical — the command itself may legitimately clear
545    // the deferred state (visible=true / foreground / window closed),
546    // and a pre-command drain would fire spurious drift on an event
547    // whose own purpose is the recovery (codex P2 PR #725 round 2).
548    //
549    // Catches windows that hid during placement and produced no
550    // further visibility events: any subsequent unrelated command
551    // past the grace promotes the deferred state to a fired drift.
552    let mut deferred = crate::wrr::drain_deferred_hidden_since_open(state, ctx.now_ms);
553    cmd_events.append(&mut deferred);
554    cmd_events
555}
556
557
558
559/// Phase B.4 follow-up — drift check. Compares host-reported counts
560/// to launcher mirror counts; emits `DriftDetected` for each
561/// disagreeing dimension. Returns `[]` when both counts match (the
562/// happy path; mirrors are in sync).
563///
564/// Two events possible (windows + pool) when both diverge in the
565/// same report — emitted in a stable order (windows first) so test
566/// assertions don't depend on HashSet iteration order.
567fn handle_report_host_counts(state: &mut State, host_windows: u32, host_pool: u32) -> Vec<Event> {
568    let mut out = Vec::new();
569    let mirror_windows = state.windows.len() as u32;
570    let mirror_pool = state.pool.len() as u32;
571    if mirror_windows != host_windows {
572        let v = state.bump_version();
573        out.push(Event::DriftDetected {
574            kind: DriftKind::Windows,
575            host_count: host_windows,
576            mirror_count: mirror_windows,
577            version: v,
578        });
579    }
580    if mirror_pool != host_pool {
581        let v = state.bump_version();
582        out.push(Event::DriftDetected {
583            kind: DriftKind::Pool,
584            host_count: host_pool,
585            mirror_count: mirror_pool,
586            version: v,
587        });
588    }
589    out
590}
591
592
593
594
595
596
597
598
599
600
601#[cfg(test)]
602mod tests;
agentmux_launcher\reducer/mod.rs

agentmux_launcher\reducer/
mod.rs
