roder_api/
thread.rs

1use std::path::PathBuf;
2use std::sync::Arc;
3
4use serde::{Deserialize, Deserializer, Serialize};
5use time::OffsetDateTime;
6
7use crate::artifacts::ContextArtifactStore;
8use crate::events::{EventEnvelope, ThreadId, TurnId};
9pub use crate::extension::{CheckpointStoreId, ThreadStoreId};
10use crate::extension_state::ExtensionStateRecord;
11use crate::inference::{TokenUsage, cache_hit_rate};
12use crate::inference_routing::ModelSelectionMode;
13use crate::remote_runner::{RunnerDestination, RunnerSessionState, ThreadRunnerBinding};
14use crate::transcript::{InputImage, TranscriptItem};
15
16mod projection;
17pub use projection::{project_thread_item_events, project_turns_from_events};
18
19#[derive(Debug, Clone, Default, PartialEq, Eq)]
20pub struct ThreadListOptions {
21    pub limit: Option<usize>,
22    pub cursor: Option<String>,
23}
24
25#[derive(Debug, Clone, Default)]
26pub struct ThreadListPage {
27    pub threads: Vec<ThreadMetadata>,
28    pub next_cursor: Option<String>,
29    pub backwards_cursor: Option<String>,
30}
31
32/// Thread IDs reserved for system event streams that are not user-visible conversations.
33pub const SYNTHETIC_EVENT_THREAD_IDS: &[&str] = &["app-server", "runtime", "thread-workflow"];
34
35/// Returns true for production-emitted synthetic event streams.
36pub fn is_synthetic_event_thread_id(thread_id: &str) -> bool {
37    SYNTHETIC_EVENT_THREAD_IDS.contains(&thread_id)
38}
39
40#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq)]
41pub struct ThreadUsageMetadata {
42    #[serde(default)]
43    pub prompt_tokens: u64,
44    #[serde(default)]
45    pub completion_tokens: u64,
46    #[serde(default)]
47    pub total_tokens: u64,
48    #[serde(default)]
49    pub cached_prompt_tokens: u64,
50    /// Subset of `prompt_tokens` written to the provider prompt cache.
51    #[serde(default)]
52    pub cache_creation_prompt_tokens: u64,
53    #[serde(default, skip_serializing_if = "Option::is_none")]
54    pub cache_hit_rate: Option<f64>,
55}
56
57impl ThreadUsageMetadata {
58    pub fn add_token_usage(&mut self, usage: &TokenUsage) {
59        self.prompt_tokens = self
60            .prompt_tokens
61            .saturating_add(u64::from(usage.prompt_tokens));
62        self.completion_tokens = self
63            .completion_tokens
64            .saturating_add(u64::from(usage.completion_tokens));
65        self.total_tokens = self
66            .total_tokens
67            .saturating_add(u64::from(usage.total_tokens));
68        self.cached_prompt_tokens = self
69            .cached_prompt_tokens
70            .saturating_add(u64::from(usage.cached_prompt_tokens));
71        self.cache_creation_prompt_tokens = self
72            .cache_creation_prompt_tokens
73            .saturating_add(u64::from(usage.cache_creation_prompt_tokens));
74        self.cache_hit_rate = if self.prompt_tokens == 0 {
75            None
76        } else if self.prompt_tokens > u64::from(u32::MAX) {
77            Some(
78                (self.cached_prompt_tokens.min(self.prompt_tokens) as f64)
79                    / (self.prompt_tokens as f64),
80            )
81        } else {
82            cache_hit_rate(self.prompt_tokens as u32, self.cached_prompt_tokens as u32)
83        };
84    }
85
86    pub fn is_empty(&self) -> bool {
87        self.prompt_tokens == 0
88            && self.completion_tokens == 0
89            && self.total_tokens == 0
90            && self.cached_prompt_tokens == 0
91            && self.cache_creation_prompt_tokens == 0
92    }
93}
94
95#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
96pub struct ThreadMetadata {
97    pub thread_id: ThreadId,
98    pub title: Option<String>,
99    #[serde(deserialize_with = "deserialize_thread_workspace")]
100    pub workspace: String,
101    #[serde(default, skip_serializing_if = "Option::is_none")]
102    pub workspace_id: Option<String>,
103    #[serde(default, skip_serializing_if = "Option::is_none")]
104    pub root_id: Option<String>,
105    pub provider: Option<String>,
106    pub model: Option<String>,
107    #[serde(default, skip_serializing_if = "Option::is_none")]
108    pub selection_mode: Option<ModelSelectionMode>,
109    /// Per-thread tool filter applied on top of the runtime allowlist. Empty = no filtering.
110    #[serde(default, skip_serializing_if = "Vec::is_empty")]
111    pub tool_allowlist: Vec<String>,
112    /// Host-supplied instructions added to the developer slot of every turn's inference request.
113    #[serde(default, skip_serializing_if = "Option::is_none")]
114    pub developer_instructions: Option<String>,
115    /**
116     * Host-executed tool specs advertised to the model on every turn of this thread. Calls to
117     * these tools pause on a `thread/toolExecutionRequested` notification until the host client
118     * answers with `tools/resolve`.
119     */
120    #[serde(default, skip_serializing_if = "Vec::is_empty")]
121    pub external_tools: Vec<crate::tools::ToolSpec>,
122    #[serde(default, skip_serializing_if = "Option::is_none")]
123    pub runner_destination: Option<RunnerDestination>,
124    #[serde(default, skip_serializing_if = "Option::is_none")]
125    pub runner_state: Option<RunnerSessionState>,
126    /**
127     * Set when the thread explicitly selected a remote runner at creation.
128     * Native coding tools for the thread route through this runner workspace;
129     * absent = local tool execution even when a runtime-level runner
130     * destination is configured.
131     */
132    #[serde(default, skip_serializing_if = "Option::is_none")]
133    pub runner_binding: Option<ThreadRunnerBinding>,
134    #[serde(with = "time::serde::rfc3339")]
135    pub created_at: OffsetDateTime,
136    #[serde(with = "time::serde::rfc3339")]
137    pub updated_at: OffsetDateTime,
138    pub message_count: u32,
139    #[serde(default, skip_serializing_if = "Option::is_none")]
140    pub usage: Option<ThreadUsageMetadata>,
141    /// Parent thread this conversation was forked from. Absent for normal threads.
142    #[serde(default, skip_serializing_if = "Option::is_none")]
143    pub parent_thread_id: Option<ThreadId>,
144    /// Parent turn the fork branched at, when the fork targeted a specific turn.
145    #[serde(default, skip_serializing_if = "Option::is_none")]
146    pub forked_from_turn_id: Option<TurnId>,
147    /**
148     * Provider-neutral workspace-fork provenance for threads whose
149     * workspace is a fork of another workspace (roadmap phase 81; the
150     * phase-90 Git-worktree MVP shape was folded into this canonical type).
151     */
152    #[serde(default, skip_serializing_if = "Option::is_none")]
153    pub workspace_fork: Option<crate::forks::WorkspaceFork>,
154}
155
156pub fn validate_thread_workspace(workspace: &str) -> anyhow::Result<String> {
157    let workspace = workspace.trim();
158    anyhow::ensure!(!workspace.is_empty(), "thread workspace is required");
159    anyhow::ensure!(
160        std::path::Path::new(workspace).is_absolute(),
161        "thread workspace must be an absolute path: {workspace}"
162    );
163    Ok(workspace.to_string())
164}
165
166fn deserialize_thread_workspace<'de, D>(deserializer: D) -> Result<String, D::Error>
167where
168    D: Deserializer<'de>,
169{
170    let workspace = String::deserialize(deserializer)?;
171    validate_thread_workspace(&workspace).map_err(serde::de::Error::custom)
172}
173
174#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
175pub struct TurnRecord {
176    pub thread_id: ThreadId,
177    pub turn_id: TurnId,
178    pub items: Vec<TranscriptItem>,
179    #[serde(with = "time::serde::rfc3339")]
180    pub created_at: OffsetDateTime,
181    #[serde(with = "time::serde::rfc3339::option")]
182    pub completed_at: Option<OffsetDateTime>,
183    #[serde(default, skip_serializing_if = "Option::is_none")]
184    pub usage: Option<TokenUsage>,
185    /// Normalized stop reason from `TurnCompleted`; `None` for failed or interrupted turns.
186    #[serde(default, skip_serializing_if = "Option::is_none")]
187    pub finish_reason: Option<String>,
188}
189
190#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
191#[serde(rename_all = "camelCase")]
192pub enum ThreadItemStatus {
193    InProgress,
194    Completed,
195    Failed,
196}
197
198#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
199#[serde(tag = "type", rename_all = "camelCase")]
200pub enum ThreadItem {
201    UserMessage {
202        id: String,
203        text: String,
204        #[serde(default, skip_serializing_if = "Vec::is_empty")]
205        images: Vec<InputImage>,
206        #[serde(default, skip_serializing_if = "Option::is_none")]
207        status: Option<ThreadItemStatus>,
208    },
209    AgentMessage {
210        id: String,
211        text: String,
212        #[serde(default, skip_serializing_if = "Option::is_none")]
213        phase: Option<String>,
214        #[serde(default, skip_serializing_if = "Option::is_none")]
215        status: Option<ThreadItemStatus>,
216    },
217    Reasoning {
218        id: String,
219        #[serde(default, skip_serializing_if = "Vec::is_empty")]
220        summary: Vec<String>,
221        #[serde(default, skip_serializing_if = "Vec::is_empty")]
222        content: Vec<String>,
223        #[serde(default, skip_serializing_if = "Option::is_none")]
224        status: Option<ThreadItemStatus>,
225    },
226    ToolExecution {
227        id: String,
228        #[serde(rename = "toolCallId")]
229        tool_call_id: String,
230        #[serde(rename = "toolName")]
231        tool_name: String,
232        status: ThreadItemStatus,
233        #[serde(default, skip_serializing_if = "Option::is_none")]
234        input: Option<serde_json::Value>,
235        #[serde(default, skip_serializing_if = "Option::is_none")]
236        output: Option<String>,
237        #[serde(default, skip_serializing_if = "Option::is_none")]
238        error: Option<String>,
239    },
240    RoutingDecision {
241        id: String,
242        decision: crate::events::InferenceRoutingDecisionEvent,
243        #[serde(default, skip_serializing_if = "Option::is_none")]
244        status: Option<ThreadItemStatus>,
245    },
246    Compaction {
247        id: String,
248        summary: String,
249        #[serde(default, skip_serializing_if = "Option::is_none")]
250        status: Option<ThreadItemStatus>,
251    },
252    Error {
253        id: String,
254        message: String,
255        #[serde(default, skip_serializing_if = "Option::is_none")]
256        status: Option<ThreadItemStatus>,
257    },
258    Raw {
259        id: String,
260        payload: serde_json::Value,
261        #[serde(default, skip_serializing_if = "Option::is_none")]
262        status: Option<ThreadItemStatus>,
263    },
264}
265
266#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
267pub struct ThreadItemTurnRecord {
268    pub thread_id: ThreadId,
269    pub turn_id: TurnId,
270    #[serde(with = "time::serde::rfc3339")]
271    pub created_at: OffsetDateTime,
272    pub items: Vec<ThreadItem>,
273}
274
275#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
276#[serde(tag = "type", rename_all = "camelCase")]
277pub enum ThreadItemDelta {
278    AgentMessageText {
279        delta: String,
280        #[serde(default, skip_serializing_if = "Option::is_none")]
281        phase: Option<String>,
282    },
283    ReasoningText {
284        delta: String,
285        #[serde(rename = "contentIndex")]
286        content_index: usize,
287    },
288    ReasoningSummaryPartAdded {
289        #[serde(rename = "summaryIndex")]
290        summary_index: usize,
291    },
292    ReasoningSummaryText {
293        delta: String,
294        #[serde(rename = "summaryIndex")]
295        summary_index: usize,
296    },
297}
298
299#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
300#[serde(tag = "type", rename_all = "camelCase")]
301pub enum ThreadItemEventKind {
302    ItemStarted {
303        item: ThreadItem,
304    },
305    ItemDelta {
306        #[serde(rename = "itemId")]
307        item_id: String,
308        delta: ThreadItemDelta,
309    },
310    ItemCompleted {
311        item: ThreadItem,
312    },
313}
314
315#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
316#[serde(rename_all = "camelCase")]
317pub struct ThreadItemEvent {
318    pub seq: u64,
319    #[serde(rename = "eventId")]
320    pub event_id: String,
321    #[serde(rename = "threadId")]
322    pub thread_id: ThreadId,
323    #[serde(rename = "turnId")]
324    pub turn_id: TurnId,
325    #[serde(with = "time::serde::rfc3339")]
326    pub timestamp: OffsetDateTime,
327    pub event: ThreadItemEventKind,
328}
329
330#[derive(Debug, Clone, Default, Serialize, Deserialize)]
331pub struct ThreadSnapshot {
332    pub metadata: Option<ThreadMetadata>,
333    pub events: Vec<EventEnvelope>,
334    pub turns: Vec<TurnRecord>,
335    #[serde(default)]
336    pub item_events: Vec<ThreadItemEvent>,
337    pub extension_states: Vec<ExtensionStateRecord>,
338}
339
340impl ThreadItem {
341    pub fn id(&self) -> &str {
342        match self {
343            ThreadItem::UserMessage { id, .. }
344            | ThreadItem::AgentMessage { id, .. }
345            | ThreadItem::Reasoning { id, .. }
346            | ThreadItem::ToolExecution { id, .. }
347            | ThreadItem::RoutingDecision { id, .. }
348            | ThreadItem::Compaction { id, .. }
349            | ThreadItem::Error { id, .. }
350            | ThreadItem::Raw { id, .. } => id,
351        }
352    }
353}
354
355#[async_trait::async_trait]
356pub trait ThreadStore: Send + Sync {
357    fn id(&self) -> ThreadStoreId;
358
359    fn local_thread_root(&self) -> Option<PathBuf> {
360        None
361    }
362
363    fn context_artifact_store(&self) -> Option<ContextArtifactStore> {
364        None
365    }
366
367    async fn create_thread(&self, metadata: ThreadMetadata) -> anyhow::Result<ThreadMetadata>;
368    async fn update_thread_metadata(
369        &self,
370        metadata: ThreadMetadata,
371    ) -> anyhow::Result<ThreadMetadata> {
372        Ok(metadata)
373    }
374    async fn list_threads(&self) -> anyhow::Result<Vec<ThreadMetadata>>;
375    async fn list_threads_page(
376        &self,
377        options: ThreadListOptions,
378    ) -> anyhow::Result<ThreadListPage> {
379        let mut threads = self.list_threads().await?;
380        threads.sort_by_key(|thread| std::cmp::Reverse(thread.updated_at));
381        let offset = options
382            .cursor
383            .as_deref()
384            .and_then(|cursor| cursor.parse::<usize>().ok())
385            .unwrap_or(0)
386            .min(threads.len());
387        let limit = options
388            .limit
389            .unwrap_or(threads.len().saturating_sub(offset));
390        let next_offset = offset.saturating_add(limit).min(threads.len());
391        let total = threads.len();
392        let page_threads = threads
393            .into_iter()
394            .skip(offset)
395            .take(limit)
396            .collect::<Vec<_>>();
397        Ok(ThreadListPage {
398            threads: page_threads,
399            next_cursor: (next_offset < total).then(|| next_offset.to_string()),
400            backwards_cursor: (offset > 0).then(|| offset.saturating_sub(limit).to_string()),
401        })
402    }
403    async fn load_thread_metadata(
404        &self,
405        thread_id: &ThreadId,
406    ) -> anyhow::Result<Option<ThreadMetadata>> {
407        Ok(self
408            .load_thread(thread_id)
409            .await?
410            .and_then(|snapshot| snapshot.metadata))
411    }
412    async fn load_thread(&self, thread_id: &ThreadId) -> anyhow::Result<Option<ThreadSnapshot>>;
413    async fn archive_thread(&self, thread_id: &ThreadId) -> anyhow::Result<bool> {
414        let _ = thread_id;
415        anyhow::bail!("thread store {} does not support archive", self.id())
416    }
417    async fn append_event(
418        &self,
419        thread_id: &ThreadId,
420        envelope: &EventEnvelope,
421    ) -> anyhow::Result<()>;
422    async fn append_item_event(
423        &self,
424        thread_id: &ThreadId,
425        item_event: &ThreadItemEvent,
426    ) -> anyhow::Result<()> {
427        let _ = (thread_id, item_event);
428        Ok(())
429    }
430    async fn append_extension_state(
431        &self,
432        thread_id: &ThreadId,
433        record: &ExtensionStateRecord,
434    ) -> anyhow::Result<()> {
435        let _ = (thread_id, record);
436        anyhow::bail!(
437            "thread store {} does not support extension state",
438            self.id()
439        )
440    }
441}
442
443pub trait ThreadStoreFactory: Send + Sync + 'static {
444    fn id(&self) -> ThreadStoreId;
445    fn create(&self) -> Arc<dyn ThreadStore>;
446}
447
448#[async_trait::async_trait]
449pub trait CheckpointStore: Send + Sync {
450    fn id(&self) -> CheckpointStoreId;
451    async fn save_snapshot(&self, snapshot: ThreadSnapshot) -> anyhow::Result<()>;
452    async fn load_snapshot(&self, thread_id: &ThreadId) -> anyhow::Result<Option<ThreadSnapshot>>;
453}
454
455pub trait CheckpointStoreFactory: Send + Sync + 'static {
456    fn id(&self) -> CheckpointStoreId;
457    fn create(&self) -> Arc<dyn CheckpointStore>;
458}
459
460#[cfg(test)]
461mod tests {
462    use super::*;
463    use crate::inference::ModelSelection;
464
465    #[test]
466    fn synthetic_event_thread_ids_are_reserved_production_ids() {
467        assert!(is_synthetic_event_thread_id("app-server"));
468        assert!(is_synthetic_event_thread_id("runtime"));
469        assert!(is_synthetic_event_thread_id("thread-workflow"));
470
471        assert!(!is_synthetic_event_thread_id("thread-discovery"));
472        assert!(!is_synthetic_event_thread_id("thread-plan"));
473        assert!(!is_synthetic_event_thread_id("thread-process"));
474        assert!(!is_synthetic_event_thread_id("thread-1"));
475    }
476
477    #[test]
478    fn thread_fork_metadata_is_additive_for_legacy_records() {
479        // A legacy thread record without any fork fields still deserializes.
480        let legacy = serde_json::json!({
481            "thread_id": "thread-old",
482            "title": null,
483            "workspace": "/workspace",
484            "provider": null,
485            "model": null,
486            "created_at": "1970-01-01T00:00:00Z",
487            "updated_at": "1970-01-01T00:00:00Z",
488            "message_count": 3
489        });
490        let metadata: ThreadMetadata = serde_json::from_value(legacy).unwrap();
491        assert!(metadata.parent_thread_id.is_none());
492        assert!(metadata.forked_from_turn_id.is_none());
493        assert!(metadata.workspace_fork.is_none());
494
495        // Non-forked threads serialize without any fork keys.
496        let value = serde_json::to_value(&metadata).unwrap();
497        assert!(value.get("parentThreadId").is_none());
498        assert!(value.get("parent_thread_id").is_none());
499        assert!(value.get("workspace_fork").is_none());
500    }
501
502    #[test]
503    fn thread_fork_metadata_round_trips_workspace_fork_provenance() {
504        let fork = crate::forks::WorkspaceFork {
505            id: "/repo/.roder/worktrees/parser-experiment".to_string(),
506            provider_id: "git-worktree".to_string(),
507            source_workspace: std::path::PathBuf::from("/repo"),
508            workspace: std::path::PathBuf::from("/repo/.roder/worktrees/parser-experiment"),
509            status: crate::forks::ForkStatus::Active,
510            provenance: crate::forks::ForkProvenance {
511                branch: Some("roder/fork/parser-experiment".to_string()),
512                source_branch: Some("main".to_string()),
513                source_commit: Some("abc123".to_string()),
514                snapshot_id: None,
515                session_id: None,
516                created_at: OffsetDateTime::UNIX_EPOCH,
517            },
518            cleanup: crate::forks::ForkCleanupPolicy::Explicit,
519            metadata: serde_json::json!({}),
520        };
521        let value = serde_json::to_value(&fork).unwrap();
522        assert_eq!(value["providerId"], "git-worktree");
523        assert_eq!(value["status"], "active");
524        assert_eq!(value["cleanup"], "explicit");
525        assert_eq!(value["provenance"]["sourceCommit"], "abc123");
526
527        let round_trip: crate::forks::WorkspaceFork = serde_json::from_value(value).unwrap();
528        assert_eq!(round_trip, fork);
529
530        // A detached-HEAD fork keeps clear provenance without a branch name.
531        let detached = crate::forks::WorkspaceFork {
532            provenance: crate::forks::ForkProvenance {
533                source_branch: None,
534                ..fork.provenance.clone()
535            },
536            ..fork
537        };
538        let value = serde_json::to_value(&detached).unwrap();
539        assert!(value["provenance"].get("sourceBranch").is_none());
540    }
541
542    #[test]
543    fn thread_metadata_timestamps_serialize_as_rfc3339_strings() {
544        let value = serde_json::to_value(ThreadMetadata {
545            thread_id: "thread-a".to_string(),
546            title: None,
547            workspace: "/workspace".to_string(),
548            workspace_id: None,
549            root_id: None,
550            provider: None,
551            model: None,
552            selection_mode: None,
553            tool_allowlist: Vec::new(),
554            developer_instructions: None,
555            external_tools: Vec::new(),
556            runner_destination: None,
557            runner_state: None,
558            runner_binding: None,
559            parent_thread_id: None,
560            forked_from_turn_id: None,
561            workspace_fork: None,
562            created_at: OffsetDateTime::UNIX_EPOCH,
563            updated_at: OffsetDateTime::UNIX_EPOCH,
564            message_count: 0,
565            usage: None,
566        })
567        .unwrap();
568
569        assert_eq!(value["created_at"], "1970-01-01T00:00:00Z");
570        assert_eq!(value["updated_at"], "1970-01-01T00:00:00Z");
571        assert_eq!(value["workspace"], "/workspace");
572    }
573
574    #[test]
575    fn thread_metadata_deserializes_without_selection_mode() {
576        let value = serde_json::json!({
577            "thread_id": "thread-a",
578            "title": null,
579            "workspace": "/workspace",
580            "provider": "codex",
581            "model": "gpt-5.5",
582            "created_at": "1970-01-01T00:00:00Z",
583            "updated_at": "1970-01-01T00:00:00Z",
584            "message_count": 0
585        });
586
587        let metadata = serde_json::from_value::<ThreadMetadata>(value).unwrap();
588
589        assert_eq!(metadata.provider.as_deref(), Some("codex"));
590        assert_eq!(metadata.model.as_deref(), Some("gpt-5.5"));
591        assert_eq!(metadata.selection_mode, None);
592    }
593
594    #[test]
595    fn thread_metadata_round_trips_auto_selection_mode() {
596        let metadata = ThreadMetadata {
597            thread_id: "thread-a".to_string(),
598            title: None,
599            workspace: "/workspace".to_string(),
600            workspace_id: None,
601            root_id: None,
602            provider: Some("codex".to_string()),
603            model: Some("gpt-5.5".to_string()),
604            selection_mode: Some(ModelSelectionMode::auto(
605                "local-router:coding",
606                "local-router",
607                "Auto: Coding",
608                ModelSelection {
609                    provider: "codex".to_string(),
610                    model: "gpt-5.5".to_string(),
611                },
612                Some("coding".to_string()),
613                Some("low".to_string()),
614            )),
615            tool_allowlist: Vec::new(),
616            developer_instructions: None,
617            external_tools: Vec::new(),
618            runner_destination: None,
619            runner_state: None,
620            runner_binding: None,
621            parent_thread_id: None,
622            forked_from_turn_id: None,
623            workspace_fork: None,
624            created_at: OffsetDateTime::UNIX_EPOCH,
625            updated_at: OffsetDateTime::UNIX_EPOCH,
626            message_count: 0,
627            usage: None,
628        };
629
630        let value = serde_json::to_value(&metadata).unwrap();
631        let round_trip = serde_json::from_value::<ThreadMetadata>(value).unwrap();
632
633        assert_eq!(round_trip, metadata);
634    }
635
636    #[test]
637    fn thread_metadata_requires_workspace_when_deserializing() {
638        let value = serde_json::json!({
639            "thread_id": "thread-a",
640            "title": null,
641            "provider": null,
642            "model": null,
643            "created_at": "1970-01-01T00:00:00Z",
644            "updated_at": "1970-01-01T00:00:00Z",
645            "message_count": 0
646        });
647
648        let result = serde_json::from_value::<ThreadMetadata>(value);
649
650        assert!(result.is_err());
651    }
652
653    #[test]
654    fn thread_metadata_rejects_blank_or_relative_workspace_when_deserializing() {
655        for workspace in ["", "project"] {
656            let value = serde_json::json!({
657                "thread_id": "thread-a",
658                "title": null,
659                "workspace": workspace,
660                "provider": null,
661                "model": null,
662                "created_at": "1970-01-01T00:00:00Z",
663                "updated_at": "1970-01-01T00:00:00Z",
664                "message_count": 0
665            });
666
667            let result = serde_json::from_value::<ThreadMetadata>(value);
668
669            assert!(result.is_err(), "workspace {workspace:?} should fail");
670        }
671    }
672
673    #[test]
674    fn thread_usage_metadata_accumulates_cache_hit_rate() {
675        let mut usage = ThreadUsageMetadata::default();
676
677        usage.add_token_usage(
678            &TokenUsage::new(100, 10, 110)
679                .with_cached_prompt_tokens(92)
680                .with_cache_creation_prompt_tokens(5),
681        );
682        usage.add_token_usage(
683            &TokenUsage::new(50, 5, 55)
684                .with_cached_prompt_tokens(43)
685                .with_cache_creation_prompt_tokens(3),
686        );
687
688        assert_eq!(usage.prompt_tokens, 150);
689        assert_eq!(usage.cached_prompt_tokens, 135);
690        assert_eq!(usage.cache_creation_prompt_tokens, 8);
691        assert!((usage.cache_hit_rate.unwrap() - 0.9).abs() < f64::EPSILON);
692    }
693
694    #[test]
695    fn thread_item_events_replay_reasoning_and_final_answer_into_stable_items() {
696        let timestamp = OffsetDateTime::UNIX_EPOCH;
697        let events = vec![
698            ThreadItemEvent {
699                seq: 1,
700                event_id: "event-1".to_string(),
701                thread_id: "thread-1".to_string(),
702                turn_id: "turn-1".to_string(),
703                timestamp,
704                event: ThreadItemEventKind::ItemStarted {
705                    item: ThreadItem::Reasoning {
706                        id: "turn-1-agent-reasoning".to_string(),
707                        summary: Vec::new(),
708                        content: vec![String::new()],
709                        status: Some(ThreadItemStatus::InProgress),
710                    },
711                },
712            },
713            ThreadItemEvent {
714                seq: 2,
715                event_id: "event-2".to_string(),
716                thread_id: "thread-1".to_string(),
717                turn_id: "turn-1".to_string(),
718                timestamp,
719                event: ThreadItemEventKind::ItemDelta {
720                    item_id: "turn-1-agent-reasoning".to_string(),
721                    delta: ThreadItemDelta::ReasoningText {
722                        delta: "Inspecting".to_string(),
723                        content_index: 0,
724                    },
725                },
726            },
727            ThreadItemEvent {
728                seq: 3,
729                event_id: "event-3".to_string(),
730                thread_id: "thread-1".to_string(),
731                turn_id: "turn-1".to_string(),
732                timestamp,
733                event: ThreadItemEventKind::ItemDelta {
734                    item_id: "turn-1-agent-final_answer".to_string(),
735                    delta: ThreadItemDelta::AgentMessageText {
736                        delta: "Done".to_string(),
737                        phase: Some("final_answer".to_string()),
738                    },
739                },
740            },
741            ThreadItemEvent {
742                seq: 4,
743                event_id: "event-4".to_string(),
744                thread_id: "thread-1".to_string(),
745                turn_id: "turn-1".to_string(),
746                timestamp,
747                event: ThreadItemEventKind::ItemCompleted {
748                    item: ThreadItem::AgentMessage {
749                        id: "turn-1-agent-final_answer".to_string(),
750                        text: "Done.".to_string(),
751                        phase: Some("final_answer".to_string()),
752                        status: Some(ThreadItemStatus::Completed),
753                    },
754                },
755            },
756        ];
757
758        let turns = project_thread_item_events(&events);
759
760        assert_eq!(turns.len(), 1);
761        assert_eq!(turns[0].turn_id, "turn-1");
762        assert_eq!(
763            turns[0].items,
764            vec![
765                ThreadItem::Reasoning {
766                    id: "turn-1-agent-reasoning".to_string(),
767                    summary: Vec::new(),
768                    content: vec!["Inspecting".to_string()],
769                    status: Some(ThreadItemStatus::InProgress),
770                },
771                ThreadItem::AgentMessage {
772                    id: "turn-1-agent-final_answer".to_string(),
773                    text: "Done.".to_string(),
774                    phase: Some("final_answer".to_string()),
775                    status: Some(ThreadItemStatus::Completed),
776                }
777            ]
778        );
779    }
780}
roder_api/thread.rs

roder_api/
thread.rs
