codeswarm_core/
lib.rs

1//! Framework-independent CodeSwarm domain state.
2//!
3//! The terminal UI and each CLI protocol adapter communicate through this
4//! event vocabulary. The reducer remains synchronous and deterministic so
5//! recorded sessions can be replayed without a terminal or subprocess.
6
7use std::collections::VecDeque;
8use std::fs::{File, OpenOptions};
9use std::io::{BufRead, BufReader, BufWriter, Write};
10use std::path::{Path, PathBuf};
11use std::sync::mpsc::{self, Receiver, Sender};
12
13use serde::{Deserialize, Serialize};
14
15pub mod agents;
16pub mod collaboration;
17pub mod contract;
18pub mod details;
19pub mod history;
20pub mod launcher;
21pub mod persistence;
22pub mod policy;
23pub mod relay;
24pub mod resources;
25pub mod settings;
26pub mod trace;
27
28/// Stable roster position in the coordinator-managed agent list.
29pub type RosterSlot = usize;
30
31#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
32pub struct Mode {
33    pub id: String,
34    pub label: String,
35}
36
37#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
38pub struct AgentCapabilities {
39    pub supports_cancel: bool,
40    pub supports_modes: bool,
41    pub supports_permissions: bool,
42    pub supports_terminals: bool,
43    pub supports_session_load: bool,
44    #[serde(default)]
45    pub supports_models: bool,
46}
47
48#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
49pub struct ToolUpdate {
50    pub id: String,
51    pub title: String,
52    pub status: ToolStatus,
53    pub detail: Option<String>,
54}
55
56/// A slash command advertised by an ACP session.  The renderer intentionally
57/// keeps only the command name at the shared event boundary; descriptions and
58/// input hints are provider-specific presentation data and are not needed to
59/// dispatch or complete a command.
60#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
61pub struct AgentCommand {
62    pub name: String,
63}
64
65/// The latest context-window counters reported by an ACP session.
66#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
67pub struct UsageUpdate {
68    pub used: u64,
69    pub size: u64,
70}
71
72#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
73pub enum ToolStatus {
74    Pending,
75    Running,
76    Completed,
77    Failed,
78}
79
80#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
81pub struct PermissionRequest {
82    pub id: String,
83    pub title: String,
84    pub options: Vec<String>,
85    /// Protocol identities aligned by index with `options`. Empty entries
86    /// fall back to the visible option label for legacy/native adapters.
87    #[serde(default)]
88    pub option_ids: Vec<String>,
89}
90
91/// The normalized answer to an adapter permission request. Native adapters
92/// may reject this explicitly when their protocol has no permission control.
93#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
94pub enum PermissionAnswer {
95    Selected { option_id: String },
96    Cancelled,
97}
98
99#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
100pub enum TerminalEvent {
101    Created { id: String, command: String },
102    Output { id: String, text: String },
103    Exited { id: String, code: i32 },
104    Released { id: String },
105}
106
107#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
108pub enum RosterUpdate {
109    Added {
110        slot: RosterSlot,
111        name: String,
112        identity: String,
113    },
114    Reloaded {
115        slot: RosterSlot,
116    },
117    Dropped {
118        slot: RosterSlot,
119    },
120    Swapped {
121        first: RosterSlot,
122        second: RosterSlot,
123    },
124    Rejected {
125        action: String,
126        detail: String,
127    },
128}
129
130#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
131pub enum AgentEvent {
132    RosterUpdated {
133        update: RosterUpdate,
134    },
135    Ready {
136        slot: RosterSlot,
137        capabilities: AgentCapabilities,
138    },
139    TurnStarted {
140        slot: RosterSlot,
141    },
142    ModesReplaced {
143        slot: RosterSlot,
144        modes: Vec<Mode>,
145        current_mode: Option<String>,
146    },
147    ModeUpdated {
148        slot: RosterSlot,
149        current_mode: String,
150    },
151    ModelsReplaced {
152        slot: RosterSlot,
153        config_id: String,
154        models: Vec<Mode>,
155        current_model: Option<String>,
156    },
157    ModelUpdated {
158        slot: RosterSlot,
159        current_model: String,
160    },
161    UserText {
162        slot: RosterSlot,
163        text: String,
164    },
165    CommandsReplaced {
166        slot: RosterSlot,
167        commands: Vec<AgentCommand>,
168    },
169    UsageUpdated {
170        slot: RosterSlot,
171        usage: UsageUpdate,
172    },
173    Text {
174        slot: RosterSlot,
175        text: String,
176    },
177    Thought {
178        slot: RosterSlot,
179        text: String,
180    },
181    Tool {
182        slot: RosterSlot,
183        update: ToolUpdate,
184    },
185    Permission {
186        slot: RosterSlot,
187        request: PermissionRequest,
188    },
189    Terminal {
190        slot: RosterSlot,
191        event: TerminalEvent,
192    },
193    TurnComplete {
194        slot: RosterSlot,
195    },
196    Failed {
197        slot: RosterSlot,
198        started: bool,
199        detail: String,
200    },
201}
202
203#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
204pub enum Effect {
205    Render,
206    DispatchPrompt { slot: RosterSlot, prompt: String },
207    OfferReload { slot: RosterSlot, crashed: bool },
208}
209
210#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
211pub struct AgentSlot {
212    pub active: bool,
213    pub capabilities: AgentCapabilities,
214    pub modes: Vec<Mode>,
215    pub current_mode: Option<String>,
216    #[serde(default)]
217    pub models: Vec<Mode>,
218    #[serde(default)]
219    pub current_model: Option<String>,
220    #[serde(default)]
221    pub commands: Vec<AgentCommand>,
222    #[serde(default)]
223    pub usage: Option<UsageUpdate>,
224}
225
226impl Default for AgentSlot {
227    fn default() -> Self {
228        Self {
229            active: true,
230            capabilities: AgentCapabilities::default(),
231            modes: Vec::new(),
232            current_mode: None,
233            models: Vec::new(),
234            current_model: None,
235            commands: Vec::new(),
236            usage: None,
237        }
238    }
239}
240
241#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
242pub struct SessionState {
243    pub slots: Vec<AgentSlot>,
244    pub active_slot: Option<RosterSlot>,
245    pub queued_prompts: VecDeque<(RosterSlot, String)>,
246    pub public_text: Vec<(RosterSlot, String)>,
247}
248
249impl SessionState {
250    pub fn new(roster_size: usize) -> Self {
251        Self {
252            slots: (0..roster_size).map(|_| AgentSlot::default()).collect(),
253            active_slot: None,
254            queued_prompts: VecDeque::new(),
255            public_text: Vec::new(),
256        }
257    }
258}
259
260/// Apply one normalized event. I/O and rendering are represented by effects,
261/// never performed in the reducer.
262pub fn reduce(state: &mut SessionState, event: AgentEvent) -> Vec<Effect> {
263    match event {
264        AgentEvent::RosterUpdated { .. } => vec![Effect::Render],
265        AgentEvent::Ready { slot, capabilities } => {
266            if let Some(agent) = state.slots.get_mut(slot) {
267                agent.capabilities = capabilities;
268            }
269            vec![Effect::Render]
270        }
271        AgentEvent::TurnStarted { slot } => {
272            state.active_slot = Some(slot);
273            vec![Effect::Render]
274        }
275        AgentEvent::ModesReplaced {
276            slot,
277            modes,
278            current_mode,
279        } => {
280            if let Some(agent) = state.slots.get_mut(slot) {
281                agent.modes = modes;
282                agent.current_mode =
283                    current_mode.filter(|id| agent.modes.iter().any(|mode| mode.id == *id));
284            }
285            vec![Effect::Render]
286        }
287        AgentEvent::CommandsReplaced { slot, commands } => {
288            if let Some(agent) = state.slots.get_mut(slot) {
289                agent.commands = commands;
290            }
291            vec![Effect::Render]
292        }
293        AgentEvent::ModeUpdated { slot, current_mode } => {
294            if let Some(agent) = state.slots.get_mut(slot) {
295                agent.current_mode = Some(current_mode);
296            }
297            vec![Effect::Render]
298        }
299        AgentEvent::ModelsReplaced {
300            slot,
301            models,
302            current_model,
303            ..
304        } => {
305            if let Some(agent) = state.slots.get_mut(slot) {
306                agent.models = models;
307                agent.current_model =
308                    current_model.filter(|id| agent.models.iter().any(|model| model.id == *id));
309            }
310            vec![Effect::Render]
311        }
312        AgentEvent::ModelUpdated {
313            slot,
314            current_model,
315        } => {
316            if let Some(agent) = state.slots.get_mut(slot) {
317                agent.current_model = Some(current_model);
318            }
319            vec![Effect::Render]
320        }
321        AgentEvent::UsageUpdated { slot, usage } => {
322            if let Some(agent) = state.slots.get_mut(slot) {
323                agent.usage = Some(usage);
324            }
325            vec![Effect::Render]
326        }
327        AgentEvent::Text { slot, text } => {
328            state.active_slot = Some(slot);
329            state.public_text.push((slot, text));
330            vec![Effect::Render]
331        }
332        AgentEvent::UserText { slot, .. } => {
333            state.active_slot = Some(slot);
334            vec![Effect::Render]
335        }
336        AgentEvent::Thought { slot, .. }
337        | AgentEvent::Tool { slot, .. }
338        | AgentEvent::Permission { slot, .. }
339        | AgentEvent::Terminal { slot, .. } => {
340            state.active_slot = Some(slot);
341            vec![Effect::Render]
342        }
343        AgentEvent::TurnComplete { .. } => {
344            state.active_slot = None;
345            let next =
346                state
347                    .queued_prompts
348                    .pop_front()
349                    .map(|(target, prompt)| Effect::DispatchPrompt {
350                        slot: target,
351                        prompt,
352                    });
353            let mut effects = vec![Effect::Render];
354            if let Some(effect) = next {
355                effects.push(effect);
356            }
357            effects
358        }
359        AgentEvent::Failed {
360            slot,
361            started,
362            detail: _,
363        } => {
364            if let Some(agent) = state.slots.get_mut(slot) {
365                agent.active = false;
366            }
367            if state.active_slot == Some(slot) {
368                state.active_slot = None;
369            }
370            vec![
371                Effect::Render,
372                Effect::OfferReload {
373                    slot,
374                    crashed: started,
375                },
376            ]
377        }
378    }
379}
380
381/// A newline-delimited event log. It deliberately records normalized events
382/// rather than UI operations, so sessions can be replayed by a future renderer
383/// or adapter host. Each append closes its file handle but does not force a
384/// storage sync; the terminal input/render loop must never block on fsync.
385#[derive(Clone, Debug)]
386pub struct EventLog {
387    path: PathBuf,
388}
389
390impl EventLog {
391    pub fn open(path: impl Into<PathBuf>) -> Self {
392        Self { path: path.into() }
393    }
394
395    pub fn path(&self) -> &Path {
396        &self.path
397    }
398
399    pub fn append(&self, event: &AgentEvent) -> std::io::Result<()> {
400        let encoded = serde_json::to_string(event)
401            .map_err(|error| std::io::Error::other(error.to_string()))?;
402        let mut file = OpenOptions::new()
403            .create(true)
404            .append(true)
405            .open(&self.path)?;
406        file.write_all(encoded.as_bytes())?;
407        file.write_all(b"\n")
408    }
409
410    /// Append an event and force it to stable storage.
411    ///
412    /// The regular [`append`](Self::append) path is deliberately lightweight
413    /// because it is called from the terminal event loop. Call this only at a
414    /// durability boundary such as a completed turn or explicit shutdown.
415    pub fn append_durable(&self, event: &AgentEvent) -> std::io::Result<()> {
416        let encoded = serde_json::to_string(event)
417            .map_err(|error| std::io::Error::other(error.to_string()))?;
418        let mut file = OpenOptions::new()
419            .create(true)
420            .append(true)
421            .open(&self.path)?;
422        file.write_all(encoded.as_bytes())?;
423        file.write_all(b"\n")?;
424        file.sync_data()
425    }
426
427    /// Force already-appended records to stable storage without writing a
428    /// duplicate event. This is useful for callers that batch lightweight
429    /// appends and checkpoint at turn boundaries.
430    pub fn sync(&self) -> std::io::Result<()> {
431        let file = OpenOptions::new().read(true).open(&self.path)?;
432        file.sync_data()
433    }
434
435    /// Start a background writer for event-loop use. The returned handle
436    /// queues records in memory and performs all file I/O on its worker
437    /// thread. Use [`BufferedEventLog::flush`] at explicit durability
438    /// boundaries such as completed turns.
439    pub fn buffered(&self) -> std::io::Result<BufferedEventLog> {
440        BufferedEventLog::open(self.path.clone())
441    }
442
443    pub fn read(&self) -> std::io::Result<Vec<AgentEvent>> {
444        let file = match File::open(&self.path) {
445            Ok(file) => file,
446            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
447            Err(error) => return Err(error),
448        };
449        BufReader::new(file)
450            .lines()
451            .enumerate()
452            .filter_map(|(line_number, result)| match result {
453                Ok(line) if line.trim().is_empty() => None,
454                Ok(line) => Some(serde_json::from_str(&line).map_err(|error| {
455                    std::io::Error::new(
456                        std::io::ErrorKind::InvalidData,
457                        format!("event log line {}: {error}", line_number + 1),
458                    )
459                })),
460                Err(error) => Some(Err(error)),
461            })
462            .collect()
463    }
464
465    pub fn replay(&self, roster_size: usize) -> std::io::Result<SessionState> {
466        let mut state = SessionState::new(roster_size);
467        for event in self.read()? {
468            reduce(&mut state, event);
469        }
470        Ok(state)
471    }
472}
473
474enum BufferedLogCommand {
475    Append(String),
476    Flush(Sender<std::io::Result<()>>),
477    Shutdown(Sender<std::io::Result<()>>),
478}
479
480/// Background event-log writer used to keep terminal input/render handling
481/// independent from filesystem latency. The channel is intentionally
482/// unbounded: dropping normalized events during a streamed turn would make
483/// replay and recovery incomplete, while the writer drains ordinary event
484/// rates faster than adapters produce them.
485pub struct BufferedEventLog {
486    sender: Sender<BufferedLogCommand>,
487    worker: Option<std::thread::JoinHandle<std::io::Result<()>>>,
488}
489
490impl std::fmt::Debug for BufferedEventLog {
491    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
492        formatter
493            .debug_struct("BufferedEventLog")
494            .field("worker_running", &self.worker.is_some())
495            .finish_non_exhaustive()
496    }
497}
498
499impl BufferedEventLog {
500    fn open(path: PathBuf) -> std::io::Result<Self> {
501        let (sender, receiver) = mpsc::channel();
502        let worker = std::thread::Builder::new()
503            .name("codeswarm-event-log".into())
504            .spawn(move || buffered_log_worker(path, receiver))?;
505        Ok(Self {
506            sender,
507            worker: Some(worker),
508        })
509    }
510
511    /// Serialize and queue a normalized event without opening a file or
512    /// waiting on a filesystem operation in the caller.
513    pub fn append(&self, event: &AgentEvent) -> std::io::Result<()> {
514        let encoded = serde_json::to_string(event)
515            .map_err(|error| std::io::Error::other(error.to_string()))?;
516        self.sender
517            .send(BufferedLogCommand::Append(format!("{encoded}\n")))
518            .map_err(|_| {
519                std::io::Error::new(
520                    std::io::ErrorKind::BrokenPipe,
521                    "event log background writer stopped",
522                )
523            })
524    }
525
526    /// Drain queued records and force them to stable storage.
527    pub fn flush(&self) -> std::io::Result<()> {
528        let (reply, result) = mpsc::channel();
529        self.sender
530            .send(BufferedLogCommand::Flush(reply))
531            .map_err(|_| {
532                std::io::Error::new(
533                    std::io::ErrorKind::BrokenPipe,
534                    "event log background writer stopped",
535                )
536            })?;
537        result.recv().map_err(|_| {
538            std::io::Error::new(
539                std::io::ErrorKind::BrokenPipe,
540                "event log background writer stopped",
541            )
542        })?
543    }
544}
545
546impl Drop for BufferedEventLog {
547    fn drop(&mut self) {
548        let (reply, result) = mpsc::channel();
549        if self
550            .sender
551            .send(BufferedLogCommand::Shutdown(reply))
552            .is_ok()
553        {
554            let _ = result.recv();
555        }
556        if let Some(worker) = self.worker.take() {
557            let _ = worker.join();
558        }
559    }
560}
561
562fn buffered_log_worker(
563    path: PathBuf,
564    receiver: Receiver<BufferedLogCommand>,
565) -> std::io::Result<()> {
566    let file = OpenOptions::new().create(true).append(true).open(path)?;
567    let mut writer = BufWriter::new(file);
568    while let Ok(command) = receiver.recv() {
569        match command {
570            BufferedLogCommand::Append(line) => writer.write_all(line.as_bytes())?,
571            BufferedLogCommand::Flush(reply) => {
572                let result = writer.flush().and_then(|()| writer.get_ref().sync_data());
573                let _ = reply.send(result);
574            }
575            BufferedLogCommand::Shutdown(reply) => {
576                let result = writer.flush().and_then(|()| writer.get_ref().sync_data());
577                let worker_result = match result {
578                    Ok(()) => {
579                        let _ = reply.send(Ok(()));
580                        Ok(())
581                    }
582                    Err(error) => {
583                        let kind = error.kind();
584                        let detail = error.to_string();
585                        let _ = reply.send(Err(std::io::Error::new(kind, detail.clone())));
586                        Err(std::io::Error::new(kind, detail))
587                    }
588                };
589                return worker_result;
590            }
591        }
592    }
593    writer.flush()
594}
595
596#[cfg(test)]
597mod tests {
598    use std::time::{SystemTime, UNIX_EPOCH};
599
600    use super::{AgentCapabilities, AgentEvent, Effect, Mode, SessionState, reduce};
601
602    #[test]
603    fn replacement_catalog_invalidates_stale_mode() {
604        let mut state = SessionState::new(1);
605        reduce(
606            &mut state,
607            AgentEvent::ModesReplaced {
608                slot: 0,
609                modes: vec![Mode {
610                    id: "read".into(),
611                    label: "Read only".into(),
612                }],
613                current_mode: Some("write".into()),
614            },
615        );
616        assert_eq!(state.slots[0].current_mode, None);
617    }
618
619    #[test]
620    fn crash_tombstones_slot_and_uses_crash_copy() {
621        let mut state = SessionState::new(2);
622        reduce(
623            &mut state,
624            AgentEvent::Ready {
625                slot: 1,
626                capabilities: AgentCapabilities::default(),
627            },
628        );
629        let effects = reduce(
630            &mut state,
631            AgentEvent::Failed {
632                slot: 1,
633                started: true,
634                detail: "process exited".into(),
635            },
636        );
637        assert!(!state.slots[1].active);
638        assert!(effects.contains(&Effect::OfferReload {
639            slot: 1,
640            crashed: true,
641        }));
642    }
643
644    #[test]
645    fn event_log_replays_into_the_same_state() {
646        let unique = SystemTime::now()
647            .duration_since(UNIX_EPOCH)
648            .expect("clock")
649            .as_nanos();
650        let path = std::env::temp_dir().join(format!("codeswarm-core-{unique}.jsonl"));
651        let log = super::EventLog::open(&path);
652        let events = [
653            AgentEvent::Text {
654                slot: 0,
655                text: "first".into(),
656            },
657            AgentEvent::Failed {
658                slot: 1,
659                started: true,
660                detail: "crashed".into(),
661            },
662        ];
663        for event in &events {
664            log.append(event).expect("append");
665        }
666        let replayed = log.replay(2).expect("replay");
667        let mut expected = SessionState::new(2);
668        for event in events {
669            reduce(&mut expected, event);
670        }
671        assert_eq!(replayed, expected);
672        std::fs::remove_file(path).expect("cleanup");
673    }
674
675    #[test]
676    fn event_log_can_checkpoint_batched_appends() {
677        let unique = SystemTime::now()
678            .duration_since(UNIX_EPOCH)
679            .expect("clock")
680            .as_nanos();
681        let path = std::env::temp_dir().join(format!("codeswarm-core-checkpoint-{unique}.jsonl"));
682        let log = super::EventLog::open(&path);
683        log.append(&AgentEvent::Text {
684            slot: 0,
685            text: "batched".into(),
686        })
687        .expect("append");
688        // `sync` is an explicit checkpoint; it is separate from the hot-path
689        // append so render/input latency cannot inherit a storage flush.
690        log.sync().expect("checkpoint");
691        assert_eq!(log.read().expect("read").len(), 1);
692        std::fs::remove_file(path).expect("cleanup");
693    }
694
695    #[test]
696    fn event_log_durable_append_is_replayable() {
697        let unique = SystemTime::now()
698            .duration_since(UNIX_EPOCH)
699            .expect("clock")
700            .as_nanos();
701        let path = std::env::temp_dir().join(format!("codeswarm-core-durable-{unique}.jsonl"));
702        let log = super::EventLog::open(&path);
703        log.append_durable(&AgentEvent::Text {
704            slot: 1,
705            text: "durable".into(),
706        })
707        .expect("durable append");
708        assert_eq!(
709            log.read().expect("read")[0].clone(),
710            AgentEvent::Text {
711                slot: 1,
712                text: "durable".into(),
713            }
714        );
715        std::fs::remove_file(path).expect("cleanup");
716    }
717
718    #[test]
719    fn buffered_event_log_drains_and_checkpoints_in_order() {
720        let unique = SystemTime::now()
721            .duration_since(UNIX_EPOCH)
722            .expect("clock")
723            .as_nanos();
724        let path = std::env::temp_dir().join(format!("codeswarm-core-buffered-{unique}.jsonl"));
725        let log = super::EventLog::open(&path);
726        let buffered = log.buffered().expect("background writer");
727        for text in ["one", "two", "three"] {
728            buffered
729                .append(&AgentEvent::Text {
730                    slot: 0,
731                    text: text.into(),
732                })
733                .expect("queue event");
734        }
735        buffered.flush().expect("checkpoint");
736        assert_eq!(
737            log.read().expect("read").into_iter().collect::<Vec<_>>(),
738            [
739                AgentEvent::Text {
740                    slot: 0,
741                    text: "one".into()
742                },
743                AgentEvent::Text {
744                    slot: 0,
745                    text: "two".into()
746                },
747                AgentEvent::Text {
748                    slot: 0,
749                    text: "three".into()
750                }
751            ]
752        );
753        drop(buffered);
754        std::fs::remove_file(path).expect("cleanup");
755    }
756
757    #[test]
758    fn activity_marks_turn_active_until_completion() {
759        let mut state = SessionState::new(1);
760        reduce(
761            &mut state,
762            AgentEvent::Text {
763                slot: 0,
764                text: "stream".into(),
765            },
766        );
767        assert_eq!(state.active_slot, Some(0));
768        reduce(&mut state, AgentEvent::TurnComplete { slot: 0 });
769        assert_eq!(state.active_slot, None);
770    }
771}
codeswarm_core/lib.rs

codeswarm_core/
lib.rs
