1use std::path::{Path, PathBuf};
2
3pub use roder_api::artifacts::CreateArtifactRequest;
4use roder_api::artifacts::{
5 ArtifactGrepPage, ArtifactReadPage, ArtifactTailPage, ContextArtifact, ContextArtifactAccess,
6 ContextArtifactId, ContextArtifactStore as SharedContextArtifactStore,
7};
8use roder_api::events::{ThreadId, TurnId};
9use time::OffsetDateTime;
10
11const DEFAULT_PAGE_LINES: usize = 200;
12const MAX_PAGE_LINES: usize = 200;
13
14#[derive(Debug, Clone)]
15pub struct ContextArtifactStore {
16 root: PathBuf,
17 layout: ArtifactStorageLayout,
18}
19
20#[derive(Debug, Clone, Copy, PartialEq, Eq)]
21enum ArtifactStorageLayout {
22 LegacyRoot,
23 ThreadScoped,
24}
25
26impl ContextArtifactStore {
27 pub fn new(root: impl Into<PathBuf>) -> Self {
28 Self {
29 root: root.into(),
30 layout: ArtifactStorageLayout::LegacyRoot,
31 }
32 }
33
34 pub fn new_thread_scoped(thread_root: impl Into<PathBuf>) -> Self {
35 Self {
36 root: thread_root.into(),
37 layout: ArtifactStorageLayout::ThreadScoped,
38 }
39 }
40
41 pub fn shared_legacy(root: impl Into<PathBuf>) -> SharedContextArtifactStore {
42 SharedContextArtifactStore::new(std::sync::Arc::new(Self::new(root)))
43 }
44
45 pub fn shared_thread_scoped(thread_root: impl Into<PathBuf>) -> SharedContextArtifactStore {
46 SharedContextArtifactStore::new(std::sync::Arc::new(Self::new_thread_scoped(thread_root)))
47 }
48
49 pub fn root(&self) -> &Path {
50 &self.root
51 }
52
53 pub fn create(&self, request: CreateArtifactRequest<'_>) -> anyhow::Result<ContextArtifact> {
54 let id = format!("artifact-{}", uuid::Uuid::new_v4());
55 let dir = self.turn_dir(request.thread_id, request.turn_id);
56 std::fs::create_dir_all(&dir)?;
57 let data_path = dir.join(format!("{id}.txt"));
58 std::fs::write(&data_path, request.bytes)?;
59 let artifact = ContextArtifact {
60 id: id.clone(),
61 kind: request.kind,
62 thread_id: request.thread_id.clone(),
63 turn_id: request.turn_id.clone(),
64 byte_count: request.bytes.len() as u64,
65 line_count: line_count_lossy(request.bytes) as u64,
66 source_tool_id: request.source_tool_id.map(ToOwned::to_owned),
67 label: request.label.map(ToOwned::to_owned),
68 store_path: data_path.display().to_string(),
69 retention_expires_at: None,
70 created_at: OffsetDateTime::now_utc(),
71 roder_owned: true,
72 };
73 self.write_metadata(&artifact)?;
74 Ok(artifact)
75 }
76
77 pub fn append(
78 &self,
79 thread_id: &ThreadId,
80 artifact_id: &ContextArtifactId,
81 bytes: &[u8],
82 ) -> anyhow::Result<ContextArtifact> {
83 let mut artifact = self.get_scoped(thread_id, artifact_id)?;
84 let path = PathBuf::from(&artifact.store_path);
85 ensure_under_root(&self.root, &path)?;
86 use std::io::Write;
87 let mut file = std::fs::OpenOptions::new().append(true).open(&path)?;
88 file.write_all(bytes)?;
89 let all_bytes = std::fs::read(&path)?;
90 artifact.byte_count = all_bytes.len() as u64;
91 artifact.line_count = line_count_lossy(&all_bytes) as u64;
92 self.write_metadata(&artifact)?;
93 Ok(artifact)
94 }
95
96 pub fn get_scoped(
97 &self,
98 thread_id: &ThreadId,
99 artifact_id: &ContextArtifactId,
100 ) -> anyhow::Result<ContextArtifact> {
101 let artifact = self.get(artifact_id)?;
102 if &artifact.thread_id != thread_id {
103 anyhow::bail!("artifact {artifact_id} does not belong to thread {thread_id}");
104 }
105 Ok(artifact)
106 }
107
108 pub fn get(&self, artifact_id: &ContextArtifactId) -> anyhow::Result<ContextArtifact> {
109 for metadata in self.metadata_paths()? {
110 if metadata.file_stem().and_then(|stem| stem.to_str()) != Some(artifact_id.as_str()) {
111 continue;
112 }
113 let text = std::fs::read_to_string(&metadata)?;
114 return Ok(serde_json::from_str(&text)?);
115 }
116 anyhow::bail!("unknown artifact {artifact_id}")
117 }
118
119 fn read_bytes_scoped(
120 &self,
121 thread_id: &ThreadId,
122 artifact_id: &ContextArtifactId,
123 ) -> anyhow::Result<(ContextArtifact, Vec<u8>)> {
124 let artifact = self.get_scoped(thread_id, artifact_id)?;
125 let path = PathBuf::from(&artifact.store_path);
126 ensure_under_root(&self.root, &path)?;
127 Ok((artifact, std::fs::read(path)?))
128 }
129
130 fn metadata_paths(&self) -> anyhow::Result<Vec<PathBuf>> {
131 let mut out = Vec::new();
132 if !self.root.exists() {
133 return Ok(out);
134 }
135 collect_metadata_paths(&self.root, &mut out)?;
136 Ok(out)
137 }
138
139 fn list_thread(&self, thread_id: &ThreadId) -> anyhow::Result<Vec<ContextArtifact>> {
140 let mut artifacts = Vec::new();
141 for metadata in self.metadata_paths()? {
142 let text = std::fs::read_to_string(metadata)?;
143 let artifact: ContextArtifact = serde_json::from_str(&text)?;
144 if &artifact.thread_id == thread_id {
145 artifacts.push(artifact);
146 }
147 }
148 artifacts.sort_by(|left, right| {
149 left.created_at
150 .cmp(&right.created_at)
151 .then(left.id.cmp(&right.id))
152 });
153 Ok(artifacts)
154 }
155
156 fn delete_scoped(
157 &self,
158 thread_id: &ThreadId,
159 artifact_id: &ContextArtifactId,
160 ) -> anyhow::Result<bool> {
161 let artifact = self.get_scoped(thread_id, artifact_id)?;
162 if !artifact.roder_owned {
163 anyhow::bail!(
164 "refusing to delete non-Roder-owned artifact {}",
165 artifact.id
166 );
167 }
168 let data_path = PathBuf::from(&artifact.store_path);
169 ensure_under_root(&self.root, &data_path)?;
170 let metadata_path = self.metadata_path(&artifact);
171 let mut deleted = false;
172 if data_path.exists() {
173 std::fs::remove_file(data_path)?;
174 deleted = true;
175 }
176 if metadata_path.exists() {
177 std::fs::remove_file(metadata_path)?;
178 deleted = true;
179 }
180 Ok(deleted)
181 }
182
183 fn write_metadata(&self, artifact: &ContextArtifact) -> anyhow::Result<()> {
184 let path = self.metadata_path(artifact);
185 std::fs::write(path, serde_json::to_string_pretty(artifact)?)?;
186 Ok(())
187 }
188
189 fn metadata_path(&self, artifact: &ContextArtifact) -> PathBuf {
190 self.turn_dir(&artifact.thread_id, &artifact.turn_id)
191 .join(format!("{}.json", artifact.id))
192 }
193
194 fn turn_dir(&self, thread_id: &ThreadId, turn_id: &TurnId) -> PathBuf {
195 let thread_dir = self.root.join(safe_component(thread_id));
196 match self.layout {
197 ArtifactStorageLayout::LegacyRoot => thread_dir.join(safe_component(turn_id)),
198 ArtifactStorageLayout::ThreadScoped => {
199 thread_dir.join("artifacts").join(safe_component(turn_id))
200 }
201 }
202 }
203}
204
205impl ContextArtifactAccess for ContextArtifactStore {
206 fn create_artifact(
207 &self,
208 request: CreateArtifactRequest<'_>,
209 ) -> anyhow::Result<ContextArtifact> {
210 self.create(request)
211 }
212
213 fn append_artifact(
214 &self,
215 thread_id: &ThreadId,
216 artifact_id: &ContextArtifactId,
217 bytes: &[u8],
218 ) -> anyhow::Result<ContextArtifact> {
219 self.append(thread_id, artifact_id, bytes)
220 }
221
222 fn list_artifacts(&self, thread_id: &ThreadId) -> anyhow::Result<Vec<ContextArtifact>> {
223 self.list_thread(thread_id)
224 }
225
226 fn read_artifact(
227 &self,
228 thread_id: &ThreadId,
229 artifact_id: &ContextArtifactId,
230 start_line: usize,
231 limit: usize,
232 ) -> anyhow::Result<ArtifactReadPage> {
233 let (artifact, bytes) = self.read_bytes_scoped(thread_id, artifact_id)?;
234 let lines = numbered_lines(&bytes);
235 let start_line = start_line.max(1);
236 let limit = clamp_limit(Some(limit));
237 let page = page_lines(&lines, start_line - 1, limit);
238 Ok(ArtifactReadPage {
239 artifact: artifact.descriptor(),
240 text: page.text,
241 start_line,
242 limit,
243 shown: page.shown,
244 total_lines: page.total,
245 next_start_line: page.next_offset.map(|offset| offset + 1),
246 truncated: page.next_offset.is_some(),
247 })
248 }
249
250 fn grep_artifact(
251 &self,
252 thread_id: &ThreadId,
253 artifact_id: &ContextArtifactId,
254 query: &str,
255 offset: usize,
256 limit: usize,
257 ) -> anyhow::Result<ArtifactGrepPage> {
258 if query.is_empty() {
259 anyhow::bail!("query is required");
260 }
261 let (artifact, bytes) = self.read_bytes_scoped(thread_id, artifact_id)?;
262 let matches = String::from_utf8_lossy(&bytes)
263 .lines()
264 .enumerate()
265 .filter(|(_, line)| line.contains(query))
266 .map(|(index, line)| format!("{}: {}", index + 1, line))
267 .collect::<Vec<_>>();
268 let limit = clamp_limit(Some(limit));
269 let page = page_lines(&matches, offset, limit);
270 Ok(ArtifactGrepPage {
271 artifact: artifact.descriptor(),
272 query: query.to_string(),
273 text: page.text,
274 offset,
275 limit,
276 shown: page.shown,
277 total_matches: page.total,
278 next_offset: page.next_offset,
279 truncated: page.next_offset.is_some(),
280 })
281 }
282
283 fn tail_artifact(
284 &self,
285 thread_id: &ThreadId,
286 artifact_id: &ContextArtifactId,
287 lines: usize,
288 ) -> anyhow::Result<ArtifactTailPage> {
289 let (artifact, bytes) = self.read_bytes_scoped(thread_id, artifact_id)?;
290 let all_lines = numbered_lines(&bytes);
291 let lines = clamp_limit(Some(lines));
292 let total = all_lines.len();
293 let start = total.saturating_sub(lines);
294 let page = page_lines(&all_lines, start, lines);
295 Ok(ArtifactTailPage {
296 artifact: artifact.descriptor(),
297 text: page.text,
298 start_line: start + 1,
299 lines,
300 shown: page.shown,
301 total_lines: page.total,
302 truncated: start > 0,
303 })
304 }
305
306 fn delete_artifact(
307 &self,
308 thread_id: &ThreadId,
309 artifact_id: &ContextArtifactId,
310 ) -> anyhow::Result<bool> {
311 self.delete_scoped(thread_id, artifact_id)
312 }
313}
314
315pub fn default_context_artifact_dir() -> PathBuf {
316 std::env::var_os("RODER_DATA_DIR")
317 .map(PathBuf::from)
318 .or_else(|| std::env::var_os("HOME").map(|home| PathBuf::from(home).join(".roder")))
319 .unwrap_or_else(|| PathBuf::from(".roder"))
320 .join("context-artifacts")
321}
322
323fn safe_component(value: &str) -> String {
324 value
325 .chars()
326 .map(|ch| {
327 if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') {
328 ch
329 } else {
330 '_'
331 }
332 })
333 .collect()
334}
335
336fn collect_metadata_paths(dir: &Path, out: &mut Vec<PathBuf>) -> anyhow::Result<()> {
337 if !dir.exists() {
338 return Ok(());
339 }
340 for entry in std::fs::read_dir(dir)? {
341 let path = entry?.path();
342 if path.is_dir() {
343 collect_metadata_paths(&path, out)?;
344 } else if path.extension().and_then(|ext| ext.to_str()) == Some("json")
345 && path
346 .file_stem()
347 .and_then(|name| name.to_str())
348 .is_some_and(|name| name.starts_with("artifact-"))
349 {
350 out.push(path);
351 }
352 }
353 Ok(())
354}
355
356fn line_count_lossy(bytes: &[u8]) -> usize {
357 let text = String::from_utf8_lossy(bytes);
358 if text.is_empty() {
359 0
360 } else {
361 text.lines().count()
362 }
363}
364
365fn numbered_lines(bytes: &[u8]) -> Vec<String> {
366 String::from_utf8_lossy(bytes)
367 .lines()
368 .enumerate()
369 .map(|(index, line)| format!("{:>5}: {}", index + 1, line))
370 .collect()
371}
372
373#[derive(Debug, Clone)]
374struct LinePage {
375 text: String,
376 shown: usize,
377 total: usize,
378 next_offset: Option<usize>,
379}
380
381fn clamp_limit(limit: Option<usize>) -> usize {
382 limit.unwrap_or(DEFAULT_PAGE_LINES).clamp(1, MAX_PAGE_LINES)
383}
384
385fn page_lines(lines: &[String], offset: usize, limit: usize) -> LinePage {
386 let total = lines.len();
387 let offset = offset.min(total);
388 let end = offset.saturating_add(limit).min(total);
389 let next_offset = (end < total).then_some(end);
390 let mut text = lines[offset..end].join("\n");
391 if let Some(next) = next_offset {
392 if !text.is_empty() {
393 text.push('\n');
394 }
395 text.push_str(&format!(
396 "[showing lines {}-{} of {total}; next_offset={next}]",
397 offset + 1,
398 end
399 ));
400 }
401 LinePage {
402 text,
403 shown: end.saturating_sub(offset),
404 total,
405 next_offset,
406 }
407}
408
409fn ensure_under_root(root: &Path, path: &Path) -> anyhow::Result<()> {
410 let root = root.canonicalize()?;
411 let path = path.canonicalize()?;
412 if !path.starts_with(root) {
413 anyhow::bail!("artifact path escapes artifact root");
414 }
415 Ok(())
416}
417
418#[cfg(test)]
419mod tests {
420 use super::*;
421 use roder_api::artifacts::ContextArtifactKind;
422
423 #[test]
424 fn artifact_store_writes_thread_scoped_reads_greps_tails_and_deletes_by_thread() {
425 let root =
426 std::env::temp_dir().join(format!("roder-context-artifacts-{}", uuid::Uuid::new_v4()));
427 let store = ContextArtifactStore::new_thread_scoped(&root);
428 let artifact = store
429 .create(CreateArtifactRequest {
430 kind: ContextArtifactKind::ToolOutput,
431 thread_id: &"thread-a".to_string(),
432 turn_id: &"turn-a".to_string(),
433 source_tool_id: Some("call-a"),
434 label: Some("stdout"),
435 bytes: b"alpha\nneedle\nomega\n",
436 })
437 .unwrap();
438
439 assert!(
440 artifact
441 .store_path
442 .starts_with(root.to_string_lossy().as_ref())
443 );
444 let store_path = Path::new(&artifact.store_path);
445 assert!(store_path.starts_with(root.join("thread-a").join("artifacts").join("turn-a")));
446 assert!(
447 store_path
448 .file_name()
449 .and_then(|name| name.to_str())
450 .is_some_and(|name| name.starts_with("artifact-"))
451 );
452 assert_eq!(artifact.line_count, 3);
453 assert_eq!(
454 store.list_artifacts(&"thread-a".to_string()).unwrap().len(),
455 1
456 );
457
458 let read = store
459 .read_artifact(&"thread-a".to_string(), &artifact.id, 2, 1)
460 .unwrap();
461 assert!(read.text.contains("2: needle"));
462 assert_eq!(read.next_start_line, Some(3));
463
464 let grep = store
465 .grep_artifact(&"thread-a".to_string(), &artifact.id, "needle", 0, 10)
466 .unwrap();
467 assert_eq!(grep.text, "2: needle");
468
469 let tail = store
470 .tail_artifact(&"thread-a".to_string(), &artifact.id, 2)
471 .unwrap();
472 assert!(tail.text.contains("2: needle"));
473 assert!(tail.text.contains("3: omega"));
474
475 let wrong_thread = store
476 .read_artifact(&"thread-b".to_string(), &artifact.id, 1, 1)
477 .unwrap_err()
478 .to_string();
479 assert!(wrong_thread.contains("does not belong to thread"));
480
481 assert!(
482 store
483 .delete_artifact(&"thread-a".to_string(), &artifact.id)
484 .unwrap()
485 );
486 assert!(
487 store
488 .list_artifacts(&"thread-a".to_string())
489 .unwrap()
490 .is_empty()
491 );
492 let _ = std::fs::remove_dir_all(root);
493 }
494
495 #[test]
496 fn artifact_store_appends_and_recounts_lossy_utf8() {
497 let root =
498 std::env::temp_dir().join(format!("roder-context-artifacts-{}", uuid::Uuid::new_v4()));
499 let store = ContextArtifactStore::new(&root);
500 let artifact = store
501 .create(CreateArtifactRequest {
502 kind: ContextArtifactKind::CommandStderr,
503 thread_id: &"thread-a".to_string(),
504 turn_id: &"turn-a".to_string(),
505 source_tool_id: None,
506 label: Some("stderr"),
507 bytes: b"one\n",
508 })
509 .unwrap();
510
511 let artifact = store
512 .append(&"thread-a".to_string(), &artifact.id, b"\xfftwo\n")
513 .unwrap();
514
515 assert_eq!(artifact.line_count, 2);
516 let read = store
517 .read_artifact(&"thread-a".to_string(), &artifact.id, 1, 10)
518 .unwrap();
519 assert!(read.text.contains("two"));
520 let _ = std::fs::remove_dir_all(root);
521 }
522}