roder_core/runtime/
thread_unload.rs

1use super::*;
2
3impl Runtime {
4    /// Cancel and drain one thread, then release its live caches and runner.
5    /// Persisted conversation data remains available for subsequent turns.
6    /// Callers should apply a deadline; failed cleanup retains resources for retry.
7    pub async fn unload_thread(&self, thread_id: &ThreadId) -> anyhow::Result<()> {
8        let _admission = self.thread_admission(thread_id).await;
9        if self.goals.active_goal(thread_id).await?.is_some() {
10            self.thread_goal_set(
11                thread_id,
12                roder_api::goals::ThreadGoalPatch {
13                    status: Some(roder_api::goals::ThreadGoalStatus::Paused),
14                    ..Default::default()
15                },
16            )
17            .await?;
18        }
19        let active = self
20            .active_turns
21            .read()
22            .await
23            .iter()
24            .filter(|(_, handle)| &handle.thread_id == thread_id)
25            .map(|(id, _)| id.clone())
26            .collect::<Vec<_>>();
27        for turn_id in active {
28            self.interrupt_turn(thread_id.clone(), turn_id).await?;
29        }
30        loop {
31            let changed = self.active_turns_changed.notified();
32            let active = self
33                .active_turns
34                .read()
35                .await
36                .values()
37                .any(|handle| &handle.thread_id == thread_id);
38            let draining = self
39                .turn_drains
40                .read()
41                .await
42                .values()
43                .any(|drain| &drain.thread_id == thread_id);
44            if !active && !draining {
45                break;
46            }
47            changed.await;
48        }
49        let lifecycle = self.turn_lifecycle_snapshot(thread_id).await?;
50        anyhow::ensure!(
51            !lifecycle
52                .records
53                .iter()
54                .any(|record| record.ownership == TurnCleanupOwnership::ProviderCleanupPending),
55            "provider cleanup is not confirmed for thread {thread_id}"
56        );
57        let slot = self.runner_sessions.lock().await.get(thread_id).cloned();
58        if let Some(slot) = slot {
59            loop {
60                let initialized = slot.initialized.notified();
61                let state = slot.state.lock().await;
62                match &*state {
63                    RunnerSessionSlotState::Initializing => {
64                        drop(state);
65                        initialized.await;
66                    }
67                    RunnerSessionSlotState::Ready(cached) => {
68                        let session = cached.session.clone();
69                        drop(state);
70                        session.close().await?;
71                        break;
72                    }
73                    _ => break,
74                }
75            }
76            self.evict_runner_session(thread_id).await;
77        }
78        self.thread_item_cache.lock().await.remove_thread(thread_id);
79        self.thread_policy_modes.write().await.remove(thread_id);
80        Ok(())
81    }
82}
roder_core/runtime/thread_unload.rs

roder_core/runtime/
thread_unload.rs
