codeswarm_adapters/
lib.rs

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

codeswarm_adapters/
lib.rs
