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};
20use scv_core::{
21    AgentError, AgentRuntime, ApprovalGate, ApprovalRequest, BudgetContextPolicy, CoreEvent,
22    EventSink, Message, ToolRisk,
23};
24use scv_protocol::{ClientMessage, PROTOCOL_VERSION, PeerInfo, ServerEvent, Usage};
25use scv_provider_openai::OpenAiProvider;
26use scv_tools::{SkillMap, builtin_registry};
27use tokio::{
28    io::{AsyncBufRead, AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader},
29    sync::{Mutex, OwnedSemaphorePermit, Semaphore, mpsc, oneshot},
30    task::JoinHandle,
31};
32use tokio_util::sync::CancellationToken;
33use uuid::Uuid;
34
35const PROMPT_LIMIT_BYTES: usize = 256 * 1024;
36const OUTPUT_QUEUE_CAPACITY: usize = 256;
37const OUTPUT_QUEUE_MIN_BYTES: usize = 16 * 1024 * 1024;
38const SHUTDOWN_GRACE: Duration = Duration::from_secs(3);
39
40pub async fn run_stdio(overrides: ConfigOverrides) -> Result<()> {
41    let stdin = tokio::io::stdin();
42    let stdout = tokio::io::stdout();
43    run(stdin, stdout, overrides).await
44}
45
46async fn run<R, W>(reader: R, writer: W, overrides: ConfigOverrides) -> Result<()>
47where
48    R: tokio::io::AsyncRead + Unpin,
49    W: tokio::io::AsyncWrite + Unpin + Send + 'static,
50{
51    let initial_output_bytes =
52        output_queue_bytes(Config::default().protocol.max_server_frame_bytes)?;
53    let (output_tx, mut output_rx) = outbound_channel(initial_output_bytes);
54    let mut writer_task = tokio::spawn(async move {
55        let mut writer = writer;
56        while let Some(frame) = output_rx.recv().await {
57            writer.write_all(&frame.bytes).await?;
58            writer.write_all(b"\n").await?;
59            writer.flush().await?;
60        }
61        Ok::<(), std::io::Error>(())
62    });
63    let (done_tx, mut done_rx) = mpsc::channel::<TurnDone>(4);
64    let approvals = Arc::new(ApprovalBroker::default());
65    let mut reader = BufReader::new(reader);
66    let mut initialized = false;
67    let mut session: Option<Session> = None;
68    let mut active: Option<ActiveTurn> = None;
69    let mut fatal = false;
70    let mut writer_finished = false;
71
72    let loop_result: Result<()> = async {
73        loop {
74        let frame_limit = session.as_ref().map_or_else(
75            || Config::default().protocol.max_client_frame_bytes,
76            |value| value.config.protocol.max_client_frame_bytes,
77        );
78        tokio::select! {
79            read = read_bounded_frame(&mut reader, frame_limit) => {
80                let frame = match read.context("read protocol input")? {
81                    FrameRead::Eof => {
82                        if let Some(active) = &active { active.cancellation.cancel(); }
83                        break;
84                    }
85                    FrameRead::TooLarge => {
86                        send_error(&output_tx, "", "invalid_request", "client frame exceeds configured limit", false, server_frame_limit(&session)).await?;
87                        continue;
88                    }
89                    FrameRead::Frame(frame) => frame,
90                };
91                if frame.is_empty() {
92                    send_error(&output_tx, "", "invalid_json", "protocol frame is empty", false, server_frame_limit(&session)).await?;
93                    continue;
94                }
95                let message = match serde_json::from_slice::<ClientMessage>(&frame) {
96                    Ok(message) => message,
97                    Err(error) => {
98                        send_error(&output_tx, "", "invalid_json", &format!("invalid protocol JSON: {error}"), false, server_frame_limit(&session)).await?;
99                        continue;
100                    }
101                };
102                match message {
103                    ClientMessage::Initialize { request_id, protocol_version, .. } => {
104                        if initialized {
105                            send_error(&output_tx, &request_id, "invalid_request", "connection is already initialized", false, server_frame_limit(&session)).await?;
106                            continue;
107                        }
108                        if protocol_version != PROTOCOL_VERSION {
109                            send_error(&output_tx, &request_id, "version_mismatch", &format!("server supports protocol {PROTOCOL_VERSION}"), true, server_frame_limit(&session)).await?;
110                            fatal = true;
111                            break;
112                        }
113                        initialized = true;
114                        send_event(&output_tx, ServerEvent::Initialized {
115                            request_id,
116                            protocol_version: PROTOCOL_VERSION,
117                            server: PeerInfo { name: "scv-server".into(), version: env!("CARGO_PKG_VERSION").into() },
118                        }, Config::default().protocol.max_server_frame_bytes).await?;
119                    }
120                    other if !initialized => {
121                        send_error(&output_tx, other.request_id(), "not_initialized", "initialize must be the first message", false, server_frame_limit(&session)).await?;
122                    }
123                    ClientMessage::SessionStart { request_id, cwd } => {
124                        if session.is_some() {
125                            send_error(&output_tx, &request_id, "invalid_request", "this connection already has a session", false, server_frame_limit(&session)).await?;
126                            continue;
127                        }
128                        match build_session(&cwd, overrides.clone()).await {
129                            Ok(new_session) => {
130                                output_tx.ensure_capacity(output_queue_bytes(
131                                    new_session.config.protocol.max_server_frame_bytes,
132                                )?)?;
133                                let event = ServerEvent::SessionStarted {
134                                    request_id,
135                                    session_id: new_session.id.clone(),
136                                    cwd: new_session.workspace.display().to_string(),
137                                    model: new_session.runtime.model().to_owned(),
138                                    context_max_tokens: new_session.config.context.max_tokens,
139                                    max_server_frame_bytes: new_session.config.protocol.max_server_frame_bytes,
140                                    max_transcript_bytes: new_session.config.tui.max_transcript_bytes,
141                                    max_transcript_items: new_session.config.tui.max_transcript_items,
142                                    max_prompt_history_bytes: new_session.config.tui.max_prompt_history_bytes,
143                                    max_prompt_history_items: new_session.config.tui.max_prompt_history_items,
144                                };
145                                send_event(&output_tx, event, new_session.config.protocol.max_server_frame_bytes).await?;
146                                session = Some(new_session);
147                            }
148                            Err(error) => {
149                                send_error(&output_tx, &request_id, "invalid_request", &error.to_string(), false, server_frame_limit(&session)).await?;
150                            }
151                        }
152                    }
153                    ClientMessage::TurnStart { request_id, session_id, prompt } => {
154                        let Some(current) = session.as_ref() else {
155                            send_error(&output_tx, &request_id, "session_not_found", "start a session first", false, server_frame_limit(&session)).await?;
156                            continue;
157                        };
158                        if current.id != session_id {
159                            send_error(&output_tx, &request_id, "session_not_found", "session id does not match", false, server_frame_limit(&session)).await?;
160                            continue;
161                        }
162                        if active.is_some() {
163                            send_error(&output_tx, &request_id, "turn_active", "a turn is already active", false, server_frame_limit(&session)).await?;
164                            continue;
165                        }
166                        if prompt.trim().is_empty() || prompt.len() > PROMPT_LIMIT_BYTES {
167                            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?;
168                            continue;
169                        }
170                        let turn_id = Uuid::new_v4().to_string();
171                        let cancellation = CancellationToken::new();
172                        let meta = TurnMeta {
173                            request_id: request_id.clone(),
174                            session_id: current.id.clone(),
175                            turn_id: turn_id.clone(),
176                            seq: Arc::clone(&current.seq),
177                            max_server_frame: current.config.protocol.max_server_frame_bytes,
178                        };
179                        send_event(&output_tx, ServerEvent::TurnStarted {
180                            request_id: request_id.clone(),
181                            session_id: current.id.clone(),
182                            turn_id: turn_id.clone(),
183                            seq: next_seq(&current.seq),
184                        }, current.config.protocol.max_server_frame_bytes).await?;
185                        let runtime = Arc::clone(&current.runtime);
186                        let history = Arc::clone(&current.history);
187                        let sink: Arc<dyn EventSink> = Arc::new(ProtocolSink {
188                            meta: meta.clone(),
189                            output: output_tx.clone(),
190                            cancellation: cancellation.clone(),
191                        });
192                        let gate: Arc<dyn ApprovalGate> = Arc::new(ProtocolApprovalGate {
193                            policy: current.config.tools.approval_policy,
194                            broker: Arc::clone(&approvals),
195                            meta,
196                            output: output_tx.clone(),
197                        });
198                        let task_cancel = cancellation.clone();
199                        let task_done = done_tx.clone();
200                        let task_request = request_id.clone();
201                        let task_session = current.id.clone();
202                        let task_turn = turn_id.clone();
203                        let task = tokio::spawn(async move {
204                            let mut history = history.lock().await;
205                            let result = runtime.run_turn(&mut history, prompt, sink, gate, task_cancel).await;
206                            let _ = task_done.send(TurnDone {
207                                request_id: task_request,
208                                session_id: task_session,
209                                turn_id: task_turn,
210                                result,
211                            }).await;
212                        });
213                        active = Some(ActiveTurn { turn_id, cancellation, task });
214                    }
215                    ClientMessage::TurnCancel { request_id, session_id, turn_id } => {
216                        match (&session, &active) {
217                            (Some(current), Some(running)) if current.id == session_id && running.turn_id == turn_id => running.cancellation.cancel(),
218                            _ => send_error(&output_tx, &request_id, "turn_not_found", "active turn was not found", false, server_frame_limit(&session)).await?,
219                        }
220                    }
221                    ClientMessage::ApprovalResolve { request_id, session_id, approval_id, approved } => {
222                        if session.as_ref().is_none_or(|current| current.id != session_id) {
223                            send_error(&output_tx, &request_id, "session_not_found", "session id does not match", false, server_frame_limit(&session)).await?;
224                        } else if !approvals.resolve(&approval_id, approved).await {
225                            send_error(&output_tx, &request_id, "approval_not_found", "approval was not found or already resolved", false, server_frame_limit(&session)).await?;
226                        }
227                    }
228                    ClientMessage::SessionClear { request_id, session_id } => {
229                        let Some(current) = session.as_ref() else {
230                            send_error(&output_tx, &request_id, "session_not_found", "session was not found", false, server_frame_limit(&session)).await?;
231                            continue;
232                        };
233                        if current.id != session_id {
234                            send_error(&output_tx, &request_id, "session_not_found", "session id does not match", false, server_frame_limit(&session)).await?;
235                        } else if active.is_some() {
236                            send_error(&output_tx, &request_id, "turn_active", "cancel the active turn before clearing", false, server_frame_limit(&session)).await?;
237                        } else {
238                            current.history.lock().await.clear();
239                            send_event(&output_tx, ServerEvent::SessionCleared {
240                                request_id,
241                                session_id: current.id.clone(),
242                                seq: next_seq(&current.seq),
243                            }, current.config.protocol.max_server_frame_bytes).await?;
244                        }
245                    }
246                }
247            }
248            writer = &mut writer_task => {
249                writer_finished = true;
250                writer.context("join protocol writer")??;
251                break;
252            }
253            done = done_rx.recv(), if active.is_some() => {
254                if let Some(done) = done {
255                    if let Some(current) = session.as_ref() {
256                        let seq = next_seq(&current.seq);
257                        let event = match done.result {
258                            Ok(outcome) => ServerEvent::TurnCompleted {
259                                request_id: done.request_id,
260                                session_id: done.session_id,
261                                turn_id: done.turn_id,
262                                seq,
263                                steps: outcome.steps,
264                                usage: Usage { input_tokens: outcome.usage.input_tokens, output_tokens: outcome.usage.output_tokens },
265                            },
266                            Err(AgentError::Cancelled) => ServerEvent::TurnCancelled {
267                                request_id: done.request_id,
268                                session_id: done.session_id,
269                                turn_id: done.turn_id,
270                                seq,
271                            },
272                            Err(error) => ServerEvent::TurnFailed {
273                                request_id: done.request_id,
274                                session_id: done.session_id,
275                                turn_id: done.turn_id,
276                                seq,
277                                code: error.code().into(),
278                                message: error.to_string(),
279                            },
280                        };
281                        send_event(&output_tx, event, current.config.protocol.max_server_frame_bytes).await?;
282                    }
283                    if let Some(active) = active.take() {
284                        let _ = active.task.await;
285                    }
286                }
287            }
288        }
289        }
290        Ok(())
291    }
292    .await;
293
294    if let Some(active) = active.take() {
295        shutdown_active_turn(active, SHUTDOWN_GRACE).await;
296    }
297    drop(output_tx);
298    let writer_result = if writer_finished {
299        Ok(())
300    } else {
301        shutdown_writer(writer_task, SHUTDOWN_GRACE).await
302    };
303    loop_result?;
304    writer_result?;
305    if fatal {
306        return Err(anyhow!("protocol version mismatch"));
307    }
308    Ok(())
309}
310
311enum FrameRead {
312    Eof,
313    Frame(Vec<u8>),
314    TooLarge,
315}
316
317struct OutboundFrame {
318    bytes: Vec<u8>,
319    _byte_permit: OwnedSemaphorePermit,
320}
321
322#[derive(Clone)]
323struct OutboundSender {
324    frames: mpsc::Sender<OutboundFrame>,
325    budget: Arc<Semaphore>,
326    capacity: Arc<AtomicUsize>,
327}
328
329#[derive(Debug, PartialEq, Eq)]
330enum OutboundSendError {
331    Cancelled,
332    Closed,
333    TimedOut,
334    FrameExceedsQueue { frame_bytes: usize, capacity: usize },
335}
336
337impl std::fmt::Display for OutboundSendError {
338    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
339        match self {
340            Self::Cancelled => formatter.write_str("outbound send cancelled"),
341            Self::Closed => formatter.write_str("protocol client disconnected"),
342            Self::TimedOut => formatter.write_str("outbound send timed out under backpressure"),
343            Self::FrameExceedsQueue {
344                frame_bytes,
345                capacity,
346            } => write!(
347                formatter,
348                "outbound frame uses {frame_bytes} bytes but queue capacity is {capacity} bytes"
349            ),
350        }
351    }
352}
353
354impl std::error::Error for OutboundSendError {}
355
356fn outbound_channel(capacity: usize) -> (OutboundSender, mpsc::Receiver<OutboundFrame>) {
357    let (frames, receiver) = mpsc::channel(OUTPUT_QUEUE_CAPACITY);
358    (
359        OutboundSender {
360            frames,
361            budget: Arc::new(Semaphore::new(capacity)),
362            capacity: Arc::new(AtomicUsize::new(capacity)),
363        },
364        receiver,
365    )
366}
367
368impl OutboundSender {
369    fn ensure_capacity(&self, required: usize) -> Result<()> {
370        if required > Semaphore::MAX_PERMITS {
371            return Err(anyhow!(
372                "outbound queue capacity {required} exceeds runtime limit {}",
373                Semaphore::MAX_PERMITS
374            ));
375        }
376        let current = self.capacity.load(Ordering::Acquire);
377        if required > current {
378            self.budget.add_permits(required - current);
379            self.capacity.store(required, Ordering::Release);
380        }
381        Ok(())
382    }
383
384    async fn send(
385        &self,
386        bytes: Vec<u8>,
387        cancellation: Option<&CancellationToken>,
388    ) -> std::result::Result<(), OutboundSendError> {
389        self.send_with_timeout(bytes, cancellation, SHUTDOWN_GRACE)
390            .await
391    }
392
393    async fn send_with_timeout(
394        &self,
395        bytes: Vec<u8>,
396        cancellation: Option<&CancellationToken>,
397        control_timeout: Duration,
398    ) -> std::result::Result<(), OutboundSendError> {
399        let frame_bytes =
400            bytes
401                .len()
402                .checked_add(1)
403                .ok_or(OutboundSendError::FrameExceedsQueue {
404                    frame_bytes: usize::MAX,
405                    capacity: self.capacity.load(Ordering::Acquire),
406                })?;
407        let capacity = self.capacity.load(Ordering::Acquire);
408        let permits =
409            u32::try_from(frame_bytes).map_err(|_| OutboundSendError::FrameExceedsQueue {
410                frame_bytes,
411                capacity,
412            })?;
413        if frame_bytes > capacity {
414            return Err(OutboundSendError::FrameExceedsQueue {
415                frame_bytes,
416                capacity,
417            });
418        }
419
420        let control_deadline = tokio::time::Instant::now() + control_timeout;
421        let acquire = Arc::clone(&self.budget).acquire_many_owned(permits);
422        let permit = if let Some(cancellation) = cancellation {
423            tokio::select! {
424                biased;
425                _ = cancellation.cancelled() => return Err(OutboundSendError::Cancelled),
426                permit = acquire => permit.map_err(|_| OutboundSendError::Closed)?,
427            }
428        } else {
429            tokio::time::timeout_at(control_deadline, acquire)
430                .await
431                .map_err(|_| OutboundSendError::TimedOut)?
432                .map_err(|_| OutboundSendError::Closed)?
433        };
434        let frame = OutboundFrame {
435            bytes,
436            _byte_permit: permit,
437        };
438        if let Some(cancellation) = cancellation {
439            tokio::select! {
440                biased;
441                _ = cancellation.cancelled() => Err(OutboundSendError::Cancelled),
442                result = self.frames.send(frame) => result.map_err(|_| OutboundSendError::Closed),
443            }
444        } else {
445            tokio::time::timeout_at(control_deadline, self.frames.send(frame))
446                .await
447                .map_err(|_| OutboundSendError::TimedOut)?
448                .map_err(|_| OutboundSendError::Closed)
449        }
450    }
451}
452
453fn output_queue_bytes(max_frame_bytes: usize) -> Result<usize> {
454    let required = max_frame_bytes
455        .checked_add(1)
456        .and_then(|bytes| bytes.checked_mul(2))
457        .ok_or_else(|| anyhow!("configured server frame limit is too large"))?
458        .max(OUTPUT_QUEUE_MIN_BYTES);
459    if required > Semaphore::MAX_PERMITS {
460        return Err(anyhow!(
461            "configured server frame limit requires an outbound queue larger than the runtime supports"
462        ));
463    }
464    Ok(required)
465}
466
467async fn read_bounded_frame<R>(reader: &mut R, max_bytes: usize) -> std::io::Result<FrameRead>
468where
469    R: AsyncBufRead + Unpin,
470{
471    let mut frame = Vec::with_capacity(max_bytes.min(8192));
472    let read_limit = u64::try_from(max_bytes)
473        .unwrap_or(u64::MAX)
474        .saturating_add(2);
475    let mut limited = reader.take(read_limit);
476    let read = limited.read_until(b'\n', &mut frame).await?;
477    drop(limited);
478    if read == 0 {
479        return Ok(FrameRead::Eof);
480    }
481    let ended_with_newline = frame.last() == Some(&b'\n');
482    while matches!(frame.last(), Some(b'\n' | b'\r')) {
483        frame.pop();
484    }
485    if frame.len() <= max_bytes {
486        return Ok(FrameRead::Frame(frame));
487    }
488    if !ended_with_newline {
489        loop {
490            let available = reader.fill_buf().await?;
491            if available.is_empty() {
492                break;
493            }
494            if let Some(end) = available.iter().position(|byte| *byte == b'\n') {
495                reader.consume(end + 1);
496                break;
497            }
498            let consumed = available.len();
499            reader.consume(consumed);
500        }
501    }
502    Ok(FrameRead::TooLarge)
503}
504
505fn server_frame_limit(session: &Option<Session>) -> usize {
506    session.as_ref().map_or_else(
507        || Config::default().protocol.max_server_frame_bytes,
508        |value| value.config.protocol.max_server_frame_bytes,
509    )
510}
511
512struct Session {
513    id: String,
514    workspace: PathBuf,
515    config: Config,
516    runtime: Arc<AgentRuntime>,
517    history: Arc<Mutex<Vec<Message>>>,
518    seq: Arc<AtomicU64>,
519}
520
521struct ActiveTurn {
522    turn_id: String,
523    cancellation: CancellationToken,
524    task: JoinHandle<()>,
525}
526
527async fn shutdown_active_turn(mut active: ActiveTurn, grace: Duration) -> bool {
528    active.cancellation.cancel();
529    if tokio::time::timeout(grace, &mut active.task).await.is_ok() {
530        true
531    } else {
532        active.task.abort();
533        let _ = active.task.await;
534        false
535    }
536}
537
538async fn shutdown_writer(
539    mut writer: JoinHandle<std::io::Result<()>>,
540    grace: Duration,
541) -> Result<()> {
542    match tokio::time::timeout(grace, &mut writer).await {
543        Ok(result) => {
544            result.context("join protocol writer")??;
545            Ok(())
546        }
547        Err(_) => {
548            writer.abort();
549            let _ = writer.await;
550            Err(anyhow!("protocol writer shutdown timed out"))
551        }
552    }
553}
554
555struct TurnDone {
556    request_id: String,
557    session_id: String,
558    turn_id: String,
559    result: Result<scv_core::TurnOutcome, AgentError>,
560}
561
562#[derive(Clone)]
563struct TurnMeta {
564    request_id: String,
565    session_id: String,
566    turn_id: String,
567    seq: Arc<AtomicU64>,
568    max_server_frame: usize,
569}
570
571fn next_seq(sequence: &AtomicU64) -> u64 {
572    sequence.fetch_add(1, Ordering::Relaxed) + 1
573}
574
575async fn build_session(cwd: &str, overrides: ConfigOverrides) -> Result<Session> {
576    let workspace = std::fs::canonicalize(cwd).with_context(|| format!("resolve cwd {cwd}"))?;
577    if !workspace.is_dir() {
578        return Err(anyhow!("cwd is not a directory"));
579    }
580    let config = Config::load(&workspace, overrides)?;
581    let api_key = std::env::var(&config.provider.api_key_env).with_context(|| {
582        format!(
583            "provider credential environment variable {} is not set",
584            config.provider.api_key_env
585        )
586    })?;
587    let (skills, skill_roots, skill_prompt) = discover_skills(&workspace, &config)?;
588    let system_prompt = build_system_prompt(&workspace, &config, &skill_prompt)?;
589    let provider = Arc::new(OpenAiProvider::new(
590        config.provider.model.clone(),
591        config.provider.base_url.clone(),
592        api_key,
593        Duration::from_secs(config.provider.timeout_seconds),
594        config.provider_limits(),
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
