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

codeswarm_adapters/
lib.rs
