onlyne_client/runtime/runloop/
config.rs

1use crate::runtime::intent::IntentMachine;
2use crate::session::dispatch::DispatchState;
3use anyhow::Result;
4use onlyne_net::backoff::Backoff;
5use onlyne_proto::Welcome;
6use onlyne_session::{AcpOptions, ProcessRunner, WorktreePolicy, backend_for_env, process_env};
7use onlyne_store::ClientStore;
8use std::path::PathBuf;
9use std::sync::{Arc, atomic::AtomicBool};
10use std::time::Duration;
11use tokio::sync::Mutex;
12
13/// Attempt ceiling used until `welcome` carries the role's own value.
14pub const DEFAULT_INTENT_ATTEMPTS: u32 = 3;
15/// Reconnect ladder in seconds (§6: 1/2/4/8/16/32/60).
16pub const RECONNECT_LADDER_SECONDS: [u64; 7] = [1, 2, 4, 8, 16, 32, 60];
17/// Pause between pull attempts that returned nothing.
18pub const PULL_PAUSE_MS: u64 = 200;
19/// Pause between intent flush passes.
20pub const FLUSH_PAUSE_MS: u64 = 200;
21/// Long-poll window the client asks the server for.
22pub const PULL_HOLD_MS: u64 = 1_000;
23/// Deliveries drained per pull.
24pub const PULL_LIMIT: u32 = 32;
25/// Poll interval for the readiness watcher.
26pub const READINESS_POLL_MS: u64 = 250;
27/// Bound on the session sweep the SIGTERM/SIGINT handler runs, short enough
28/// that an operator's own grace period still sees the process leave.
29pub const SHUTDOWN_CLOSE_BUDGET: Duration = Duration::from_secs(8);
30/// Key holding the durable event cursor in `config_cache`.
31pub const EVENT_CURSOR_KEY: &str = "event_seq";
32/// Retry delay for a request that the transport answered `NotReady`.
33pub const NOT_READY_PAUSE_MS: u64 = 200;
34/// Poll cadence for terminal facts emitted by a self-driven session backend.
35pub const OUTCOME_POLL_MS: u64 = 100;
36
37/// Ladder used until `welcome` carries the role's own values.
38pub fn default_intent_backoff() -> Vec<u64> {
39    vec![1_000, 2_000, 4_000]
40}
41
42/// Reconnect ladder as durations, capped at the last rung.
43pub fn reconnect_backoff() -> Backoff {
44    Backoff::with_limits(
45        Duration::from_secs(RECONNECT_LADDER_SECONDS[0]),
46        Duration::from_secs(RECONNECT_LADDER_SECONDS[6]),
47    )
48}
49
50#[derive(Debug, Clone)]
51pub struct ClientInit {
52    pub workspace: PathBuf,
53    pub role: String,
54    pub server: String,
55    pub key_path: PathBuf,
56    pub cert_pin: String,
57    /// The workspace config's `[orca] worktree` value: `host`, `inherit`, or a
58    /// literal Orca worktree selector. Only an Orca session backend reads it.
59    pub orca_worktree: String,
60    /// Seconds a running session may sit without Applied progress before a stall
61    /// fault is reported. Zero disables the report.
62    pub stall_report_secs: u64,
63    /// Seconds a dropped plugin connection may stay away before this client
64    /// retires the task-free session it left behind. Zero disables the sweep.
65    pub reconnect_grace_secs: u64,
66    /// Workspace `config.toml` `backend`. Empty means auto. `ONLYNE_BACKEND`
67    /// in the process environment takes precedence when it is nonempty.
68    pub backend: String,
69    /// The workspace config's `[acp]` table. Only the ACP session backend reads
70    /// it: the mode, model and reasoning effort handed to the agent when a
71    /// session opens, and what to answer when the agent asks for permission.
72    pub acp: onlyne_config::AcpSection,
73}
74
75impl ClientInit {
76    pub fn new(
77        workspace: impl Into<PathBuf>,
78        role: impl Into<String>,
79        server: impl Into<String>,
80        key_path: impl Into<PathBuf>,
81        cert_pin: impl Into<String>,
82    ) -> Self {
83        Self {
84            workspace: workspace.into(),
85            role: role.into(),
86            server: server.into(),
87            key_path: key_path.into(),
88            cert_pin: cert_pin.into(),
89            orca_worktree: "host".to_string(),
90            stall_report_secs: onlyne_config::DEFAULT_STALL_REPORT_SECS,
91            reconnect_grace_secs: onlyne_config::DEFAULT_RECONNECT_GRACE_SECS,
92            backend: String::new(),
93            acp: onlyne_config::AcpSection::default(),
94        }
95    }
96
97    /// Adopt the `[orca] worktree` policy the workspace config carries.
98    pub fn with_orca_worktree(mut self, worktree: impl Into<String>) -> Self {
99        self.orca_worktree = worktree.into();
100        self
101    }
102    pub fn with_stall_report_secs(mut self, secs: u64) -> Self {
103        self.stall_report_secs = secs;
104        self
105    }
106    /// Adopt the `[client] reconnect_grace_secs` value the workspace config carries.
107    pub fn with_reconnect_grace_secs(mut self, secs: u64) -> Self {
108        self.reconnect_grace_secs = secs;
109        self
110    }
111    pub fn with_backend(mut self, backend: impl Into<String>) -> Self {
112        self.backend = backend.into();
113        self
114    }
115    /// Adopt the `[acp]` table the workspace config carries.
116    pub fn with_acp(mut self, acp: onlyne_config::AcpSection) -> Self {
117        self.acp = acp;
118        self
119    }
120}
121
122/// The `[acp]` table in the shape a session backend can read without a config
123/// dependency. The only decision made here is the one the backend acts on: the
124/// permission word, already validated by the config loader, becomes whether this
125/// client grants an agent's request.
126pub fn acp_options(acp: &onlyne_config::AcpSection) -> AcpOptions {
127    AcpOptions {
128        mode: acp.mode.clone(),
129        model: acp.model.clone(),
130        reasoning_effort: acp.reasoning_effort.clone(),
131        allow_permissions: acp.permission == "allow",
132    }
133}
134
135#[derive(Clone)]
136pub struct RunState {
137    pub accept_new: Arc<AtomicBool>,
138    pub store: ClientStore,
139    pub intents: Arc<parking_lot::Mutex<IntentMachine>>,
140    pub dispatch: DispatchState,
141    pub welcome: Arc<Mutex<Option<Welcome>>>,
142    pub stall_report_secs: u64,
143    /// Seconds a dropped plugin connection may stay away before this client
144    /// retires the task-free session it left behind. Zero disables the sweep.
145    pub reconnect_grace_secs: u64,
146}
147
148impl RunState {
149    pub fn new(init: &ClientInit, store: ClientStore) -> Result<Self> {
150        let requested = std::env::var("ONLYNE_BACKEND")
151            .ok()
152            .filter(|value| !value.is_empty())
153            .unwrap_or_else(|| init.backend.clone());
154        let backend = backend_for_env(
155            &requested,
156            &process_env(),
157            Arc::new(ProcessRunner),
158            WorktreePolicy::from_config(&init.orca_worktree),
159            &acp_options(&init.acp),
160        )?;
161        let dispatch = DispatchState::new(
162            init.role.clone(),
163            init.workspace.clone(),
164            Vec::new(),
165            1,
166            Arc::from(backend),
167            store.clone(),
168        );
169        let intents = IntentMachine::new(
170            store.clone(),
171            DEFAULT_INTENT_ATTEMPTS,
172            default_intent_backoff(),
173        );
174        let accept_new = dispatch.accept_new();
175        Ok(Self {
176            accept_new,
177            store,
178            intents: Arc::new(parking_lot::Mutex::new(intents)),
179            dispatch,
180            welcome: Arc::new(Mutex::new(None)),
181            stall_report_secs: init.stall_report_secs,
182            reconnect_grace_secs: init.reconnect_grace_secs,
183        })
184    }
185
186    /// Adopt the role slice the server sent with `welcome`.
187    pub(super) async fn adopt(&self, welcome: &Welcome) {
188        self.dispatch
189            .reconfigure(crate::session::slice::RoleSlice::from_welcome(welcome));
190        // The topology name is the address the host backends group sessions
191        // under, so it is recorded with the rest of what the server says about
192        // this role. `welcome.cluster` is the server's own `[server] name`.
193        self.dispatch.set_topology(&welcome.cluster);
194        {
195            let mut intents = self.intents.lock();
196            if let Some(attempts) = welcome.intent_attempts {
197                intents.attempts = attempts;
198            }
199            if let Some(ladder) = welcome
200                .intent_backoff_ms
201                .as_ref()
202                .filter(|ladder| !ladder.is_empty())
203            {
204                intents.backoff_ms = ladder.clone();
205            }
206        }
207        *self.welcome.lock().await = Some(welcome.clone());
208    }
209
210    /// Durable event cursor for the next `subscribe`. Zero asks the server for
211    /// its current head.
212    pub(super) fn cursor(&self) -> u64 {
213        self.store
214            .config(EVENT_CURSOR_KEY)
215            .ok()
216            .flatten()
217            .and_then(|value| value.parse().ok())
218            .unwrap_or(0)
219    }
220
221    pub(super) fn set_cursor(&self, seq: u64) {
222        if let Err(error) = self.store.put_config(EVENT_CURSOR_KEY, &seq.to_string()) {
223            tracing::warn!(error = %error, "event cursor was not stored");
224        }
225    }
226}
onlyne_client/runtime/runloop/config.rs

onlyne_client/runtime/runloop/
config.rs
