agentmux_srv/reducer/
lifecycle.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4use agentmux_common::ipc::{ErrorCode, Event};
5
6use crate::state::State;
7
8use super::Ctx;
9
10use agentmux_common::ipc::{ClientKind, LifecyclePhase};
11use crate::state::{ProcessRecord, ProcessState};
12
13pub(super) fn handle_register(
14    state: &mut State,
15    kind: ClientKind,
16    pid: u32,
17    version: String,
18) -> Vec<Event> {
19    // Idempotent on duplicate Register from the same PID — preserve
20    // the original record (per launcher's pattern). Accept fresh
21    // Registers only when the PID has no record OR the existing
22    // record is Exited (PID recycled).
23    let prior_state = state.processes.get(&pid).map(|r| r.state);
24    let allow_register = match prior_state {
25        None => true,
26        Some(ProcessState::Exited { .. }) => true,
27        _ => false,
28    };
29    if !allow_register {
30        let v = state.bump_version();
31        return vec![Event::Error {
32            code: ErrorCode::AlreadyRegistered,
33            message: format!("pid {} is already registered with srv", pid),
34            fatal: false,
35            version: v,
36        }];
37    }
38
39    let mut out = Vec::with_capacity(3);
40
41    // Insert the process record.
42    state.processes.insert(
43        pid,
44        ProcessRecord {
45            state: ProcessState::Running,
46            #[cfg(test)]
47            version: version.clone(),
48        },
49    );
50    let v = state.bump_version();
51    out.push(Event::ProcessSpawned {
52        pid,
53        kind,
54        client_version: version,
55        version: v,
56    });
57
58    // First Register transitions srv to Running. Same lifecycle
59    // pattern as launcher.
60    if state.lifecycle == LifecyclePhase::Starting {
61        state.lifecycle = LifecyclePhase::Running;
62        let v = state.bump_version();
63        out.push(Event::LifecyclePhaseChanged {
64            from: LifecyclePhase::Starting,
65            to: LifecyclePhase::Running,
66            version: v,
67        });
68    }
69
70    let client_id = state.alloc_client_id();
71    let v = state.bump_version();
72    // Sentinel launcher_pid / launcher_version on Registered — IPC
73    // server patches these to the real srv identity before broadcast.
74    out.push(Event::Registered {
75        client_id,
76        launcher_pid: 0,
77        launcher_version: String::new(),
78        version: v,
79    });
80    out
81}
82
83pub(super) fn handle_goodbye(state: &mut State, ctx: &Ctx) -> Vec<Event> {
84    let Some(pid) = ctx.registered_pid else {
85        return Vec::new();
86    };
87    let Some(record) = state.processes.get_mut(&pid) else {
88        return Vec::new();
89    };
90    if matches!(record.state, ProcessState::Exited { .. }) {
91        return Vec::new();
92    }
93    record.state = ProcessState::Exited { code: 0 };
94    let v = state.bump_version();
95    vec![Event::ProcessExited {
96        pid,
97        code: 0,
98        version: v,
99    }]
100}
101
102#[cfg(test)]
103mod tests {
104    use super::*;
105    use crate::reducer::test_support::*;
106    use crate::reducer::update;
107    use agentmux_common::ipc::Command;
108
109    #[test]
110    fn first_register_transitions_to_running() {
111        let mut state = State::default();
112        let events = update(
113            &mut state,
114            Command::Register {
115                kind: ClientKind::Host,
116                pid: 100,
117                version: "0.0.0".into(),
118            },
119            &ctx(1),
120        );
121        assert_eq!(state.lifecycle, LifecyclePhase::Running);
122        assert!(state.processes.contains_key(&100));
123        assert_eq!(events.len(), 3);
124        assert!(matches!(&events[0], Event::ProcessSpawned { pid: 100, .. }));
125        assert!(matches!(
126            &events[1],
127            Event::LifecyclePhaseChanged {
128                from: LifecyclePhase::Starting,
129                to: LifecyclePhase::Running,
130                ..
131            }
132        ));
133        assert!(matches!(&events[2], Event::Registered { .. }));
134    }
135
136    #[test]
137    fn second_register_does_not_re_emit_lifecycle_change() {
138        let mut state = State::default();
139        let _ = update(
140            &mut state,
141            Command::Register {
142                kind: ClientKind::Host,
143                pid: 100,
144                version: "v1".into(),
145            },
146            &ctx(1),
147        );
148        let events = update(
149            &mut state,
150            Command::Register {
151                kind: ClientKind::Tool,
152                pid: 200,
153                version: "v2".into(),
154            },
155            &ctx(2),
156        );
157        assert_eq!(state.lifecycle, LifecyclePhase::Running);
158        // ProcessSpawned + Registered, no LifecyclePhaseChanged.
159        assert_eq!(events.len(), 2);
160        assert!(events
161            .iter()
162            .all(|e| !matches!(e, Event::LifecyclePhaseChanged { .. })));
163    }
164
165    #[test]
166    fn duplicate_register_returns_already_registered_error() {
167        let mut state = State::default();
168        let _ = update(
169            &mut state,
170            Command::Register {
171                kind: ClientKind::Host,
172                pid: 100,
173                version: "v1".into(),
174            },
175            &ctx(1),
176        );
177        let events = update(
178            &mut state,
179            Command::Register {
180                kind: ClientKind::Host,
181                pid: 100,
182                version: "v2".into(),
183            },
184            &ctx(2),
185        );
186        assert!(matches!(
187            &events[0],
188            Event::Error {
189                code: ErrorCode::AlreadyRegistered,
190                ..
191            }
192        ));
193        // Original record preserved.
194        assert_eq!(&state.processes[&100].version, "v1");
195    }
196
197    #[test]
198    fn register_replaces_exited_record_for_recycled_pid() {
199        let mut state = State::default();
200        let _ = update(
201            &mut state,
202            Command::Register {
203                kind: ClientKind::Host,
204                pid: 100,
205                version: "v1".into(),
206            },
207            &ctx(1),
208        );
209        let _ = update(&mut state, Command::Goodbye, &ctx_with_pid(1, 100));
210        // Re-register with same PID — allowed because prior is Exited.
211        let events = update(
212            &mut state,
213            Command::Register {
214                kind: ClientKind::Host,
215                pid: 100,
216                version: "v2".into(),
217            },
218            &ctx(2),
219        );
220        assert!(matches!(&events[0], Event::ProcessSpawned { .. }));
221        assert_eq!(&state.processes[&100].version, "v2");
222        assert!(matches!(state.processes[&100].state, ProcessState::Running));
223    }
224
225    #[test]
226    fn ping_returns_pong_with_same_nonce() {
227        let mut state = State::default();
228        let events = update(&mut state, Command::Ping { nonce: 42 }, &ctx(1));
229        assert!(matches!(&events[0], Event::Pong { nonce: 42, .. }));
230    }
231
232    #[test]
233    fn goodbye_marks_process_exited() {
234        let mut state = State::default();
235        let _ = update(
236            &mut state,
237            Command::Register {
238                kind: ClientKind::Host,
239                pid: 100,
240                version: "v".into(),
241            },
242            &ctx(1),
243        );
244        let events = update(&mut state, Command::Goodbye, &ctx_with_pid(1, 100));
245        assert!(matches!(
246            &events[0],
247            Event::ProcessExited { pid: 100, code: 0, .. }
248        ));
249        assert!(matches!(
250            state.processes[&100].state,
251            ProcessState::Exited { code: 0 }
252        ));
253    }
254
255    #[test]
256    fn versions_strictly_monotonic_across_sequence() {
257        let mut state = State::default();
258        let mut versions = vec![];
259        for pid in [100, 200, 300] {
260            let events = update(
261                &mut state,
262                Command::Register {
263                    kind: ClientKind::Host,
264                    pid,
265                    version: "v".into(),
266                },
267                &ctx(pid as u64),
268            );
269            versions.extend(events.iter().map(extract_version));
270        }
271        for w in versions.windows(2) {
272            assert!(w[1] > w[0], "version regression: {} -> {}", w[0], w[1]);
273        }
274    }
275}
agentmux_srv/reducer/lifecycle.rs

agentmux_srv/reducer/
lifecycle.rs
