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

codeswarm_core/
lib.rs
