opencode_ctl/
engine.rs

1//! The engine: settings, host, environment and the transport behind every tool.
2
3use std::num::NonZeroUsize;
4use std::path::PathBuf;
5use std::sync::{Arc, Mutex};
6use std::time::Duration;
7
8use serde_json::Value;
9
10use crate::cli::env_nonempty;
11use crate::clock::Clock;
12use crate::compact::{cap_json, redact, redact_text, unwrap_envelope, Cap};
13use crate::created_set::CreatedSet;
14use crate::error::{ErrorCode, ToolError, ToolResult};
15use crate::host::Host;
16use crate::liveness::ProcessProbe;
17use crate::openapi::{Operation, OperationTable, TableSource, SNAPSHOT_VERSION};
18use crate::settings::{Note, Settings};
19use crate::transport::{Lifecycle, Params, Transport, WaitOutcome};
20
21/// What the pairing environment says, read once at startup. A blank value reads as `None`.
22#[derive(Debug, Clone)]
23pub struct EnvSnapshot {
24    pub session: Option<String>,
25    pub directory: Option<PathBuf>,
26    pub server: Option<String>,
27    pub pair: Option<String>,
28    pub host_env: Option<String>,
29    pub cwd: PathBuf,
30    pub home: PathBuf,
31    pub server_password_present: bool,
32}
33
34impl EnvSnapshot {
35    pub fn from_env(ctl_home: PathBuf) -> Self {
36        Self::from_lookup(
37            &|k| env_nonempty(k),
38            std::env::current_dir().unwrap_or_default(),
39            ctl_home,
40        )
41    }
42
43    pub fn from_lookup(get: &dyn Fn(&str) -> Option<String>, cwd: PathBuf, home: PathBuf) -> Self {
44        let nonblank = |k: &str| crate::cli::blank_is_unset(get(k));
45        Self {
46            session: nonblank("OPENCODE_CTL_SESSION"),
47            directory: nonblank("OPENCODE_CTL_DIR").map(PathBuf::from),
48            server: nonblank("OPENCODE_CTL_SERVER"),
49            pair: nonblank("OPENCODE_CTL_PAIR"),
50            host_env: nonblank("OPENCODE_CTL_HOST"),
51            cwd,
52            home,
53            server_password_present: nonblank("OPENCODE_SERVER_PASSWORD").is_some(),
54        }
55    }
56
57    pub fn for_tests(cwd: PathBuf, home: PathBuf) -> Self {
58        Self::from_lookup(&|_| None, cwd, home)
59    }
60}
61
62#[derive(Debug, Default)]
63pub struct OpenapiCache {
64    pub live: Option<OperationTable>,
65    pub pid: Option<u64>,
66    pub version: Option<String>,
67    pub last_error_was_service_down: bool,
68}
69
70pub struct Engine<T: Transport, C: Clock> {
71    pub settings: Settings,
72    pub notes: Vec<Note>,
73    pub host: Host,
74    pub env: EnvSnapshot,
75    pub probe: Arc<dyn ProcessProbe>,
76    transport: T,
77    clock: C,
78    openapi: tokio::sync::Mutex<OpenapiCache>,
79    diff_cache: Mutex<lru::LruCache<DiffKey, (Value, i64)>>,
80}
81
82/// `(session, from, to, context)` of a cached `session.diff` read.
83pub type DiffKey = (String, String, String, Option<u32>);
84
85const DIFF_CACHE_CAPACITY: usize = 2;
86const DIFF_CACHE_TTL_MS: i64 = 10_000;
87const DIFF_CACHE_ENTRY_MAX: usize = 32 * 1024 * 1024;
88
89const EXEMPT_ARRAYS: &[&str] = &["pending_permissions", "pending_forms"];
90
91impl<T: Transport, C: Clock> Engine<T, C> {
92    pub fn new(
93        settings: Settings,
94        notes: Vec<Note>,
95        host: Host,
96        env: EnvSnapshot,
97        transport: T,
98        clock: C,
99        probe: Arc<dyn ProcessProbe>,
100    ) -> Self {
101        Self {
102            settings,
103            notes,
104            host,
105            env,
106            probe,
107            transport,
108            clock,
109            openapi: tokio::sync::Mutex::new(OpenapiCache::default()),
110            diff_cache: Mutex::new(lru::LruCache::new(
111                NonZeroUsize::new(DIFF_CACHE_CAPACITY).expect("nonzero"),
112            )),
113        }
114    }
115
116    /// A cached diff, promoted to most recently used; an entry older than the TTL is dropped (checked on every lookup).
117    pub fn diff_cache_get(&self, key: &DiffKey) -> Option<Value> {
118        let now = self.clock.now_ms();
119        let mut cache = self.diff_cache.lock().unwrap_or_else(|e| e.into_inner());
120        let fresh = cache
121            .peek(key)
122            .map(|(_, at)| now - *at < DIFF_CACHE_TTL_MS)?;
123        if fresh {
124            cache.get(key).map(|(v, _)| v.clone())
125        } else {
126            cache.pop(key);
127            None
128        }
129    }
130
131    /// Cache `value` unless its serialised size exceeds 32 MiB.
132    pub fn diff_cache_put(&self, key: DiffKey, value: &Value) {
133        if value.to_string().len() > DIFF_CACHE_ENTRY_MAX {
134            return;
135        }
136        let now = self.clock.now_ms();
137        let mut cache = self.diff_cache.lock().unwrap_or_else(|e| e.into_inner());
138        cache.put(key, (value.clone(), now));
139    }
140
141    pub fn clock(&self) -> &C {
142        &self.clock
143    }
144
145    /// The live table (fetched once) or, when the fetch fails, the bundled snapshot with a warning.
146    pub async fn operation_table(&self) -> (OperationTable, TableSource, Option<String>) {
147        let mut cache = self.openapi.lock().await;
148        if let Some(live) = &cache.live {
149            return (live.clone(), TableSource::Live, None);
150        }
151        let failure = match self.transport.fetch_openapi().await {
152            Ok(doc) => match OperationTable::from_openapi(&doc, TableSource::Live) {
153                Ok(table) => {
154                    cache.live = Some(table.clone());
155                    cache.last_error_was_service_down = false;
156                    return (table, TableSource::Live, None);
157                }
158                Err(e) => format!("live OpenAPI document unusable: {e}"),
159            },
160            Err(e) => {
161                cache.last_error_was_service_down = e.code == ErrorCode::ServiceDown;
162                format!(
163                    "live OpenAPI document unavailable: {}",
164                    redact_text(&e.message)
165                )
166            }
167        };
168        (
169            OperationTable::snapshot(),
170            TableSource::Snapshot,
171            Some(format!(
172                "{failure}; using the bundled snapshot v{SNAPSHOT_VERSION}"
173            )),
174        )
175    }
176
177    /// Drop the live table when `info` (a `server.info` result) shows another pid or version, or
178    /// when the service has just recovered from `SERVICE_DOWN`.
179    pub async fn invalidate_if_changed(&self, info: &Value) {
180        let pid = info.get("pid").and_then(Value::as_u64);
181        let version = info
182            .get("version")
183            .and_then(Value::as_str)
184            .map(str::to_string);
185        let mut cache = self.openapi.lock().await;
186        let changed = (cache.pid.is_some() && cache.pid != pid)
187            || (cache.version.is_some() && cache.version != version);
188        if changed || cache.last_error_was_service_down {
189            cache.live = None;
190            cache.last_error_was_service_down = false;
191        }
192        cache.pid = pid;
193        cache.version = version;
194    }
195
196    /// The `server.info` version last observed by `invalidate_if_changed`.
197    pub async fn observed_server_version(&self) -> Option<String> {
198        self.openapi.lock().await.version.clone()
199    }
200
201    pub async fn op(&self, id: &str) -> Result<Operation, ToolError> {
202        let (table, _, _) = self.operation_table().await;
203        table.get(id).cloned().ok_or_else(|| {
204            ToolError::new(
205                ErrorCode::UnknownOperation,
206                format!("unknown operation {id}"),
207            )
208            .hint(&format!("nearest: {}", table.nearest(id, 3).join(", ")))
209        })
210    }
211
212    pub async fn cli_version(&self) -> Result<String, ToolError> {
213        Ok(self.transport.cli_version().await?)
214    }
215
216    pub async fn lifecycle(&self, verb: Lifecycle) -> Result<String, ToolError> {
217        Ok(self.transport.lifecycle(verb).await?)
218    }
219
220    pub async fn wait_idle(
221        &self,
222        session_id: &str,
223        deadline: Duration,
224    ) -> Result<WaitOutcome, ToolError> {
225        Ok(self.transport.wait_idle(session_id, deadline).await?)
226    }
227
228    /// One operation call: `(data, cursor, meta)` with the envelope unwrapped and secrets redacted.
229    pub async fn call_json(
230        &self,
231        id: &str,
232        params: Params,
233        body: Option<Value>,
234    ) -> Result<(Value, Option<Value>, Option<Value>), ToolError> {
235        let op = self.op(id).await?;
236        let deadline = Duration::from_secs(self.settings.call_timeout_s);
237        let raw = self
238            .transport
239            .call(&op, &params, body.as_ref(), deadline)
240            .await?;
241        let (mut data, mut cursor, mut meta) = unwrap_envelope(raw)?;
242        redact(&mut data);
243        cursor.iter_mut().for_each(redact);
244        meta.iter_mut().for_each(redact);
245        Ok((data, cursor, meta))
246    }
247
248    /// The tool argument, else the paired session; either must look like a session id.
249    pub fn resolve_session(&self, arg: Option<&str>) -> Result<String, ToolError> {
250        let id = match arg {
251            Some(a) => a,
252            None => self.env.session.as_deref().ok_or_else(|| {
253                ToolError::new(
254                    ErrorCode::NoSession,
255                    "no session_id given and this server is not paired",
256                )
257                .hint("pass session_id, or start the controller through opencode-pair")
258            })?,
259        };
260        if valid_session_id(id) {
261            Ok(id.to_string())
262        } else {
263            Err(ToolError::new(
264                ErrorCode::SessionNotFound,
265                format!("'{}' is not a session id", redact_text(id)),
266            )
267            .hint("session ids match ^ses_[A-Za-z0-9]{26}$; list them with session_list"))
268        }
269    }
270
271    /// The created-session set kept in the ctl home.
272    pub fn created_set(&self) -> CreatedSet {
273        CreatedSet::new(self.env.home.clone())
274    }
275
276    /// On `Host::Opencode` a session-scoped tool accepts only a session this controller created.
277    /// The set is re-read from disk on every call, so a refusal never rests on stale memory.
278    pub fn guard_created(&self, id: &str) -> Result<(), ToolError> {
279        if self.host != Host::Opencode {
280            return Ok(());
281        }
282        match self.created_set().contains(id) {
283            Ok(Some(_)) => Ok(()),
284            Ok(None) => Err(ToolError::new(
285                ErrorCode::RefusedOnHost,
286                format!(
287                    "{} was not created by this controller: on the OpenCode host only sessions made with session_create are accepted",
288                    redact_text(id)
289                ),
290            )
291            .hint("create the worker session first")),
292            Err(e) => Err(ToolError::new(
293                ErrorCode::IoError,
294                format!("cannot read the created-session set: {e}"),
295            )),
296        }
297    }
298
299    /// Append a new session to the set, then probe at most five OTHER ids of that file and prune
300    /// those the service reports as `SESSION_NOT_FOUND` (any other error leaves the entry).
301    pub async fn record_created(
302        &self,
303        directory: &std::path::Path,
304        id: &str,
305    ) -> Result<(), ToolError> {
306        let set = self.created_set();
307        set.append(directory, id).map_err(|e| {
308            ToolError::new(
309                ErrorCode::IoError,
310                format!("session {id} was created but could not be recorded: {e}"),
311            )
312            .hint("the session exists; recording it failed, so this host will refuse it")
313        })?;
314        let others: Vec<String> = set
315            .ids_for(directory)
316            .into_iter()
317            .filter(|o| o != id)
318            .take(crate::created_set::PROBE_LIMIT)
319            .collect();
320        for other in others {
321            let probe = self
322                .call_json(
323                    "session.get",
324                    vec![("sessionID".to_string(), other.clone())],
325                    None,
326                )
327                .await;
328            if let Err(e) = probe {
329                if e.code == ErrorCode::SessionNotFound {
330                    let _ = set.prune(&other);
331                }
332            }
333        }
334        Ok(())
335    }
336
337    pub fn resolve_directory(&self, arg: Option<&std::path::Path>) -> PathBuf {
338        match arg {
339            Some(p) => p.to_path_buf(),
340            None => self
341                .env
342                .directory
343                .clone()
344                .unwrap_or_else(|| self.env.cwd.clone()),
345        }
346    }
347
348    pub fn finish(&self, value: Value) -> Value {
349        cap_json(
350            value,
351            &Cap {
352                max_bytes: self.settings.output_max_bytes,
353                exempt_arrays: EXEMPT_ARRAYS,
354                exempt_entry_max: 1024,
355                exempt_len_max: 20,
356            },
357        )
358    }
359
360    /// Every tool result passes here: redaction, then the byte cap; errors keep the closed shape.
361    pub fn render(&self, result: ToolResult) -> rmcp::model::CallToolResult {
362        use rmcp::model::{CallToolResult, ContentBlock};
363        match result {
364            Ok(mut v) => {
365                redact(&mut v);
366                CallToolResult::success(vec![ContentBlock::text(self.finish(v).to_string())])
367            }
368            Err(mut e) => {
369                e.message = redact_text(&e.message);
370                e.hint = e.hint.map(|h| redact_text(&h));
371                if let Some(d) = e.details.as_mut() {
372                    redact(d);
373                }
374                CallToolResult::error(vec![ContentBlock::text(e.to_json().to_string())])
375            }
376        }
377    }
378}
379
380pub fn valid_session_id(id: &str) -> bool {
381    id.strip_prefix("ses_")
382        .is_some_and(|r| r.len() == 26 && r.chars().all(|c| c.is_ascii_alphanumeric()))
383}
opencode_ctl/engine.rs

opencode_ctl/
engine.rs
