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    ModesReplaced {
140        slot: RosterSlot,
141        modes: Vec<Mode>,
142        current_mode: Option<String>,
143    },
144    ModeUpdated {
145        slot: RosterSlot,
146        current_mode: String,
147    },
148    ModelsReplaced {
149        slot: RosterSlot,
150        config_id: String,
151        models: Vec<Mode>,
152        current_model: Option<String>,
153    },
154    ModelUpdated {
155        slot: RosterSlot,
156        current_model: String,
157    },
158    UserText {
159        slot: RosterSlot,
160        text: String,
161    },
162    CommandsReplaced {
163        slot: RosterSlot,
164        commands: Vec<AgentCommand>,
165    },
166    UsageUpdated {
167        slot: RosterSlot,
168        usage: UsageUpdate,
169    },
170    Text {
171        slot: RosterSlot,
172        text: String,
173    },
174    Thought {
175        slot: RosterSlot,
176        text: String,
177    },
178    Tool {
179        slot: RosterSlot,
180        update: ToolUpdate,
181    },
182    Permission {
183        slot: RosterSlot,
184        request: PermissionRequest,
185    },
186    Terminal {
187        slot: RosterSlot,
188        event: TerminalEvent,
189    },
190    TurnComplete {
191        slot: RosterSlot,
192    },
193    Failed {
194        slot: RosterSlot,
195        started: bool,
196        detail: String,
197    },
198}
199
200#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
201pub enum Effect {
202    Render,
203    DispatchPrompt { slot: RosterSlot, prompt: String },
204    OfferReload { slot: RosterSlot, crashed: bool },
205}
206
207#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
208pub struct AgentSlot {
209    pub active: bool,
210    pub capabilities: AgentCapabilities,
211    pub modes: Vec<Mode>,
212    pub current_mode: Option<String>,
213    #[serde(default)]
214    pub models: Vec<Mode>,
215    #[serde(default)]
216    pub current_model: Option<String>,
217    #[serde(default)]
218    pub commands: Vec<AgentCommand>,
219    #[serde(default)]
220    pub usage: Option<UsageUpdate>,
221}
222
223impl Default for AgentSlot {
224    fn default() -> Self {
225        Self {
226            active: true,
227            capabilities: AgentCapabilities::default(),
228            modes: Vec::new(),
229            current_mode: None,
230            models: Vec::new(),
231            current_model: None,
232            commands: Vec::new(),
233            usage: None,
234        }
235    }
236}
237
238#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
239pub struct SessionState {
240    pub slots: Vec<AgentSlot>,
241    pub active_slot: Option<RosterSlot>,
242    pub queued_prompts: VecDeque<(RosterSlot, String)>,
243    pub public_text: Vec<(RosterSlot, String)>,
244}
245
246impl SessionState {
247    pub fn new(roster_size: usize) -> Self {
248        Self {
249            slots: (0..roster_size).map(|_| AgentSlot::default()).collect(),
250            active_slot: None,
251            queued_prompts: VecDeque::new(),
252            public_text: Vec::new(),
253        }
254    }
255}
256
257/// Apply one normalized event. I/O and rendering are represented by effects,
258/// never performed in the reducer.
259pub fn reduce(state: &mut SessionState, event: AgentEvent) -> Vec<Effect> {
260    match event {
261        AgentEvent::RosterUpdated { .. } => vec![Effect::Render],
262        AgentEvent::Ready { slot, capabilities } => {
263            if let Some(agent) = state.slots.get_mut(slot) {
264                agent.capabilities = capabilities;
265            }
266            vec![Effect::Render]
267        }
268        AgentEvent::ModesReplaced {
269            slot,
270            modes,
271            current_mode,
272        } => {
273            if let Some(agent) = state.slots.get_mut(slot) {
274                agent.modes = modes;
275                agent.current_mode =
276                    current_mode.filter(|id| agent.modes.iter().any(|mode| mode.id == *id));
277            }
278            vec![Effect::Render]
279        }
280        AgentEvent::CommandsReplaced { slot, commands } => {
281            if let Some(agent) = state.slots.get_mut(slot) {
282                agent.commands = commands;
283            }
284            vec![Effect::Render]
285        }
286        AgentEvent::ModeUpdated { slot, current_mode } => {
287            if let Some(agent) = state.slots.get_mut(slot) {
288                agent.current_mode = Some(current_mode);
289            }
290            vec![Effect::Render]
291        }
292        AgentEvent::ModelsReplaced {
293            slot,
294            models,
295            current_model,
296            ..
297        } => {
298            if let Some(agent) = state.slots.get_mut(slot) {
299                agent.models = models;
300                agent.current_model =
301                    current_model.filter(|id| agent.models.iter().any(|model| model.id == *id));
302            }
303            vec![Effect::Render]
304        }
305        AgentEvent::ModelUpdated {
306            slot,
307            current_model,
308        } => {
309            if let Some(agent) = state.slots.get_mut(slot) {
310                agent.current_model = Some(current_model);
311            }
312            vec![Effect::Render]
313        }
314        AgentEvent::UsageUpdated { slot, usage } => {
315            if let Some(agent) = state.slots.get_mut(slot) {
316                agent.usage = Some(usage);
317            }
318            vec![Effect::Render]
319        }
320        AgentEvent::Text { slot, text } => {
321            state.active_slot = Some(slot);
322            state.public_text.push((slot, text));
323            vec![Effect::Render]
324        }
325        AgentEvent::UserText { slot, .. } => {
326            state.active_slot = Some(slot);
327            vec![Effect::Render]
328        }
329        AgentEvent::Thought { slot, .. }
330        | AgentEvent::Tool { slot, .. }
331        | AgentEvent::Permission { slot, .. }
332        | AgentEvent::Terminal { slot, .. } => {
333            state.active_slot = Some(slot);
334            vec![Effect::Render]
335        }
336        AgentEvent::TurnComplete { .. } => {
337            state.active_slot = None;
338            let next =
339                state
340                    .queued_prompts
341                    .pop_front()
342                    .map(|(target, prompt)| Effect::DispatchPrompt {
343                        slot: target,
344                        prompt,
345                    });
346            let mut effects = vec![Effect::Render];
347            if let Some(effect) = next {
348                effects.push(effect);
349            }
350            effects
351        }
352        AgentEvent::Failed {
353            slot,
354            started,
355            detail: _,
356        } => {
357            if let Some(agent) = state.slots.get_mut(slot) {
358                agent.active = false;
359            }
360            if state.active_slot == Some(slot) {
361                state.active_slot = None;
362            }
363            vec![
364                Effect::Render,
365                Effect::OfferReload {
366                    slot,
367                    crashed: started,
368                },
369            ]
370        }
371    }
372}
373
374/// A newline-delimited event log. It deliberately records normalized events
375/// rather than UI operations, so sessions can be replayed by a future renderer
376/// or adapter host. Each append closes its file handle but does not force a
377/// storage sync; the terminal input/render loop must never block on fsync.
378#[derive(Clone, Debug)]
379pub struct EventLog {
380    path: PathBuf,
381}
382
383impl EventLog {
384    pub fn open(path: impl Into<PathBuf>) -> Self {
385        Self { path: path.into() }
386    }
387
388    pub fn path(&self) -> &Path {
389        &self.path
390    }
391
392    pub fn append(&self, event: &AgentEvent) -> std::io::Result<()> {
393        let encoded = serde_json::to_string(event)
394            .map_err(|error| std::io::Error::other(error.to_string()))?;
395        let mut file = OpenOptions::new()
396            .create(true)
397            .append(true)
398            .open(&self.path)?;
399        file.write_all(encoded.as_bytes())?;
400        file.write_all(b"\n")
401    }
402
403    /// Append an event and force it to stable storage.
404    ///
405    /// The regular [`append`](Self::append) path is deliberately lightweight
406    /// because it is called from the terminal event loop. Call this only at a
407    /// durability boundary such as a completed turn or explicit shutdown.
408    pub fn append_durable(&self, event: &AgentEvent) -> std::io::Result<()> {
409        let encoded = serde_json::to_string(event)
410            .map_err(|error| std::io::Error::other(error.to_string()))?;
411        let mut file = OpenOptions::new()
412            .create(true)
413            .append(true)
414            .open(&self.path)?;
415        file.write_all(encoded.as_bytes())?;
416        file.write_all(b"\n")?;
417        file.sync_data()
418    }
419
420    /// Force already-appended records to stable storage without writing a
421    /// duplicate event. This is useful for callers that batch lightweight
422    /// appends and checkpoint at turn boundaries.
423    pub fn sync(&self) -> std::io::Result<()> {
424        let file = OpenOptions::new().read(true).open(&self.path)?;
425        file.sync_data()
426    }
427
428    /// Start a background writer for event-loop use. The returned handle
429    /// queues records in memory and performs all file I/O on its worker
430    /// thread. Use [`BufferedEventLog::flush`] at explicit durability
431    /// boundaries such as completed turns.
432    pub fn buffered(&self) -> std::io::Result<BufferedEventLog> {
433        BufferedEventLog::open(self.path.clone())
434    }
435
436    pub fn read(&self) -> std::io::Result<Vec<AgentEvent>> {
437        let file = match File::open(&self.path) {
438            Ok(file) => file,
439            Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
440            Err(error) => return Err(error),
441        };
442        BufReader::new(file)
443            .lines()
444            .enumerate()
445            .filter_map(|(line_number, result)| match result {
446                Ok(line) if line.trim().is_empty() => None,
447                Ok(line) => Some(serde_json::from_str(&line).map_err(|error| {
448                    std::io::Error::new(
449                        std::io::ErrorKind::InvalidData,
450                        format!("event log line {}: {error}", line_number + 1),
451                    )
452                })),
453                Err(error) => Some(Err(error)),
454            })
455            .collect()
456    }
457
458    pub fn replay(&self, roster_size: usize) -> std::io::Result<SessionState> {
459        let mut state = SessionState::new(roster_size);
460        for event in self.read()? {
461            reduce(&mut state, event);
462        }
463        Ok(state)
464    }
465}
466
467enum BufferedLogCommand {
468    Append(String),
469    Flush(Sender<std::io::Result<()>>),
470    Shutdown(Sender<std::io::Result<()>>),
471}
472
473/// Background event-log writer used to keep terminal input/render handling
474/// independent from filesystem latency. The channel is intentionally
475/// unbounded: dropping normalized events during a streamed turn would make
476/// replay and recovery incomplete, while the writer drains ordinary event
477/// rates faster than adapters produce them.
478pub struct BufferedEventLog {
479    sender: Sender<BufferedLogCommand>,
480    worker: Option<std::thread::JoinHandle<std::io::Result<()>>>,
481}
482
483impl std::fmt::Debug for BufferedEventLog {
484    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
485        formatter
486            .debug_struct("BufferedEventLog")
487            .field("worker_running", &self.worker.is_some())
488            .finish_non_exhaustive()
489    }
490}
491
492impl BufferedEventLog {
493    fn open(path: PathBuf) -> std::io::Result<Self> {
494        let (sender, receiver) = mpsc::channel();
495        let worker = std::thread::Builder::new()
496            .name("codeswarm-event-log".into())
497            .spawn(move || buffered_log_worker(path, receiver))?;
498        Ok(Self {
499            sender,
500            worker: Some(worker),
501        })
502    }
503
504    /// Serialize and queue a normalized event without opening a file or
505    /// waiting on a filesystem operation in the caller.
506    pub fn append(&self, event: &AgentEvent) -> std::io::Result<()> {
507        let encoded = serde_json::to_string(event)
508            .map_err(|error| std::io::Error::other(error.to_string()))?;
509        self.sender
510            .send(BufferedLogCommand::Append(format!("{encoded}\n")))
511            .map_err(|_| {
512                std::io::Error::new(
513                    std::io::ErrorKind::BrokenPipe,
514                    "event log background writer stopped",
515                )
516            })
517    }
518
519    /// Drain queued records and force them to stable storage.
520    pub fn flush(&self) -> std::io::Result<()> {
521        let (reply, result) = mpsc::channel();
522        self.sender
523            .send(BufferedLogCommand::Flush(reply))
524            .map_err(|_| {
525                std::io::Error::new(
526                    std::io::ErrorKind::BrokenPipe,
527                    "event log background writer stopped",
528                )
529            })?;
530        result.recv().map_err(|_| {
531            std::io::Error::new(
532                std::io::ErrorKind::BrokenPipe,
533                "event log background writer stopped",
534            )
535        })?
536    }
537}
538
539impl Drop for BufferedEventLog {
540    fn drop(&mut self) {
541        let (reply, result) = mpsc::channel();
542        if self
543            .sender
544            .send(BufferedLogCommand::Shutdown(reply))
545            .is_ok()
546        {
547            let _ = result.recv();
548        }
549        if let Some(worker) = self.worker.take() {
550            let _ = worker.join();
551        }
552    }
553}
554
555fn buffered_log_worker(
556    path: PathBuf,
557    receiver: Receiver<BufferedLogCommand>,
558) -> std::io::Result<()> {
559    let file = OpenOptions::new().create(true).append(true).open(path)?;
560    let mut writer = BufWriter::new(file);
561    while let Ok(command) = receiver.recv() {
562        match command {
563            BufferedLogCommand::Append(line) => writer.write_all(line.as_bytes())?,
564            BufferedLogCommand::Flush(reply) => {
565                let result = writer.flush().and_then(|()| writer.get_ref().sync_data());
566                let _ = reply.send(result);
567            }
568            BufferedLogCommand::Shutdown(reply) => {
569                let result = writer.flush().and_then(|()| writer.get_ref().sync_data());
570                let worker_result = match result {
571                    Ok(()) => {
572                        let _ = reply.send(Ok(()));
573                        Ok(())
574                    }
575                    Err(error) => {
576                        let kind = error.kind();
577                        let detail = error.to_string();
578                        let _ = reply.send(Err(std::io::Error::new(kind, detail.clone())));
579                        Err(std::io::Error::new(kind, detail))
580                    }
581                };
582                return worker_result;
583            }
584        }
585    }
586    writer.flush()
587}
588
589#[cfg(test)]
590mod tests {
591    use std::time::{SystemTime, UNIX_EPOCH};
592
593    use super::{AgentCapabilities, AgentEvent, Effect, Mode, SessionState, reduce};
594
595    #[test]
596    fn replacement_catalog_invalidates_stale_mode() {
597        let mut state = SessionState::new(1);
598        reduce(
599            &mut state,
600            AgentEvent::ModesReplaced {
601                slot: 0,
602                modes: vec![Mode {
603                    id: "read".into(),
604                    label: "Read only".into(),
605                }],
606                current_mode: Some("write".into()),
607            },
608        );
609        assert_eq!(state.slots[0].current_mode, None);
610    }
611
612    #[test]
613    fn crash_tombstones_slot_and_uses_crash_copy() {
614        let mut state = SessionState::new(2);
615        reduce(
616            &mut state,
617            AgentEvent::Ready {
618                slot: 1,
619                capabilities: AgentCapabilities::default(),
620            },
621        );
622        let effects = reduce(
623            &mut state,
624            AgentEvent::Failed {
625                slot: 1,
626                started: true,
627                detail: "process exited".into(),
628            },
629        );
630        assert!(!state.slots[1].active);
631        assert!(effects.contains(&Effect::OfferReload {
632            slot: 1,
633            crashed: true,
634        }));
635    }
636
637    #[test]
638    fn event_log_replays_into_the_same_state() {
639        let unique = SystemTime::now()
640            .duration_since(UNIX_EPOCH)
641            .expect("clock")
642            .as_nanos();
643        let path = std::env::temp_dir().join(format!("codeswarm-core-{unique}.jsonl"));
644        let log = super::EventLog::open(&path);
645        let events = [
646            AgentEvent::Text {
647                slot: 0,
648                text: "first".into(),
649            },
650            AgentEvent::Failed {
651                slot: 1,
652                started: true,
653                detail: "crashed".into(),
654            },
655        ];
656        for event in &events {
657            log.append(event).expect("append");
658        }
659        let replayed = log.replay(2).expect("replay");
660        let mut expected = SessionState::new(2);
661        for event in events {
662            reduce(&mut expected, event);
663        }
664        assert_eq!(replayed, expected);
665        std::fs::remove_file(path).expect("cleanup");
666    }
667
668    #[test]
669    fn event_log_can_checkpoint_batched_appends() {
670        let unique = SystemTime::now()
671            .duration_since(UNIX_EPOCH)
672            .expect("clock")
673            .as_nanos();
674        let path = std::env::temp_dir().join(format!("codeswarm-core-checkpoint-{unique}.jsonl"));
675        let log = super::EventLog::open(&path);
676        log.append(&AgentEvent::Text {
677            slot: 0,
678            text: "batched".into(),
679        })
680        .expect("append");
681        // `sync` is an explicit checkpoint; it is separate from the hot-path
682        // append so render/input latency cannot inherit a storage flush.
683        log.sync().expect("checkpoint");
684        assert_eq!(log.read().expect("read").len(), 1);
685        std::fs::remove_file(path).expect("cleanup");
686    }
687
688    #[test]
689    fn event_log_durable_append_is_replayable() {
690        let unique = SystemTime::now()
691            .duration_since(UNIX_EPOCH)
692            .expect("clock")
693            .as_nanos();
694        let path = std::env::temp_dir().join(format!("codeswarm-core-durable-{unique}.jsonl"));
695        let log = super::EventLog::open(&path);
696        log.append_durable(&AgentEvent::Text {
697            slot: 1,
698            text: "durable".into(),
699        })
700        .expect("durable append");
701        assert_eq!(
702            log.read().expect("read")[0].clone(),
703            AgentEvent::Text {
704                slot: 1,
705                text: "durable".into(),
706            }
707        );
708        std::fs::remove_file(path).expect("cleanup");
709    }
710
711    #[test]
712    fn buffered_event_log_drains_and_checkpoints_in_order() {
713        let unique = SystemTime::now()
714            .duration_since(UNIX_EPOCH)
715            .expect("clock")
716            .as_nanos();
717        let path = std::env::temp_dir().join(format!("codeswarm-core-buffered-{unique}.jsonl"));
718        let log = super::EventLog::open(&path);
719        let buffered = log.buffered().expect("background writer");
720        for text in ["one", "two", "three"] {
721            buffered
722                .append(&AgentEvent::Text {
723                    slot: 0,
724                    text: text.into(),
725                })
726                .expect("queue event");
727        }
728        buffered.flush().expect("checkpoint");
729        assert_eq!(
730            log.read().expect("read").into_iter().collect::<Vec<_>>(),
731            [
732                AgentEvent::Text {
733                    slot: 0,
734                    text: "one".into()
735                },
736                AgentEvent::Text {
737                    slot: 0,
738                    text: "two".into()
739                },
740                AgentEvent::Text {
741                    slot: 0,
742                    text: "three".into()
743                }
744            ]
745        );
746        drop(buffered);
747        std::fs::remove_file(path).expect("cleanup");
748    }
749
750    #[test]
751    fn activity_marks_turn_active_until_completion() {
752        let mut state = SessionState::new(1);
753        reduce(
754            &mut state,
755            AgentEvent::Text {
756                slot: 0,
757                text: "stream".into(),
758            },
759        );
760        assert_eq!(state.active_slot, Some(0));
761        reduce(&mut state, AgentEvent::TurnComplete { slot: 0 });
762        assert_eq!(state.active_slot, None);
763    }
764}
codeswarm_core/lib.rs

codeswarm_core/
lib.rs
