artifactize_core/inspect/
logs.rs

1//! Bounded runtime evidence queries. Full captures stay on disk, never in a tool result.
2use crate::{store, types::RequestId};
3use serde::{Deserialize, Serialize};
4use std::{
5    collections::VecDeque,
6    io::{BufRead, BufReader, Cursor, Read},
7    path::Path,
8};
9
10/// Keep two streams, including JSON escaping and MCP's text mirror, below 64 KiB.
11const STREAM_BYTES: usize = 10 * 1024;
12/// One query is a diagnostic excerpt, not a replacement for the saved log.
13const MAX_LINES: usize = 200;
14/// A short default tail supplies diagnostic context without filling an agent context.
15const DEFAULT_TAIL: usize = 50;
16/// Bound one unusually long line independently of the stream page.
17const LINE_BYTES: usize = 2048;
18
19#[derive(Debug, Clone, Copy, Default, Deserialize, Serialize, clap::ValueEnum)]
20#[serde(rename_all = "lowercase")]
21pub enum Stream {
22    Stdout,
23    Stderr,
24    #[default]
25    Both,
26}
27
28#[derive(Debug, Clone, Default, Deserialize, clap::Args)]
29#[serde(deny_unknown_fields)]
30pub struct LogQuery {
31    /// Select stdout, stderr, or both (default).
32    #[arg(long, value_enum, default_value = "both")]
33    #[serde(default)]
34    pub stream: Stream,
35    /// Last N matching lines; defaults to 50 when no range is given (maximum 200).
36    #[arg(long, conflicts_with_all = ["start", "end"])]
37    pub tail: Option<usize>,
38    /// First line, inclusive and one-based, in the original retained stream.
39    #[arg(long, conflicts_with = "tail")]
40    pub start: Option<usize>,
41    /// Last line, inclusive and one-based.
42    #[arg(long, conflicts_with = "tail")]
43    pub end: Option<usize>,
44    /// Literal, case-sensitive search; line numbers remain original.
45    #[arg(long)]
46    pub search: Option<String>,
47}
48#[derive(Debug, Serialize)]
49#[serde(rename_all = "snake_case")]
50pub enum Evidence {
51    Complete,
52    Partial,
53    SavedAudit,
54    Missing,
55    NotRuntime,
56}
57#[derive(Debug, Serialize)]
58pub struct Line {
59    pub number: usize,
60    pub text: String,
61    pub clipped: bool,
62}
63#[derive(Debug, Serialize)]
64pub struct LogExcerpt {
65    pub stream: Stream,
66    pub evidence: Evidence,
67    pub note: String,
68    pub lines: Vec<Line>,
69    /// Continue a forward range/search at this original line; null for a tail query.
70    pub next_start: Option<usize>,
71}
72#[derive(Debug, Serialize)]
73pub struct Logs {
74    pub request: RequestId,
75    pub producer: RequestId,
76    pub streams: Vec<LogExcerpt>,
77}
78
79pub async fn logs(state: &Path, id: &RequestId, query: LogQuery) -> Result<Logs, String> {
80    validate(&query)?;
81    let view = store::read_request(state, id).await?;
82    // Reused/joined results point at the execution's original request, not this Run's output.
83    let producer = view
84        .request
85        .provenance
86        .as_ref()
87        .map_or(id, |p| &p.request_id)
88        .clone();
89    let source = if view.request.origin.is_some() {
90        None
91    } else if producer == *id {
92        Some(view.clone())
93    } else {
94        // Remote mirrors need not carry the original local request.
95        store::read_request(state, &producer).await.ok()
96    };
97    let runtime = matches!(
98        view.request.result(),
99        Some(store::ExecutionResult::Runtime(_))
100    ) || matches!(
101        view.request.profile,
102        crate::config::StoredProfile::Runtime { .. }
103    );
104    let state = state.to_path_buf();
105    let id = id.clone();
106    crate::task::joined(
107        tokio::task::spawn_blocking(move || {
108            let streams = match query.stream {
109                Stream::Both => vec![Stream::Stdout, Stream::Stderr],
110                stream => vec![stream],
111            }
112            .into_iter()
113            .map(|stream| excerpt(&state, source.as_ref(), &view, runtime, stream, &query))
114            .collect::<Result<_, _>>()?;
115            Ok(Logs {
116                request: id,
117                producer,
118                streams,
119            })
120        })
121        .await,
122    )
123    .map_err(|e| e.to_string())?
124}
125
126fn validate(query: &LogQuery) -> Result<(), String> {
127    if query.tail.is_some_and(|n| n == 0 || n > MAX_LINES)
128        || query.start == Some(0)
129        || query.end == Some(0)
130        || query.start.zip(query.end).is_some_and(|(a, b)| a > b)
131        || (query.tail.is_some() && (query.start.is_some() || query.end.is_some()))
132        || query
133            .search
134            .as_ref()
135            .is_some_and(|s| s.is_empty() || s.len() > LINE_BYTES)
136    {
137        return Err("Use positive line numbers, tail 1–200, an ordered range, and a nonempty search up to 2048 bytes; tail and range are exclusive.".into());
138    }
139    Ok(())
140}
141
142fn excerpt(
143    state: &Path,
144    source: Option<&store::RequestView>,
145    view: &store::RequestView,
146    runtime: bool,
147    stream: Stream,
148    query: &LogQuery,
149) -> Result<LogExcerpt, String> {
150    let absent = |evidence, note: &str| LogExcerpt {
151        stream,
152        evidence,
153        note: note.into(),
154        lines: vec![],
155        next_start: None,
156    };
157    if !runtime {
158        return Ok(absent(
159            Evidence::NotRuntime,
160            "No runtime stdout/stderr; inspect request/session evidence.",
161        ));
162    }
163    let name = match stream {
164        Stream::Stdout => "stdout",
165        Stream::Stderr => "stderr",
166        Stream::Both => unreachable!(),
167    };
168    if let Some(directory) = source.and_then(|v| v.request.run_dir.as_ref()) {
169        let path = directory.join(format!("runtime.{name}"));
170        // Saved paths are not authority to read outside this state's Run output.
171        if path.exists() {
172            let root =
173                crate::platform::canonicalize(&state.join("runs")).map_err(|e| e.to_string())?;
174            let canonical = crate::platform::canonicalize(&path).map_err(|e| e.to_string())?;
175            if !crate::platform::is_within(&canonical, &root) {
176                return Err("Runtime log is outside saved Run output.".into());
177            }
178            let file = std::fs::File::open(&canonical).map_err(|e| e.to_string())?;
179            let complete = path.with_extension(format!("{name}.complete")).is_file();
180            // A live producer may keep appending: query only the bytes present at open.
181            let length = file.metadata().map_err(|e| e.to_string())?.len();
182            let (lines, next_start) = select(BufReader::new(file.take(length)), query)?;
183            return Ok(LogExcerpt {
184                stream,
185                evidence: if complete {
186                    Evidence::Complete
187                } else {
188                    Evidence::Partial
189                },
190                note: if complete {
191                    "Full local stream retained until Run output is pruned."
192                } else {
193                    "Capture is active or ended without confirmed EOF; retained bytes only."
194                }
195                .into(),
196                lines,
197                next_start,
198            });
199        }
200    }
201    let result = view
202        .request
203        .result()
204        .or_else(|| source.and_then(|v| v.request.result()));
205    if let Some(store::ExecutionResult::Runtime(result)) = result {
206        let audit = match stream {
207            Stream::Stdout => result.stdout.value(),
208            _ => result.stderr.value(),
209        };
210        if let Some(text) = audit {
211            let (lines, next_start) = select(Cursor::new(text.as_bytes()), query)?;
212            let note = match result.truncated.value() {
213                Some(false) => {
214                    "Full capture unavailable (older, remote or pruned); saved cleaned audit reports no truncation."
215                }
216                _ => {
217                    "Full capture unavailable (older, remote or pruned); audit may contain only the first 128 KiB. Missing tail cannot be recovered."
218                }
219            };
220            return Ok(LogExcerpt {
221                stream,
222                evidence: Evidence::SavedAudit,
223                note: note.into(),
224                lines,
225                next_start,
226            });
227        }
228    }
229    Ok(absent(
230        Evidence::Missing,
231        "No retained stream or saved audit: not started, remote, older or pruned evidence.",
232    ))
233}
234
235/// Scan bounded chunks even for newline-free output. Search overlaps chunks so a
236/// literal crossing a buffer boundary is found without retaining the entire line.
237fn select(
238    mut reader: impl BufRead,
239    query: &LogQuery,
240) -> Result<(Vec<Line>, Option<usize>), String> {
241    let tail = query
242        .tail
243        .or_else(|| (query.start.is_none() && query.end.is_none()).then_some(DEFAULT_TAIL));
244    let mut selected = VecDeque::new();
245    let mut bytes = 0;
246    let mut number = 0;
247    loop {
248        let mut text = Vec::new();
249        let mut suffix = VecDeque::new();
250        let mut overlap = Vec::new();
251        let mut length = 0;
252        let mut matched = query.search.is_none();
253        loop {
254            let chunk = reader.fill_buf().map_err(|e| e.to_string())?;
255            if chunk.is_empty() {
256                break;
257            }
258            let end = chunk.iter().position(|&b| b == b'\n').map(|n| n + 1);
259            let chunk = &chunk[..end.unwrap_or(chunk.len())];
260            length += chunk.len();
261            text.extend_from_slice(
262                &chunk[..chunk.len().min(LINE_BYTES.saturating_sub(text.len()))],
263            );
264            suffix.extend(chunk.iter().copied());
265            suffix.drain(..suffix.len().saturating_sub(LINE_BYTES / 2));
266            if let Some(search) = &query.search {
267                overlap.extend_from_slice(chunk);
268                matched |= overlap
269                    .windows(search.len())
270                    .any(|w| w == search.as_bytes());
271                let keep = overlap.len().saturating_sub(search.len().saturating_sub(1));
272                overlap.drain(..keep);
273            }
274            let consumed = chunk.len();
275            reader.consume(consumed);
276            if end.is_some() {
277                break;
278            }
279        }
280        if length == 0 {
281            break;
282        }
283        number += 1;
284        if query.end.is_some_and(|end| number > end) {
285            break;
286        }
287        if number < query.start.unwrap_or(1) || !matched {
288            continue;
289        }
290        let text = if length > LINE_BYTES {
291            text.truncate(LINE_BYTES / 2);
292            text.extend_from_slice(b"\n[... line clipped ...]\n");
293            text.extend(suffix);
294            text
295        } else {
296            text
297        };
298        let text = String::from_utf8_lossy(&crate::runtime::clean_output(&text))
299            .trim_end_matches(['\r', '\n'])
300            .to_owned();
301        let line = Line {
302            number,
303            text,
304            clipped: length > LINE_BYTES,
305        };
306        let size = serde_json::to_vec(&line).map_err(|e| e.to_string())?.len();
307        if tail.is_none() && (selected.len() == MAX_LINES || bytes + size > STREAM_BYTES) {
308            return Ok((selected.into(), Some(number)));
309        }
310        bytes += size;
311        selected.push_back(line);
312        while selected.len() > tail.unwrap_or(MAX_LINES) || bytes > STREAM_BYTES {
313            let removed = selected.pop_front().expect("nonempty page");
314            bytes -= serde_json::to_vec(&removed).expect("line serializes").len();
315        }
316    }
317    Ok((selected.into(), None))
318}
artifactize_core/inspect/logs.rs

artifactize_core/inspect/
logs.rs
