scv_server/
lib.rs

1//! Peon's authoritative stdio server.
2
3mod config;
4
5use std::{
6    collections::HashMap,
7    io::Read as _,
8    path::{Path, PathBuf},
9    sync::{
10        Arc,
11        atomic::{AtomicU64, AtomicUsize, Ordering},
12    },
13    time::Duration,
14};
15
16use anyhow::{Context, Result, anyhow};
17use async_trait::async_trait;
18use config::Config;
19pub use config::{ApprovalPolicy, ConfigOverrides};
20pub fn init_user_config() -> anyhow::Result<std::path::PathBuf> { config::Config::init_user_config() }
21use scv_core::{
22    AgentError, AgentRuntime, ApprovalGate, ApprovalRequest, BudgetContextPolicy, CoreEvent,
23    EventSink, Message, ToolRisk,
24};
25use scv_protocol::{ClientMessage, PROTOCOL_VERSION, PeerInfo, ServerEvent, Usage};
26use scv_provider_openai::OpenAiProvider;
27use scv_tools::{SkillMap, builtin_registry};
28use tokio::{
29    io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
30    sync::{Mutex, OwnedSemaphorePermit, Semaphore, mpsc, oneshot},
31    task::JoinHandle,
32};
33use tokio_util::sync::CancellationToken;
34use uuid::Uuid;
35
36const PROMPT_LIMIT_BYTES: usize = 256 * 1024;
37const OUTPUT_QUEUE_CAPACITY: usize = 256;
38const OUTPUT_QUEUE_MIN_BYTES: usize = 16 * 1024 * 1024;
39const SHUTDOWN_GRACE: Duration = Duration::from_secs(3);
40
41pub async fn run_stdio(overrides: ConfigOverrides) -> Result<()> {
42    let stdin = tokio::io::stdin();
43    let stdout = tokio::io::stdout();
44    run(stdin, stdout, overrides).await
45}
46
47async fn run<R, W>(reader: R, writer: W, overrides: ConfigOverrides) -> Result<()>
48where
49    R: tokio::io::AsyncRead + Unpin,
50    W: tokio::io::AsyncWrite + Unpin + Send + 'static,
51{
52    let initial_output_bytes =
53        output_queue_bytes(Config::default().protocol.max_server_frame_bytes)?;
54    let (output_tx, mut output_rx) = outbound_channel(initial_output_bytes);
55    let mut writer_task = tokio::spawn(async move {
56        let mut writer = writer;
57        while let Some(frame) = output_rx.recv().await {
58            writer.write_all(&frame.bytes).await?;
59            writer.write_all(b"\n").await?;
60            writer.flush().await?;
61        }
62        Ok::<(), std::io::Error>(())
63    });
64    let (done_tx, mut done_rx) = mpsc::channel::<TurnDone>(4);
65    let approvals = Arc::new(ApprovalBroker::default());
66    let mut reader = BufReader::new(reader);
67    let mut initialized = false;
68    let mut session: Option<Session> = None;
69    let mut active: Option<ActiveTurn> = None;
70    let mut fatal = false;
71    let mut writer_finished = false;
72
73    let loop_result: Result<()> = async {
74        loop {
75        let frame_limit = session.as_ref().map_or_else(
76            || Config::default().protocol.max_client_frame_bytes,
77            |value| value.config.protocol.max_client_frame_bytes,
78        );
79        tokio::select! {
80            read = read_bounded_frame(&mut reader, frame_limit) => {
81                let frame = match read.context("read protocol input")? {
82                    FrameRead::Eof => {
83                        if let Some(active) = &active { active.cancellation.cancel(); }
84                        break;
85                    }
86                    FrameRead::TooLarge => {
87                        send_error(&output_tx, "", "invalid_request", "client frame exceeds configured limit", false, server_frame_limit(&session)).await?;
88                        continue;
89                    }
90                    FrameRead::Frame(frame) => frame,
91                };
92                if frame.is_empty() {
93                    send_error(&output_tx, "", "invalid_json", "protocol frame is empty", false, server_frame_limit(&session)).await?;
94                    continue;
95                }
96                let message = match serde_json::from_slice::<ClientMessage>(&frame) {
97                    Ok(message) => message,
98                    Err(error) => {
99                        send_error(&output_tx, "", "invalid_json", &format!("invalid protocol JSON: {error}"), false, server_frame_limit(&session)).await?;
100                        continue;
101                    }
102                };
103                match message {
104                    ClientMessage::Initialize { request_id, protocol_version, .. } => {
105                        if initialized {
106                            send_error(&output_tx, &request_id, "invalid_request", "connection is already initialized", false, server_frame_limit(&session)).await?;
107                            continue;
108                        }
109                        if protocol_version != PROTOCOL_VERSION {
110                            send_error(&output_tx, &request_id, "version_mismatch", &format!("server supports protocol {PROTOCOL_VERSION}"), true, server_frame_limit(&session)).await?;
111                            fatal = true;
112                            break;
113                        }
114                        initialized = true;
115                        send_event(&output_tx, ServerEvent::Initialized {
116                            request_id,
117                            protocol_version: PROTOCOL_VERSION,
118                            server: PeerInfo { name: "scv-server".into(), version: env!("CARGO_PKG_VERSION").into() },
119                        }, Config::default().protocol.max_server_frame_bytes).await?;
120                    }
121                    other if !initialized => {
122                        send_error(&output_tx, other.request_id(), "not_initialized", "initialize must be the first message", false, server_frame_limit(&session)).await?;
123                    }
124                    ClientMessage::SessionStart { request_id, cwd } => {
125                        if session.is_some() {
126                            send_error(&output_tx, &request_id, "invalid_request", "this connection already has a session", false, server_frame_limit(&session)).await?;
127                            continue;
128                        }
129                        match build_session(&cwd, overrides.clone()).await {
130                            Ok(new_session) => {
131                                output_tx.ensure_capacity(output_queue_bytes(
132                                    new_session.config.protocol.max_server_frame_bytes,
133                                )?)?;
134                                let event = ServerEvent::SessionStarted {
135                                    request_id,
136                                    session_id: new_session.id.clone(),
137                                    cwd: new_session.workspace.display().to_string(),
138                                    model: new_session.runtime.model().to_owned(),
139                                    context_max_tokens: new_session.config.context.max_tokens,
140                                    max_server_frame_bytes: new_session.config.protocol.max_server_frame_bytes,
141                                    max_transcript_bytes: new_session.config.tui.max_transcript_bytes,
142                                    max_transcript_items: new_session.config.tui.max_transcript_items,
143                                    max_prompt_history_bytes: new_session.config.tui.max_prompt_history_bytes,
144                                    max_prompt_history_items: new_session.config.tui.max_prompt_history_items,
145                                };
146                                send_event(&output_tx, event, new_session.config.protocol.max_server_frame_bytes).await?;
147                                session = Some(new_session);
148                            }
149                            Err(error) => {
150                                send_error(&output_tx, &request_id, "invalid_request", &error.to_string(), false, server_frame_limit(&session)).await?;
151                            }
152                        }
153                    }
154                    ClientMessage::TurnStart { request_id, session_id, prompt } => {
155                        let Some(current) = session.as_ref() else {
156                            send_error(&output_tx, &request_id, "session_not_found", "start a session first", false, server_frame_limit(&session)).await?;
157                            continue;
158                        };
159                        if current.id != session_id {
160                            send_error(&output_tx, &request_id, "session_not_found", "session id does not match", false, server_frame_limit(&session)).await?;
161                            continue;
162                        }
163                        if active.is_some() {
164                            send_error(&output_tx, &request_id, "turn_active", "a turn is already active", false, server_frame_limit(&session)).await?;
165                            continue;
166                        }
167                        if prompt.trim().is_empty() || prompt.len() > PROMPT_LIMIT_BYTES {
168                            send_error(&output_tx, &request_id, "invalid_request", "prompt must be non-empty and no larger than 256 KiB", false, server_frame_limit(&session)).await?;
169                            continue;
170                        }
171                        let turn_id = Uuid::new_v4().to_string();
172                        let cancellation = CancellationToken::new();
173                        let meta = TurnMeta {
174                            request_id: request_id.clone(),
175                            session_id: current.id.clone(),
176                            turn_id: turn_id.clone(),
177                            seq: Arc::clone(&current.seq),
178                            max_server_frame: current.config.protocol.max_server_frame_bytes,
179                        };
180                        send_event(&output_tx, ServerEvent::TurnStarted {
181                            request_id: request_id.clone(),
182                            session_id: current.id.clone(),
183                            turn_id: turn_id.clone(),
184                            seq: next_seq(&current.seq),
185                        }, current.config.protocol.max_server_frame_bytes).await?;
186                        let runtime = Arc::clone(&current.runtime);
187                        let history = Arc::clone(&current.history);
188                        let sink: Arc<dyn EventSink> = Arc::new(ProtocolSink {
189                            meta: meta.clone(),
190                            output: output_tx.clone(),
191                            cancellation: cancellation.clone(),
192                        });
193                        let gate: Arc<dyn ApprovalGate> = Arc::new(ProtocolApprovalGate {
194                            policy: current.config.tools.approval_policy,
195                            broker: Arc::clone(&approvals),
196                            meta,
197                            output: output_tx.clone(),
198                        });
199                        let task_cancel = cancellation.clone();
200                        let task_done = done_tx.clone();
201                        let task_request = request_id.clone();
202                        let task_session = current.id.clone();
203                        let task_turn = turn_id.clone();
204                        let task = tokio::spawn(async move {
205                            let mut history = history.lock().await;
206                            let result = runtime.run_turn(&mut history, prompt, sink, gate, task_cancel).await;
207                            let _ = task_done.send(TurnDone {
208                                request_id: task_request,
209                                session_id: task_session,
210                                turn_id: task_turn,
211                                result,
212                            }).await;
213                        });
214                        active = Some(ActiveTurn { turn_id, cancellation, task });
215                    }
216                    ClientMessage::TurnCancel { request_id, session_id, turn_id } => {
217                        match (&session, &active) {
218                            (Some(current), Some(running)) if current.id == session_id && running.turn_id == turn_id => running.cancellation.cancel(),
219                            _ => send_error(&output_tx, &request_id, "turn_not_found", "active turn was not found", false, server_frame_limit(&session)).await?,
220                        }
221                    }
222                    ClientMessage::ApprovalResolve { request_id, session_id, approval_id, approved } => {
223                        if session.as_ref().is_none_or(|current| current.id != session_id) {
224                            send_error(&output_tx, &request_id, "session_not_found", "session id does not match", false, server_frame_limit(&session)).await?;
225                        } else if !approvals.resolve(&approval_id, approved).await {
226                            send_error(&output_tx, &request_id, "approval_not_found", "approval was not found or already resolved", false, server_frame_limit(&session)).await?;
227                        }
228                    }
229                    ClientMessage::SessionClear { request_id, session_id } => {
230                        let Some(current) = session.as_ref() else {
231                            send_error(&output_tx, &request_id, "session_not_found", "session was not found", false, server_frame_limit(&session)).await?;
232                            continue;
233                        };
234                        if current.id != session_id {
235                            send_error(&output_tx, &request_id, "session_not_found", "session id does not match", false, server_frame_limit(&session)).await?;
236                        } else if active.is_some() {
237                            send_error(&output_tx, &request_id, "turn_active", "cancel the active turn before clearing", false, server_frame_limit(&session)).await?;
238                        } else {
239                            current.history.lock().await.clear();
240                            send_event(&output_tx, ServerEvent::SessionCleared {
241                                request_id,
242                                session_id: current.id.clone(),
243                                seq: next_seq(&current.seq),
244                            }, current.config.protocol.max_server_frame_bytes).await?;
245                        }
246                    }
247                }
248            }
249            writer = &mut writer_task => {
250                writer_finished = true;
251                writer.context("join protocol writer")??;
252                break;
253            }
254            done = done_rx.recv(), if active.is_some() => {
255                if let Some(done) = done {
256                    if let Some(current) = session.as_ref() {
257                        let seq = next_seq(&current.seq);
258                        let event = match done.result {
259                            Ok(outcome) => ServerEvent::TurnCompleted {
260                                request_id: done.request_id,
261                                session_id: done.session_id,
262                                turn_id: done.turn_id,
263                                seq,
264                                steps: outcome.steps,
265                                usage: Usage { input_tokens: outcome.usage.input_tokens, output_tokens: outcome.usage.output_tokens },
266                            },
267                            Err(AgentError::Cancelled) => ServerEvent::TurnCancelled {
268                                request_id: done.request_id,
269                                session_id: done.session_id,
270                                turn_id: done.turn_id,
271                                seq,
272                            },
273                            Err(error) => ServerEvent::TurnFailed {
274                                request_id: done.request_id,
275                                session_id: done.session_id,
276                                turn_id: done.turn_id,
277                                seq,
278                                code: error.code().into(),
279                                message: error.to_string(),
280                            },
281                        };
282                        send_event(&output_tx, event, current.config.protocol.max_server_frame_bytes).await?;
283                    }
284                    if let Some(active) = active.take() {
285                        let _ = active.task.await;
286                    }
287                }
288            }
289        }
290        }
291        Ok(())
292    }
293    .await;
294
295    if let Some(active) = active.take() {
296        shutdown_active_turn(active, SHUTDOWN_GRACE).await;
297    }
298    drop(output_tx);
299    let writer_result = if writer_finished {
300        Ok(())
301    } else {
302        shutdown_writer(writer_task, SHUTDOWN_GRACE).await
303    };
304    loop_result?;
305    writer_result?;
306    if fatal {
307        return Err(anyhow!("protocol version mismatch"));
308    }
309    Ok(())
310}
311
312enum FrameRead {
313    Eof,
314    Frame(Vec<u8>),
315    TooLarge,
316}
317
318struct OutboundFrame {
319    bytes: Vec<u8>,
320    _byte_permit: OwnedSemaphorePermit,
321}
322
323#[derive(Clone)]
324struct OutboundSender {
325    frames: mpsc::Sender<OutboundFrame>,
326    budget: Arc<Semaphore>,
327    capacity: Arc<AtomicUsize>,
328}
329
330#[derive(Debug, PartialEq, Eq)]
331enum OutboundSendError {
332    Cancelled,
333    Closed,
334    TimedOut,
335    FrameExceedsQueue { frame_bytes: usize, capacity: usize },
336}
337
338impl std::fmt::Display for OutboundSendError {
339    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
340        match self {
341            Self::Cancelled => formatter.write_str("outbound send cancelled"),
342            Self::Closed => formatter.write_str("protocol client disconnected"),
343            Self::TimedOut => formatter.write_str("outbound send timed out under backpressure"),
344            Self::FrameExceedsQueue {
345                frame_bytes,
346                capacity,
347            } => write!(
348                formatter,
349                "outbound frame uses {frame_bytes} bytes but queue capacity is {capacity} bytes"
350            ),
351        }
352    }
353}
354
355impl std::error::Error for OutboundSendError {}
356
357fn outbound_channel(capacity: usize) -> (OutboundSender, mpsc::Receiver<OutboundFrame>) {
358    let (frames, receiver) = mpsc::channel(OUTPUT_QUEUE_CAPACITY);
359    (
360        OutboundSender {
361            frames,
362            budget: Arc::new(Semaphore::new(capacity)),
363            capacity: Arc::new(AtomicUsize::new(capacity)),
364        },
365        receiver,
366    )
367}
368
369impl OutboundSender {
370    fn ensure_capacity(&self, required: usize) -> Result<()> {
371        if required > Semaphore::MAX_PERMITS {
372            return Err(anyhow!(
373                "outbound queue capacity {required} exceeds runtime limit {}",
374                Semaphore::MAX_PERMITS
375            ));
376        }
377        let current = self.capacity.load(Ordering::Acquire);
378        if required > current {
379            self.budget.add_permits(required - current);
380            self.capacity.store(required, Ordering::Release);
381        }
382        Ok(())
383    }
384
385    async fn send(
386        &self,
387        bytes: Vec<u8>,
388        cancellation: Option<&CancellationToken>,
389    ) -> std::result::Result<(), OutboundSendError> {
390        self.send_with_timeout(bytes, cancellation, SHUTDOWN_GRACE)
391            .await
392    }
393
394    async fn send_with_timeout(
395        &self,
396        bytes: Vec<u8>,
397        cancellation: Option<&CancellationToken>,
398        control_timeout: Duration,
399    ) -> std::result::Result<(), OutboundSendError> {
400        let frame_bytes =
401            bytes
402                .len()
403                .checked_add(1)
404                .ok_or(OutboundSendError::FrameExceedsQueue {
405                    frame_bytes: usize::MAX,
406                    capacity: self.capacity.load(Ordering::Acquire),
407                })?;
408        let capacity = self.capacity.load(Ordering::Acquire);
409        let permits =
410            u32::try_from(frame_bytes).map_err(|_| OutboundSendError::FrameExceedsQueue {
411                frame_bytes,
412                capacity,
413            })?;
414        if frame_bytes > capacity {
415            return Err(OutboundSendError::FrameExceedsQueue {
416                frame_bytes,
417                capacity,
418            });
419        }
420
421        let control_deadline = tokio::time::Instant::now() + control_timeout;
422        let acquire = Arc::clone(&self.budget).acquire_many_owned(permits);
423        let permit = if let Some(cancellation) = cancellation {
424            tokio::select! {
425                biased;
426                _ = cancellation.cancelled() => return Err(OutboundSendError::Cancelled),
427                permit = acquire => permit.map_err(|_| OutboundSendError::Closed)?,
428            }
429        } else {
430            tokio::time::timeout_at(control_deadline, acquire)
431                .await
432                .map_err(|_| OutboundSendError::TimedOut)?
433                .map_err(|_| OutboundSendError::Closed)?
434        };
435        let frame = OutboundFrame {
436            bytes,
437            _byte_permit: permit,
438        };
439        if let Some(cancellation) = cancellation {
440            tokio::select! {
441                biased;
442                _ = cancellation.cancelled() => Err(OutboundSendError::Cancelled),
443                result = self.frames.send(frame) => result.map_err(|_| OutboundSendError::Closed),
444            }
445        } else {
446            tokio::time::timeout_at(control_deadline, self.frames.send(frame))
447                .await
448                .map_err(|_| OutboundSendError::TimedOut)?
449                .map_err(|_| OutboundSendError::Closed)
450        }
451    }
452}
453
454fn output_queue_bytes(max_frame_bytes: usize) -> Result<usize> {
455    let required = max_frame_bytes
456        .checked_add(1)
457        .and_then(|bytes| bytes.checked_mul(2))
458        .ok_or_else(|| anyhow!("configured server frame limit is too large"))?
459        .max(OUTPUT_QUEUE_MIN_BYTES);
460    if required > Semaphore::MAX_PERMITS {
461        return Err(anyhow!(
462            "configured server frame limit requires an outbound queue larger than the runtime supports"
463        ));
464    }
465    Ok(required)
466}
467
468async fn read_bounded_frame<R>(reader: &mut R, max_bytes: usize) -> std::io::Result<FrameRead>
469where
470    R: AsyncBufRead + Unpin,
471{
472    let mut frame = Vec::with_capacity(max_bytes.min(8192));
473    let read_limit = u64::try_from(max_bytes)
474        .unwrap_or(u64::MAX)
475        .saturating_add(2);
476    let mut limited = reader.take(read_limit);
477    let read = limited.read_until(b'\n', &mut frame).await?;
478    drop(limited);
479    if read == 0 {
480        return Ok(FrameRead::Eof);
481    }
482    let ended_with_newline = frame.last() == Some(&b'\n');
483    while matches!(frame.last(), Some(b'\n' | b'\r')) {
484        frame.pop();
485    }
486    if frame.len() <= max_bytes {
487        return Ok(FrameRead::Frame(frame));
488    }
489    if !ended_with_newline {
490        loop {
491            let available = reader.fill_buf().await?;
492            if available.is_empty() {
493                break;
494            }
495            if let Some(end) = available.iter().position(|byte| *byte == b'\n') {
496                reader.consume(end + 1);
497                break;
498            }
499            let consumed = available.len();
500            reader.consume(consumed);
501        }
502    }
503    Ok(FrameRead::TooLarge)
504}
505
506fn server_frame_limit(session: &Option<Session>) -> usize {
507    session.as_ref().map_or_else(
508        || Config::default().protocol.max_server_frame_bytes,
509        |value| value.config.protocol.max_server_frame_bytes,
510    )
511}
512
513struct Session {
514    id: String,
515    workspace: PathBuf,
516    config: Config,
517    runtime: Arc<AgentRuntime>,
518    history: Arc<Mutex<Vec<Message>>>,
519    seq: Arc<AtomicU64>,
520}
521
522struct ActiveTurn {
523    turn_id: String,
524    cancellation: CancellationToken,
525    task: JoinHandle<()>,
526}
527
528async fn shutdown_active_turn(mut active: ActiveTurn, grace: Duration) -> bool {
529    active.cancellation.cancel();
530    if tokio::time::timeout(grace, &mut active.task).await.is_ok() {
531        true
532    } else {
533        active.task.abort();
534        let _ = active.task.await;
535        false
536    }
537}
538
539async fn shutdown_writer(
540    mut writer: JoinHandle<std::io::Result<()>>,
541    grace: Duration,
542) -> Result<()> {
543    match tokio::time::timeout(grace, &mut writer).await {
544        Ok(result) => {
545            result.context("join protocol writer")??;
546            Ok(())
547        }
548        Err(_) => {
549            writer.abort();
550            let _ = writer.await;
551            Err(anyhow!("protocol writer shutdown timed out"))
552        }
553    }
554}
555
556struct TurnDone {
557    request_id: String,
558    session_id: String,
559    turn_id: String,
560    result: Result<scv_core::TurnOutcome, AgentError>,
561}
562
563#[derive(Clone)]
564struct TurnMeta {
565    request_id: String,
566    session_id: String,
567    turn_id: String,
568    seq: Arc<AtomicU64>,
569    max_server_frame: usize,
570}
571
572fn next_seq(sequence: &AtomicU64) -> u64 {
573    sequence.fetch_add(1, Ordering::Relaxed) + 1
574}
575
576async fn build_session(cwd: &str, overrides: ConfigOverrides) -> Result<Session> {
577    let workspace = std::fs::canonicalize(cwd).with_context(|| format!("resolve cwd {cwd}"))?;
578    if !workspace.is_dir() {
579        return Err(anyhow!("cwd is not a directory"));
580    }
581    let config = Config::load(&workspace, overrides)?;
582    let provider_config = config.provider.clone();
583    let api_key = provider_config.api_key.clone().or_else(|| {
584        provider_config.api_key_env.as_deref().and_then(|name| std::env::var(name).ok())
585    }).filter(|key| !key.trim().is_empty()).ok_or_else(|| anyhow!("provider credential is not configured; set provider.api_key or provider.api_key_env"))?;
586    let (skills, skill_roots, skill_prompt) = discover_skills(&workspace, &config)?;
587    let system_prompt = build_system_prompt(&workspace, &config, &skill_prompt)?;
588    let provider = Arc::new(OpenAiProvider::new(
589        provider_config.model.clone(),
590        provider_config.base_url.clone(),
591        api_key,
592        Duration::from_secs(provider_config.timeout_seconds),
593        config.provider_limits(),
594        provider_config.headers.clone(),
595    )?);
596    let tools = Arc::new(builtin_registry(
597        config.tools(),
598        skills,
599        skill_roots,
600        config.skills.max_skill_bytes,
601        config.adapters(),
602    )?);
603    let context = Arc::new(BudgetContextPolicy::new((&config.context).into())?);
604    let runtime = Arc::new(AgentRuntime::new(
605        provider,
606        tools,
607        context,
608        config.core_agent(system_prompt),
609        workspace.clone(),
610    ));
611    Ok(Session {
612        id: Uuid::new_v4().to_string(),
613        workspace,
614        config,
615        runtime,
616        history: Arc::new(Mutex::new(Vec::new())),
617        seq: Arc::new(AtomicU64::new(0)),
618    })
619}
620
621fn build_system_prompt(workspace: &Path, config: &Config, skills: &str) -> Result<String> {
622    let mut prompt = config.agent.system_prompt.clone();
623    prompt.push_str(&format!(
624        "\nCurrent working directory: {}\n",
625        workspace.display()
626    ));
627    let agents_path = workspace.join("AGENTS.md");
628    if agents_path.is_file() {
629        let canonical = std::fs::canonicalize(&agents_path).context("resolve project AGENTS.md")?;
630        if !canonical.starts_with(workspace) {
631            return Err(anyhow!("project AGENTS.md escaped workspace"));
632        }
633        let (bytes, truncated) = read_prefix(&canonical, config.tools.max_read_bytes)
634            .context("read project AGENTS.md")?;
635        let instructions = std::str::from_utf8(&bytes).context("project AGENTS.md is not UTF-8")?;
636        prompt.push_str("\n# Project instructions\n");
637        prompt.push_str(instructions);
638        if truncated {
639            prompt.push_str("\n[AGENTS.md truncated by configured read limit]\n");
640        }
641    }
642    if !skills.is_empty() {
643        prompt.push_str("\n# Available skills\n");
644        prompt.push_str(skills);
645        prompt.push_str("\nUse read_skill with a skill name when its workflow applies.\n");
646    }
647    Ok(prompt)
648}
649
650fn discover_skills(workspace: &Path, config: &Config) -> Result<(SkillMap, Vec<PathBuf>, String)> {
651    let mut skills = SkillMap::new();
652    let mut roots = Vec::new();
653    let project_root = workspace.join(&config.skills.project_dir);
654    for (root, must_be_workspace) in [(&project_root, true), (&config.skills.user_dir, false)] {
655        if !root.is_dir() {
656            continue;
657        }
658        let canonical = std::fs::canonicalize(root)
659            .with_context(|| format!("resolve skill root {}", root.display()))?;
660        if must_be_workspace && !canonical.starts_with(workspace) {
661            return Err(anyhow!("project skill root escaped workspace"));
662        }
663        roots.push(canonical.clone());
664        let mut entries: Vec<_> = std::fs::read_dir(&canonical)
665            .with_context(|| format!("read skill root {}", canonical.display()))?
666            .filter_map(Result::ok)
667            .collect();
668        entries.sort_by_key(|entry| entry.file_name());
669        for entry in entries {
670            if skills.len() >= config.skills.max_skills {
671                break;
672            }
673            let path = entry.path().join("SKILL.md");
674            if !path.is_file() {
675                continue;
676            }
677            let canonical_file = std::fs::canonicalize(&path)
678                .with_context(|| format!("resolve skill {}", path.display()))?;
679            if !canonical_file.starts_with(&canonical) {
680                continue;
681            }
682            let name = entry.file_name().to_string_lossy().to_string();
683            skills.entry(name).or_insert(canonical_file);
684        }
685    }
686    let mut names: Vec<_> = skills.keys().cloned().collect();
687    names.sort();
688    let mut listing = String::new();
689    for name in names {
690        let path = &skills[&name];
691        let bytes = read_prefix(path, config.skills.max_skill_bytes)
692            .map(|(bytes, _)| bytes)
693            .unwrap_or_default();
694        let content = String::from_utf8_lossy(&bytes);
695        let description = skill_description(&content);
696        listing.push_str(&format!("- {name}: {description}\n"));
697    }
698    Ok((skills, roots, listing))
699}
700
701fn read_prefix(path: &Path, max_bytes: usize) -> std::io::Result<(Vec<u8>, bool)> {
702    let file = std::fs::File::open(path)?;
703    let mut bytes = Vec::with_capacity(max_bytes.min(8192));
704    file.take(
705        u64::try_from(max_bytes)
706            .unwrap_or(u64::MAX)
707            .saturating_add(1),
708    )
709    .read_to_end(&mut bytes)?;
710    let truncated = bytes.len() > max_bytes;
711    bytes.truncate(max_bytes);
712    Ok((bytes, truncated))
713}
714
715fn skill_description(content: &str) -> String {
716    if let Some(frontmatter) = content.strip_prefix("---\n")
717        && let Some((header, _)) = frontmatter.split_once("\n---")
718    {
719        for line in header.lines() {
720            if let Some(description) = line.strip_prefix("description:") {
721                return description.trim().trim_matches('"').to_owned();
722            }
723        }
724    }
725    content
726        .lines()
727        .map(str::trim)
728        .find(|line| !line.is_empty() && !line.starts_with('#'))
729        .unwrap_or("No description provided")
730        .chars()
731        .take(240)
732        .collect()
733}
734
735struct ProtocolSink {
736    meta: TurnMeta,
737    output: OutboundSender,
738    cancellation: CancellationToken,
739}
740
741#[async_trait]
742impl EventSink for ProtocolSink {
743    async fn emit(&self, event: CoreEvent) -> Result<(), AgentError> {
744        let seq = next_seq(&self.meta.seq);
745        let event = match event {
746            CoreEvent::AssistantDelta { content } => ServerEvent::AssistantDelta {
747                request_id: self.meta.request_id.clone(),
748                session_id: self.meta.session_id.clone(),
749                turn_id: self.meta.turn_id.clone(),
750                seq,
751                content,
752            },
753            CoreEvent::AssistantCompleted { content } => ServerEvent::AssistantCompleted {
754                request_id: self.meta.request_id.clone(),
755                session_id: self.meta.session_id.clone(),
756                turn_id: self.meta.turn_id.clone(),
757                seq,
758                content,
759            },
760            CoreEvent::ToolProposed {
761                call_id,
762                name,
763                arguments,
764            } => ServerEvent::ToolProposed {
765                request_id: self.meta.request_id.clone(),
766                session_id: self.meta.session_id.clone(),
767                turn_id: self.meta.turn_id.clone(),
768                seq,
769                call_id,
770                name,
771                arguments,
772            },
773            CoreEvent::ToolStarted { call_id, name } => ServerEvent::ToolStarted {
774                request_id: self.meta.request_id.clone(),
775                session_id: self.meta.session_id.clone(),
776                turn_id: self.meta.turn_id.clone(),
777                seq,
778                call_id,
779                name,
780            },
781            CoreEvent::ToolCompleted {
782                call_id,
783                name,
784                output,
785            } => ServerEvent::ToolCompleted {
786                request_id: self.meta.request_id.clone(),
787                session_id: self.meta.session_id.clone(),
788                turn_id: self.meta.turn_id.clone(),
789                seq,
790                call_id,
791                name,
792                success: !output.is_error,
793                output: output.content,
794                truncated: output.truncated,
795            },
796            CoreEvent::ContextCompacted {
797                before_tokens,
798                after_tokens,
799                removed_messages,
800            } => ServerEvent::ContextCompacted {
801                request_id: self.meta.request_id.clone(),
802                session_id: self.meta.session_id.clone(),
803                turn_id: self.meta.turn_id.clone(),
804                seq,
805                before_tokens,
806                after_tokens,
807                removed_messages,
808            },
809            CoreEvent::SessionTrimmed {
810                removed_messages,
811                history_bytes,
812            } => ServerEvent::SessionTrimmed {
813                request_id: self.meta.request_id.clone(),
814                session_id: self.meta.session_id.clone(),
815                seq,
816                removed_messages,
817                history_bytes,
818            },
819        };
820        send_turn_event(
821            &self.output,
822            event,
823            self.meta.max_server_frame,
824            &self.cancellation,
825        )
826        .await
827    }
828}
829
830#[derive(Default)]
831struct ApprovalBroker {
832    pending: Mutex<HashMap<String, oneshot::Sender<bool>>>,
833}
834
835impl ApprovalBroker {
836    async fn insert(&self, id: String, sender: oneshot::Sender<bool>) {
837        self.pending.lock().await.insert(id, sender);
838    }
839
840    async fn remove(&self, id: &str) {
841        self.pending.lock().await.remove(id);
842    }
843
844    async fn resolve(&self, id: &str, approved: bool) -> bool {
845        let sender = self.pending.lock().await.remove(id);
846        sender.is_some_and(|sender| sender.send(approved).is_ok())
847    }
848}
849
850struct ProtocolApprovalGate {
851    policy: ApprovalPolicy,
852    broker: Arc<ApprovalBroker>,
853    meta: TurnMeta,
854    output: OutboundSender,
855}
856
857#[async_trait]
858impl ApprovalGate for ProtocolApprovalGate {
859    async fn approve(
860        &self,
861        request: ApprovalRequest,
862        cancellation: CancellationToken,
863    ) -> Result<bool, AgentError> {
864        match self.policy {
865            ApprovalPolicy::OnRisk if request.risk == ToolRisk::ReadOnly => return Ok(true),
866            ApprovalPolicy::Never => return Ok(request.risk == ToolRisk::ReadOnly),
867            ApprovalPolicy::Always | ApprovalPolicy::OnRisk => {}
868        }
869        let approval_id = Uuid::new_v4().to_string();
870        let (sender, receiver) = oneshot::channel();
871        self.broker.insert(approval_id.clone(), sender).await;
872        let event = ServerEvent::ApprovalRequested {
873            request_id: self.meta.request_id.clone(),
874            session_id: self.meta.session_id.clone(),
875            turn_id: self.meta.turn_id.clone(),
876            seq: next_seq(&self.meta.seq),
877            approval_id: approval_id.clone(),
878            call_id: request.call_id,
879            name: request.name,
880            risk: request.risk.as_str().into(),
881            cwd: request.cwd.display().to_string(),
882            summary: request.summary,
883        };
884        if let Err(error) = send_turn_event(
885            &self.output,
886            event,
887            self.meta.max_server_frame,
888            &cancellation,
889        )
890        .await
891        {
892            self.broker.remove(&approval_id).await;
893            return Err(error);
894        }
895        tokio::select! {
896            result = receiver => result.map_err(|_| AgentError::Cancelled),
897            _ = cancellation.cancelled() => {
898                self.broker.remove(&approval_id).await;
899                Err(AgentError::Cancelled)
900            }
901        }
902    }
903}
904
905async fn send_event(output: &OutboundSender, event: ServerEvent, max_bytes: usize) -> Result<()> {
906    let bytes = encode_event(&event, max_bytes)?;
907    output.send(bytes, None).await.map_err(anyhow::Error::new)
908}
909
910async fn send_turn_event(
911    output: &OutboundSender,
912    event: ServerEvent,
913    max_bytes: usize,
914    cancellation: &CancellationToken,
915) -> Result<(), AgentError> {
916    let bytes = encode_event(&event, max_bytes)
917        .map_err(|error| AgentError::ResponseLimit(error.to_string()))?;
918    match output.send(bytes, Some(cancellation)).await {
919        Ok(()) => Ok(()),
920        Err(OutboundSendError::Cancelled) => Err(AgentError::Cancelled),
921        Err(error) => Err(AgentError::Internal(error.to_string())),
922    }
923}
924
925fn encode_event(event: &ServerEvent, max_bytes: usize) -> Result<Vec<u8>> {
926    let bytes = serde_json::to_vec(event).context("serialize protocol event")?;
927    if bytes.len() > max_bytes {
928        return Err(anyhow!("server event exceeds configured frame limit"));
929    }
930    Ok(bytes)
931}
932
933async fn send_error(
934    output: &OutboundSender,
935    request_id: &str,
936    code: &str,
937    message: &str,
938    fatal: bool,
939    max_bytes: usize,
940) -> Result<()> {
941    send_event(
942        output,
943        ServerEvent::Error {
944            request_id: (!request_id.is_empty()).then(|| request_id.to_owned()),
945            code: code.into(),
946            message: message.into(),
947            fatal,
948        },
949        max_bytes,
950    )
951    .await
952}
953
954#[cfg(test)]
955mod tests {
956    use std::{
957        future::pending,
958        io::Cursor,
959        sync::atomic::{AtomicBool, Ordering},
960    };
961
962    use super::*;
963
964    struct DropSignal(Arc<AtomicBool>);
965
966    impl Drop for DropSignal {
967        fn drop(&mut self) {
968            self.0.store(true, Ordering::Release);
969        }
970    }
971
972    #[tokio::test]
973    async fn bounded_reader_discards_an_oversized_line() {
974        let input = format!("{}\n{{}}\n", "x".repeat(10));
975        let mut reader = BufReader::new(Cursor::new(input.into_bytes()));
976        assert!(matches!(
977            read_bounded_frame(&mut reader, 4).await.unwrap(),
978            FrameRead::TooLarge
979        ));
980        match read_bounded_frame(&mut reader, 4).await.unwrap() {
981            FrameRead::Frame(frame) => assert_eq!(frame, b"{}"),
982            _ => panic!("expected the frame following the oversized line"),
983        }
984    }
985
986    #[tokio::test]
987    async fn bounded_reader_accepts_exact_crlf_limit() {
988        let mut reader = BufReader::new(Cursor::new(b"1234\r\n".to_vec()));
989        match read_bounded_frame(&mut reader, 4).await.unwrap() {
990            FrameRead::Frame(frame) => assert_eq!(frame, b"1234"),
991            _ => panic!("expected an exact-limit frame"),
992        }
993    }
994
995    #[tokio::test]
996    async fn outbound_byte_backpressure_is_cancellation_aware() {
997        let (output, mut receiver) = outbound_channel(5);
998        output.send(vec![0; 4], None).await.unwrap();
999
1000        let cancellation = CancellationToken::new();
1001        let blocked = tokio::spawn({
1002            let output = output.clone();
1003            let cancellation = cancellation.clone();
1004            async move { output.send(vec![1; 4], Some(&cancellation)).await }
1005        });
1006        tokio::task::yield_now().await;
1007        assert!(!blocked.is_finished());
1008
1009        cancellation.cancel();
1010        assert_eq!(blocked.await.unwrap(), Err(OutboundSendError::Cancelled));
1011
1012        drop(receiver.recv().await.unwrap());
1013        output.send(vec![2; 4], None).await.unwrap();
1014    }
1015
1016    #[tokio::test]
1017    async fn outbound_control_send_times_out_under_byte_backpressure() {
1018        let (output, _receiver) = outbound_channel(5);
1019        output.send(vec![0; 4], None).await.unwrap();
1020        let result = output
1021            .send_with_timeout(vec![1; 4], None, Duration::from_millis(10))
1022            .await;
1023        assert_eq!(result, Err(OutboundSendError::TimedOut));
1024    }
1025
1026    #[tokio::test]
1027    async fn active_turn_shutdown_aborts_after_grace_period() {
1028        let cancellation = CancellationToken::new();
1029        let dropped = Arc::new(AtomicBool::new(false));
1030        let (started_tx, started_rx) = oneshot::channel();
1031        let task = tokio::spawn({
1032            let dropped = Arc::clone(&dropped);
1033            async move {
1034                let _signal = DropSignal(dropped);
1035                let _ = started_tx.send(());
1036                pending::<()>().await;
1037            }
1038        });
1039        started_rx.await.unwrap();
1040
1041        let graceful = shutdown_active_turn(
1042            ActiveTurn {
1043                turn_id: "turn".into(),
1044                cancellation,
1045                task,
1046            },
1047            Duration::from_millis(10),
1048        )
1049        .await;
1050
1051        assert!(!graceful);
1052        assert!(dropped.load(Ordering::Acquire));
1053    }
1054
1055    #[tokio::test]
1056    async fn writer_shutdown_aborts_after_grace_period() {
1057        let dropped = Arc::new(AtomicBool::new(false));
1058        let (started_tx, started_rx) = oneshot::channel();
1059        let writer = tokio::spawn({
1060            let dropped = Arc::clone(&dropped);
1061            async move {
1062                let _signal = DropSignal(dropped);
1063                let _ = started_tx.send(());
1064                pending::<std::io::Result<()>>().await
1065            }
1066        });
1067        started_rx.await.unwrap();
1068
1069        let result = shutdown_writer(writer, Duration::from_millis(10)).await;
1070
1071        assert!(result.is_err());
1072        assert!(dropped.load(Ordering::Acquire));
1073    }
1074}
scv_server/lib.rs

scv_server/
lib.rs
