1. 1//! vak-server: HTTP+SSE wrapper around vak-core. The TUI, web clients,
2. 2//! IDE extensions, and the desktop shell are all just consumers of these
3. 3//! endpoints — one headless agent core, many surfaces.
4. 4//!
5. 5//! Endpoints:
6. 6//! - `GET /health`
7. 7//! - `POST /sessions` → {session_id}
8. 8//! - `GET /sessions` → persisted session summaries (sidebar)
9. 9//! - `POST /sessions/:id/attach` → resume a persisted session into memory
10. 10//! - `POST /sessions/:id/run` {prompt} → 202 (events stream on SSE)
11. 11//! - `POST /sessions/:id/steering` {text} → 202
12. 12//! - `POST /sessions/:id/approvals/:rid` {approve} → resolve a pending gate
13. 13//! - `GET /sessions/:id/events` → SSE of AgentEvent JSON
14. 14//! - `GET /sessions/:id/transcript` → derived messages + usage
15. 15//! - `GET /sessions/:id/transcript.md` → markdown export (shared renderer)
16. 16//! - `GET /sessions/:id/diff` → git diff + status of the workspace
17. 17//! - `GET /sessions/:id/checkpoints` → workspace snapshots (time travel)
18. 18//! - `POST /sessions/:id/checkpoints/:seq/restore` → rewind the workspace
19. 19//! - `POST /sessions/:id/archive` {archived} → toggle sidebar visibility
20. 20//! - `GET /skills` → discovered skills (name + description)
21. 21//! - `GET /fs/file?path=` → read a file confined to cwd
22. 22//! - `PUT /fs/file` {path, content} → write a file confined to cwd
23. 23//! - `POST /config/mode` {mode} → switch permission mode at runtime
24. 24//! - `PUT /config/key` {provider, key} → store a provider credential (0600)
25. 25//! - `DELETE /config/key` {provider} → revoke a stored credential
26. 26//! - `GET /providers` → provider picker data (no secrets)
27. 27//! - `GET /providers/:name/models` → models the stored key can reach
28. 28//! - `GET /providers/:name/status` → provider-published account metadata
29. 29//! - `PATCH/DELETE /memory/:note_id` → amend / forget one memory note
30. 30//! - `GET /search?all=true` → cross-project recall (23-memory)
31. 31//! - `GET /doctor?session=` → HealthReport JSON (29-personal-os P3)
32. 32//! - `POST /backup/export` → directory backup of the home dir
33. 33//! - `POST /backup/import` → restore with skip-or-rename conflicts
34. 34//! - `GET /digest?days=N` → usage digest over the trailing window
35. 35//! - `GET /inbox?limit=&unread=true` → inbox entries + unread count (29-personal-os P6)
36. 36//! - `POST /inbox/:id/ack` → idempotent read-state tombstone
37. 37//! - `GET /inbox/unread_count` → live unread total
38. 38//! - `POST /gateway/inbound` → surface message routed to its bound session (22-gateway)
39. 39//! - `GET /gateway/status` → gateway enabled flag + binding table
40. 40//! - `DELETE /gateway/bindings/:key` → unbind a surface from its session
41. 41//! - `POST /agent-network/capabilities` → issue an explicitly scoped agent capability
42. 42//! - `POST /agent-network/messages` → broker a bounded workspace message
43. 43//! - `GET /agent-network/messages` → receive queued workspace messages
44. 44//! - `GET/POST /presentations` → inspect/register validated experience-pack records
45. 45//! - `POST /presentations/revisions` → validate and store a disabled immutable revision preview
46. 46//! - `POST /presentations/:id/:revision/activate` → explicitly activate one scoped revision
47. 47//! - `POST /presentations/:id/deactivate` → remove one scoped activation
48. 48//! - `DELETE /presentations/plugins/:plugin_id` → revoke a plugin's presentation records
49. 49
50. 50mod admin;
51. 51mod admin_ui;
52. 52mod agent_chats;
53. 53pub mod agents;
54. 54mod bus;
55. 55mod channels;
56. 56mod client_events;
57. 57mod client_ui;
58. 58mod core_pool;
59. 59mod coworking;
60. 60mod delivery;
61. 61mod embedded_ui;
62. 62mod events;
63. 63mod feeds;
64. 64pub mod gateway;
65. 65mod heartbeat;
66. 66mod inbox;
67. 67mod office_workspace;
68. 68mod operations;
69. 69mod projection;
70. 70mod rate_limit;
71. 71mod service_control;
72. 72mod site;
73. 73mod stream;
74. 74pub mod surfaces;
75. 75mod voice;
76. 76mod web;
77. 77
78. 78#[cfg(test)]
79. 79#[allow(clippy::unwrap_used, clippy::expect_used)]
80. 80pub(crate) mod test_support {
81. 81 /// A throwaway `AppState` rooted in a temp dir.
82. 82 ///
83. 83 /// `set_sessions_home` is not optional: without it `sessions_home()`
84. 84 /// falls back to the developer's real data home, and `AppState::new`
85. 85 /// loads (and can seed) the gateway allowlist store there.
86. 86 pub(crate) fn state() -> crate::AppState {
87. 87 let dir = tempfile::tempdir().unwrap();
88. 88 let core = vak_core::Core::new(dir.path().to_path_buf()).unwrap();
89. 89 core.set_sessions_home(dir.path().join("home"));
90. 90 // The TempDir guard is deliberately leaked: these states outlive
91. 91 // the call and a removed directory would fail reads mid-test.
92. 92 std::mem::forget(dir);
93. 93 crate::AppState::new(core)
94. 94 }
95. 95}
96. 96
97. 97use std::collections::{HashMap, HashSet};
98. 98use std::io::Write;
99. 99use std::path::PathBuf;
100. 100use std::sync::{Arc, Mutex};
101. 101use std::time::{Duration, Instant};
102. 102
103. 103use chrono::Utc;
104. 104
105. 105use axum::extract::{Path, Query, State};
106. 106use axum::http::StatusCode;
107. 107use axum::response::IntoResponse;
108. 108use axum::response::sse::{Event, KeepAlive, Sse};
109. 109use axum::routing::{delete, get, post, put};
110. 110use axum::{Json, Router};
111. 111use tokio::sync::{broadcast, mpsc, oneshot};
112. 112use tokio_util::sync::CancellationToken;
113. 113
114. 114use vak_agent::{AgentEvent, Approver, SteeringQueues};
115. 115use vak_core::Core;
116. 116pub use vak_core::tasks::{TaskDef, WtMeta};
117. 117use vak_llm::Provider;
118. 118use vak_plugin::{InstallOptions, InstallScope, MarketplaceTrust, PluginStore, SignatureEvidence};
119. 119use vak_session::SessionLog;
120. 120
121. 121/// Ceiling on simultaneously-cached session handles. Generous on purpose:
122. 122/// eviction should be invisible to interactive use and only bound a
123. 123/// long-running gateway process.
124. 124const MAX_LIVE_SESSIONS: usize = 128;
125. 125
126. 126/// Test-only override so eviction can be exercised without creating 129
127. 127/// real sessions. Zero (the default) means "use `MAX_LIVE_SESSIONS`".
128. 128/// Process-global like `pin_test_data_home`; a test that sets it restores
129. 129/// zero afterward so it cannot leak into an unrelated concurrent test.
130. 130#[cfg(test)]
131. 131static MAX_LIVE_SESSIONS_TEST_OVERRIDE: std::sync::atomic::AtomicUsize =
132. 132 std::sync::atomic::AtomicUsize::new(0);
133. 133
134. 134fn max_live_sessions() -> usize {
135. 135 #[cfg(test)]
136. 136 {
137. 137 let over = MAX_LIVE_SESSIONS_TEST_OVERRIDE.load(std::sync::atomic::Ordering::SeqCst);
138. 138 if over != 0 {
139. 139 return over;
140. 140 }
141. 141 }
142. 142 MAX_LIVE_SESSIONS
143. 143}
144. 144
145. 145pub(crate) struct SessionHandle {
146. 146 pub(crate) id: String,
147. 147 /// The `Core` this session runs under, resolved once at creation.
148. 148 ///
149. 149 /// A session's contract freezes when it is created (invariant 17), and
150. 150 /// its permission ceiling is part of that contract — so it belongs on
151. 151 /// the handle rather than being re-read from `state.core` at dispatch.
152. 152 /// The gateway already passes its pooled `Core` explicitly into turn
153. 153 /// execution; this is the same fact, held where every run path can see
154. 154 /// it, which is what lets a session be capped *below* the workspace
155. 155 /// mode (doc 46 security invariant 5).
156. 156 pub(crate) core: Core,
157. 157 /// Workspace this session's tools/diffs operate in (main cwd, or a
158. 158 /// best-of-N worktree).
159. 159 pub(crate) cwd: PathBuf,
160. 160 pub(crate) session: Arc<Mutex<Option<SessionLog>>>,
161. 161 /// Latest host-admitted intent, retained while the runner owns the log.
162. 162 pub(crate) intent: Arc<Mutex<Option<vak_session::types::IntentRecord>>>,
163. 163 pub(crate) steering: Arc<SteeringQueues>,
164. 164 /// Cancel for the CURRENT run only; replaced with a fresh token when a
165. 165 /// run ends so one `/cancel` doesn't poison every later run.
166. 166 pub(crate) cancel: Arc<std::sync::Mutex<CancellationToken>>,
167. 167 /// Live events for the MAIN transcript, with replay so a dropped
168. 168 /// connection can resume rather than lose the gap (events::EventBus).
169. 169 pub(crate) events_tx: events::EventBus,
170. 170 /// Content-free wakeup for shared candidate comments.
171. 171 pub(crate) coworking_comments_tx: tokio::sync::broadcast::Sender<()>,
172. 172 /// Pending approval gates scoped to THIS session — a client holding
173. 173 /// session A can never resolve session B's approvals.
174. 174 pub(crate) pending: Arc<Mutex<HashMap<String, ApprovalRequest>>>,
175. 175 /// Lifecycle facts produced while the runner owns the SessionLog. They
176. 176 /// are appended atomically when the runner returns the ledger.
177. 177 pub(crate) activity_buffer: Arc<Mutex<Vec<vak_session::ActivityRecord>>>,
178. 178 /// Reconnectable presentation state while the runner owns the ledger.
179. 179 pub(crate) presentation: Arc<Mutex<vak_delivery::OutputTimeline>>,
180. 180 /// Notified when an SSE consumer attaches, so runs don't start (and
181. 181 /// finish) before anyone is listening.
182. 182 pub(crate) subscribed: Arc<tokio::sync::Notify>,
183. 183 /// Side-chat stream + cancel: branched turns that read the session
184. 184 /// context but never land on the main chain.
185. 185 /// Live events for the `/btw` side branch. Same bus type as the
186. 186 /// main transcript so both resume identically.
187. 187 pub(crate) side_events_tx: events::EventBus,
188. 188 /// Last time a request resolved this handle, for idle eviction.
189. 189 pub(crate) last_touched: Mutex<std::time::Instant>,
190. 190 pub(crate) side_cancel: Arc<std::sync::Mutex<CancellationToken>>,
191. 191 /// Admission identities currently owned by this handle. This closes the
192. 192 /// retry race while the runner owns the ledger.
193. 193 pub(crate) admissions: Arc<Mutex<HashSet<String>>>,
194. 194}
195. 195
196. 196#[derive(Clone)]
197. 197pub struct AppState {
198. 198 pub core: Core,
199. 199 /// Monotonic start point for this server state. Keeping it on the state
200. 200 /// avoids reporting the first Operations request as process start and
201. 201 /// keeps embedded/test routers independent from one another.
202. 202 started_at: Instant,
203. 203 /// Port used by the bound server. Plain routers use the configured
204. 204 /// operations default; `serve_with` overwrites this with its actual
205. 205 /// listener port so health probes and the console never drift from the
206. 206 /// process being inspected.
207. 207 ops_port: u16,
208. 208 sessions: Arc<Mutex<HashMap<String, Arc<SessionHandle>>>>,
209. 209 /// Ephemeral, observation-based coworking presence. It is deliberately
210. 210 /// outside the append-only work ledger and expires without disconnect
211. 211 /// bookkeeping when a browser vanishes.
212. 212 coworking_presence: Arc<Mutex<HashMap<String, HashMap<String, CoworkingPresence>>>>,
213. 213 /// Live best-of-N runs keyed by child session id.
214. 214 pub(crate) best_runs: Arc<Mutex<HashMap<String, BestRunMeta>>>,
215. 215 /// Scheduled tasks for this workspace (store shape owned by vak-core).
216. 216 tasks: Arc<Mutex<HashMap<String, TaskDef>>>,
217. 217 /// In-memory cron markers: task id → next scheduled local fire. Interval
218. 218 /// tasks keep using `last_run_at`; only `schedule:` tasks appear here.
219. 219 next_fire: Arc<Mutex<HashMap<String, chrono::DateTime<chrono::Local>>>>,
220. 220 /// Script tasks currently executing (no child session to inspect, so
221. 221 /// this stands in for the busy-check that prompt tasks get).
222. 222 script_inflight: Arc<Mutex<std::collections::HashSet<String>>>,
223. 223 /// Managed dev servers (preview pane), keyed by session::name.
224. 224 procs: Arc<Mutex<HashMap<String, ManagedProc>>>,
225. 225 /// Gateway surface bindings + enable gate (docs/design/22-gateway.md).
226. 226 pub(crate) gateway: Arc<gateway::GatewayState>,
227. 227 /// Proactive heartbeat runtime (docs/design/29-personal-os.md P7).
228. 228 pub(crate) heartbeat: Arc<heartbeat::HeartbeatRuntime>,
229. 229 /// Global event hub for admin console SSE streaming.
230. 230 pub(crate) hub: events::EventHub,
231. 231 /// SQLite FTS5 session index (rebuildable from JSONL).
232. 232 pub(crate) store: Option<vak_store::Store>,
233. 233 /// Expected auth token (login endpoint compares against it).
234. 234 pub(crate) auth_token: Arc<String>,
235. 235 /// Workspace the browser client currently has open, when it has moved
236. 236 /// away from the one this process started in (docs/design/48-web-client.md
237. 237 /// §5). `None` means "the process's own workspace".
238. 238 ///
239. 239 /// Only *new* sessions are affected: a session freezes its `Core` at
240. 240 /// creation (invariant 17), so switching the active workspace never
241. 241 /// retargets work already under way — it decides where the next task
242. 242 /// will live, which is exactly what an operator switching projects
243. 243 /// means by it.
244. 244 pub(crate) active_core: Arc<Mutex<Option<Core>>>,
245. 245 /// Number of live voice websocket sessions. Admission is checked against
246. 246 /// the effective configuration at connection time and released on exit.
247. 247 pub(crate) voice_active: Arc<std::sync::atomic::AtomicUsize>,
248. 248 /// Per-minute budget shared by every paid voice request.
249. 249 pub(crate) voice_requests: Arc<voice::RequestWindow>,
250. 250}
251. 251
252. 252#[derive(Clone)]
253. 253pub struct BestRunMeta {
254. 254 pub repo: PathBuf,
255. 255 pub wt_path: PathBuf,
256. 256 pub branch: String,
257. 257}
258. 258
259. 259impl AppState {
260. 260 pub fn new(core: Core) -> Self {
261. 261 let gateway = Arc::new(gateway::GatewayState::load(&core, false));
262. 262 let hub = events::init_global();
263. 263 // Canonical layout (doc 32): the FTS index is a rebuildable cache,
264. 264 // never user data — it lives under Library/Caches / XDG_CACHE_HOME.
265. 265 let store = vak_store::Store::open(&core.cache_home()).ok();
266. 266 if store.is_none() {
267. 267 eprintln!("[warn] store open failed, search will use fallback");
268. 268 }
269. 269 // Token selection lives here so every router flavor (plain,
270. 270 // gateway, secured) shares one identity for auth + login.
271. 271 let auth_token = Arc::new(
272. 272 std::env::var("VAK_GATEWAY_TOKEN")
273. 273 .ok()
274. 274 .filter(|t| !t.trim().is_empty())
275. 275 .or_else(|| vak_config::get_var("VAK_GATEWAY_TOKEN"))
276. 276 .filter(|t| !t.trim().is_empty())
277. 277 .unwrap_or_else(|| format!("vk_{}", uuid::Uuid::now_v7())),
278. 278 );
279. 279 AppState {
280. 280 core,
281. 281 started_at: Instant::now(),
282. 282 ops_port: vak_ops::OpsConfig::detect().port,
283. 283 sessions: Arc::new(Mutex::new(HashMap::new())),
284. 284 coworking_presence: Arc::new(Mutex::new(HashMap::new())),
285. 285 best_runs: Arc::new(Mutex::new(HashMap::new())),
286. 286 tasks: Arc::new(Mutex::new(HashMap::new())),
287. 287 next_fire: Arc::new(Mutex::new(HashMap::new())),
288. 288 script_inflight: Arc::new(Mutex::new(std::collections::HashSet::new())),
289. 289 procs: Arc::new(Mutex::new(HashMap::new())),
290. 290 gateway,
291. 291 heartbeat: Arc::new(heartbeat::HeartbeatRuntime::new()),
292. 292 hub,
293. 293 store,
294. 294 auth_token,
295. 295 active_core: Arc::new(Mutex::new(None)),
296. 296 voice_active: Arc::new(std::sync::atomic::AtomicUsize::new(0)),
297. 297 voice_requests: Arc::new(voice::RequestWindow::new()),
298. 298 }
299. 299 }
300. 300
301. 301 /// The `Core` new work should run under: the browser's chosen
302. 302 /// workspace if it has picked one, else this process's own.
303. 303 pub(crate) fn active_core(&self) -> Core {
304. 304 self.active_core
305. 305 .lock()
306. 306 .unwrap_or_else(std::sync::PoisonError::into_inner)
307. 307 .clone()
308. 308 .unwrap_or_else(|| self.core.clone())
309. 309 }
310. 310
311. 311 /// Force-enable the gateway (`serve --gateway`) before the state is
312. 312 /// shared; the config gate alone governs every other entry point.
313. 313 pub fn enable_gateway(&mut self) {
314. 314 if let Some(gw) = Arc::get_mut(&mut self.gateway) {
315. 315 gw.set_enabled(true);
316. 316 }
317. 317 }
318. 318
319. 319 fn get(&self, id: &str) -> Option<Arc<SessionHandle>> {
320. 320 let handle = self
321. 321 .sessions
322. 322 .lock()
323. 323 .unwrap_or_else(std::sync::PoisonError::into_inner)
324. 324 .get(id)
325. 325 .cloned();
326. 326 if let Some(handle) = &handle
327. 327 && let Ok(mut touched) = handle.last_touched.lock()
328. 328 {
329. 329 *touched = std::time::Instant::now();
330. 330 }
331. 331 handle
332. 332 }
333. 333
334. 334 /// Drop the least-recently-touched idle sessions once the live set exceeds
335. 335 /// [`MAX_LIVE_SESSIONS`].
336. 336 ///
337. 337 /// The map was insert-only. Each handle pins the whole ledger in memory
338. 338 /// (`Vec<Entry>` of every message, tool result, and receipt) plus a
339. 339 /// presentation snapshot and two broadcast channels, so a long-lived
340. 340 /// gateway process grew without bound — and because `SessionLog::open`
341. 341 /// holds an exclusive file lock for the handle's lifetime, every session
342. 342 /// the daemon ever touched stayed locked against the CLI.
343. 343 ///
344. 344 /// Eviction is deliberately conservative: a session is only a candidate
345. 345 /// when nothing else holds a reference, no SSE client is subscribed, and
346. 346 /// the runner is not holding the ledger. `/sessions/{id}/attach` re-opens
347. 347 /// an evicted session from disk, so this is a cache bound, not a
348. 348 /// lifecycle.
349. 349 fn evict_idle_sessions(&self) {
350. 350 let cap = max_live_sessions();
351. 351 let mut sessions = self
352. 352 .sessions
353. 353 .lock()
354. 354 .unwrap_or_else(std::sync::PoisonError::into_inner);
355. 355 if sessions.len() <= cap {
356. 356 return;
357. 357 }
358. 358 let mut idle: Vec<(std::time::Instant, String)> = sessions
359. 359 .iter()
360. 360 .filter(|(_, handle)| {
361. 361 // `events_tx` always carries the handle's own internal
362. 362 // projector (`register_handle`), so `receiver_count()` can
363. 363 // never read zero; `external_subscribers()` excludes it.
364. 364 // `side_events_tx` has no such internal subscriber, so a
365. 365 // plain `receiver_count()` remains correct there.
366. 366 Arc::strong_count(handle) == 1
367. 367 && handle.events_tx.external_subscribers() == 0
368. 368 && handle.side_events_tx.receiver_count() == 0
369. 369 && handle.session.lock().is_ok_and(|guard| guard.is_some())
370. 370 })
371. 371 .filter_map(|(id, handle)| {
372. 372 let touched = *handle.last_touched.lock().ok()?;
373. 373 Some((touched, id.clone()))
374. 374 })
375. 375 .collect();
376. 376 idle.sort_by_key(|(touched, _)| *touched);
377. 377 let mut over = sessions.len().saturating_sub(cap);
378. 378 for (_, id) in idle {
379. 379 if over == 0 {
380. 380 break;
381. 381 }
382. 382 sessions.remove(&id);
383. 383 over -= 1;
384. 384 }
385. 385 }
386. 386
387. 387 /// Drops a session's live handle, so a trashed session is not served
388. 388 /// from memory after it leaves every list.
389. 389 fn forget_session(&self, id: &str) {
390. 390 self.sessions
391. 391 .lock()
392. 392 .unwrap_or_else(std::sync::PoisonError::into_inner)
393. 393 .remove(id);
394. 394 }
395. 395
396. 396 /// Snapshot of every live session handle (admin surfaces aggregate
397. 397 /// across sessions; nothing here crosses a session's approval scope —
398. 398 /// answering still goes through the per-session endpoint).
399. 399 pub(crate) fn live_handles(&self) -> Vec<Arc<SessionHandle>> {
400. 400 self.sessions
401. 401 .lock()
402. 402 .unwrap_or_else(std::sync::PoisonError::into_inner)
403. 403 .values()
404. 404 .cloned()
405. 405 .collect()
406. 406 }
407. 407}
408. 408
409. 409#[cfg(test)]
410. 410#[allow(clippy::unwrap_used, clippy::expect_used)]
411. 411mod eviction_tests {
412. 412 use super::*;
413. 413
414. 414 /// Restores the process-global test override on drop, so a panic mid-test
415. 415 /// cannot leave a tiny cap active for an unrelated concurrent test.
416. 416 struct RestoreCap;
417. 417 impl Drop for RestoreCap {
418. 418 fn drop(&mut self) {
419. 419 MAX_LIVE_SESSIONS_TEST_OVERRIDE.store(0, std::sync::atomic::Ordering::SeqCst);
420. 420 }
421. 421 }
422. 422
423. 423 /// Finding 3: every handle's `events_tx` carries the internal projector
424. 424 /// subscription from `register_handle`, so the old `receiver_count() ==
425. 425 /// 0` eviction guard never held — a long-lived process kept every
426. 426 /// session's ledger (and its exclusive file lock) in memory forever.
427. 427 /// `external_subscribers()` fixes the guard; this proves eviction
428. 428 /// actually runs once the live set exceeds the cap.
429. 429 #[tokio::test]
430. 430 async fn idle_sessions_beyond_the_cap_are_evicted() {
431. 431 vak_config::paths::isolate_home_for_tests();
432. 432 MAX_LIVE_SESSIONS_TEST_OVERRIDE.store(3, std::sync::atomic::Ordering::SeqCst);
433. 433 let _restore = RestoreCap;
434. 434
435. 435 let dir = tempfile::tempdir().unwrap();
436. 436 let core = Core::new(dir.path().to_path_buf()).unwrap();
437. 437 core.set_sessions_home(dir.path().join("home"));
438. 438 let state = AppState::new(core.clone());
439. 439
440. 440 let mut ids = Vec::new();
441. 441 for _ in 0..5 {
442. 442 let session = core.start_session().await.unwrap();
443. 443 let id = session.header().unwrap().session_id.clone();
444. 444 // No SSE client, no run in progress, and the returned Arc is
445. 445 // dropped immediately — exactly the "nothing references it"
446. 446 // shape `evict_idle_sessions` looks for.
447. 447 let _ = register_handle(
448. 448 &state,
449. 449 id.clone(),
450. 450 session,
451. 451 core.cwd().clone(),
452. 452 core.clone(),
453. 453 );
454. 454 ids.push(id);
455. 455 }
456. 456
457. 457 let live: Vec<String> = state.sessions.lock().unwrap().keys().cloned().collect();
458. 458 assert_eq!(
459. 459 live.len(),
460. 460 3,
461. 461 "expected eviction down to the test cap, got {live:?}"
462. 462 );
463. 463 // Eviction drops the least-recently-touched handles first, so the
464. 464 // most recently created session must survive.
465. 465 assert!(
466. 466 live.contains(ids.last().unwrap()),
467. 467 "the newest session must not be evicted: {live:?}"
468. 468 );
469. 469 assert!(
470. 470 !live.contains(&ids[0]),
471. 471 "the oldest session must be evicted first: {live:?}"
472. 472 );
473. 473 }
474. 474}
475. 475
476. 476#[derive(Clone)]
477. 477struct CoworkingPresence {
478. 478 display_name: String,
479. 479 seen_at: Instant,
480. 480 office_room_id: Option<String>,
481. 481 office_anchor: Option<String>,
482. 482}
483. 483
484. 484#[derive(Clone)]
485. 485pub struct ApprovalRequest {
486. 486 pub id: String,
487. 487 pub tool: String,
488. 488 pub args_json: String,
489. 489 pub reason: String,
490. 490 pub requested_at: chrono::DateTime<chrono::Utc>,
491. 491 respond: Arc<Mutex<Option<oneshot::Sender<bool>>>>,
492. 492 answered_by: Arc<Mutex<Option<(String, String)>>>,
493. 493 delegated_to: Arc<Mutex<Option<String>>>,
494. 494}
495. 495
496. 496impl ApprovalRequest {
497. 497 pub fn respond(&self, approve: bool) {
498. 498 if let Some(tx) = self
499. 499 .respond
500. 500 .lock()
501. 501 .unwrap_or_else(std::sync::PoisonError::into_inner)
502. 502 .take()
503. 503 {
504. 504 let _ = tx.send(approve);
505. 505 }
506. 506 }
507. 507}
508. 508
509. 509/// How long an HTTP-surfaced gate waits for a console or desktop client to
510. 510/// answer before failing closed.
511. 511///
512. 512/// There was no bound at all, which was survivable while every run behind
513. 513/// this approver had a human watching an SSE stream — and was not, once the
514. 514/// scheduler started firing runs through the same path. An unanswered gate
515. 515/// held the session handle open forever, so the routine never completed and
516. 516/// its slot never freed. Generous, because a person may genuinely be away
517. 517/// from the tab, but finite: a run that fails closed can be retried, and one
518. 518/// that hangs cannot.
519. 519const HTTP_APPROVAL_TIMEOUT: Duration = Duration::from_secs(900);
520. 520
521. 521struct HttpApprover {
522. 522 events_tx: events::EventBus,
523. 523 pending: Arc<Mutex<HashMap<String, ApprovalRequest>>>,
524. 524 /// Owning session, so admin-console surfaces can attribute gates.
525. 525 session_id: String,
526. 526 activity_buffer: Arc<Mutex<Vec<vak_session::ActivityRecord>>>,
527. 527 /// False when this run has no client watching — a scheduled routine, a
528. 528 /// best-of-N leg. The gate is then a foregone denial, and saying so
529. 529 /// through `answerable()` is what lets `vak_core::reach` drop the
530. 530 /// capability from the turn instead of letting the model discover it by
531. 531 /// blocking on a question nobody will read.
532. 532 answerable: bool,
533. 533}
534. 534
535. 535#[async_trait::async_trait]
536. 536impl Approver for HttpApprover {
537. 537 fn answerable(&self) -> bool {
538. 538 self.answerable
539. 539 }
540. 540
541. 541 async fn approve(&self, tool: &str, args_json: &str, reason: &str) -> bool {
542. 542 if !self.answerable {
543. 543 return false;
544. 544 }
545. 545 let id = uuid::Uuid::now_v7().to_string();
546. 546 let (respond, rx) = oneshot::channel();
547. 547 let answered_by = Arc::new(Mutex::new(None));
548. 548 self.pending
549. 549 .lock()
550. 550 .unwrap_or_else(std::sync::PoisonError::into_inner)
551. 551 .insert(
552. 552 id.clone(),
553. 553 ApprovalRequest {
554. 554 id: id.clone(),
555. 555 tool: tool.to_string(),
556. 556 args_json: args_json.to_string(),
557. 557 reason: reason.to_string(),
558. 558 requested_at: chrono::Utc::now(),
559. 559 respond: Arc::new(Mutex::new(Some(respond))),
560. 560 answered_by: answered_by.clone(),
561. 561 delegated_to: Arc::new(Mutex::new(None)),
562. 562 },
563. 563 );
564. 564 let _ = self.events_tx.send(AgentEvent::ApprovalRequested {
565. 565 id: id.clone(),
566. 566 tool: tool.to_string(),
567. 567 args_json: args_json.to_string(),
568. 568 reason: reason.to_string(),
569. 569 });
570. 570 self.activity_buffer
571. 571 .lock()
572. 572 .unwrap_or_else(std::sync::PoisonError::into_inner)
573. 573 .push(vak_session::ActivityRecord {
574. 574 activity_id: format!("approval-{id}"),
575. 575 turn: None,
576. 576 kind: vak_session::ActivityKind::Approval,
577. 577 status: vak_session::ActivityStatus::Pending,
578. 578 label: format!("Approval required for {tool}"),
579. 579 detail: Some(reason.to_string()),
580. 580 data: [
581. 581 ("request_id".into(), id.clone()),
582. 582 ("tool".into(), tool.to_string()),
583. 583 ("args_json".into(), args_json.to_string()),
584. 584 ]
585. 585 .into(),
586. 586 });
587. 587 if let Some(hub) = events::global() {
588. 588 hub.emit(events::SystemEvent::ApprovalRequested {
589. 589 id: id.clone(),
590. 590 session_id: self.session_id.clone(),
591. 591 tool: tool.to_string(),
592. 592 reason: reason.to_string(),
593. 593 });
594. 594 }
595. 595 // Bounded, and failing closed on expiry — the same contract
596. 596 // `GatewayApprover` already had. Dropping the entry before returning
597. 597 // means a reply that arrives after the deadline resolves nothing
598. 598 // rather than answering a gate the run has already moved past.
599. 599 let approved = match tokio::time::timeout(HTTP_APPROVAL_TIMEOUT, rx).await {
600. 600 Ok(answer) => answer.unwrap_or(false),
601. 601 Err(_) => {
602. 602 eprintln!(
603. 603 "[approvals] gate {} for `{tool}` expired after {}s; denied",
604. 604 &id[..8.min(id.len())],
605. 605 HTTP_APPROVAL_TIMEOUT.as_secs()
606. 606 );
607. 607 false
608. 608 }
609. 609 };
610. 610 self.pending
611. 611 .lock()
612. 612 .unwrap_or_else(std::sync::PoisonError::into_inner)
613. 613 .remove(&id);
614. 614 let answered_by = answered_by
615. 615 .lock()
616. 616 .unwrap_or_else(std::sync::PoisonError::into_inner)
617. 617 .clone();
618. 618 self.activity_buffer
619. 619 .lock()
620. 620 .unwrap_or_else(std::sync::PoisonError::into_inner)
621. 621 .push(vak_session::ActivityRecord {
622. 622 activity_id: format!("approval-{id}"),
623. 623 turn: None,
624. 624 kind: vak_session::ActivityKind::Approval,
625. 625 status: if approved {
626. 626 vak_session::ActivityStatus::Succeeded
627. 627 } else {
628. 628 vak_session::ActivityStatus::Denied
629. 629 },
630. 630 label: format!(
631. 631 "Approval {} for {tool}",
632. 632 if approved { "granted" } else { "denied" }
633. 633 ),
634. 634 detail: Some(reason.to_string()),
635. 635 data: [
636. 636 ("request_id".into(), id.clone()),
637. 637 ("tool".into(), tool.to_string()),
638. 638 ("args_json".into(), args_json.to_string()),
639. 639 ]
640. 640 .into_iter()
641. 641 .chain(answered_by.into_iter().flat_map(|(actor_id, actor_name)| {
642. 642 [
643. 643 ("actor_id".into(), actor_id),
644. 644 ("actor_name".into(), actor_name),
645. 645 ]
646. 646 }))
647. 647 .collect(),
648. 648 });
649. 649 if let Some(hub) = events::global() {
650. 650 hub.emit(if approved {
651. 651 events::SystemEvent::ApprovalGranted {
652. 652 id: id.clone(),
653. 653 tool: tool.to_string(),
654. 654 }
655. 655 } else {
656. 656 events::SystemEvent::ApprovalDenied {
657. 657 id,
658. 658 tool: tool.to_string(),
659. 659 }
660. 660 });
661. 661 }
662. 662 approved
663. 663 }
664. 664}
665. 665
666. 666/// Resolve the Agent-scoped `Core` for the rest of an async handler body, or
667. 667/// return early with `resolve_scoped_core`'s own error `Response` (forwarding
668. 668/// its status/message unchanged, e.g. 409 CONFLICT for a paused/archived
669. 669/// Agent). Every endpoint scoped this way needs the identical
670. 670/// match-and-early-return, so it lives here once instead of copy-pasted at
671. 671/// each of the 50+ call sites (see `resolve_scoped_core` below).
672. 672macro_rules! scoped_core {
673. 673 ($state:expr, $session_id:expr, $agent:expr) => {
674. 674 match resolve_scoped_core($state, $session_id, $agent) {
675. 675 Ok(core) => core,
676. 676 Err(response) => return response,
677. 677 }
678. 678 };
679. 679}
680. 680
681. 681pub fn router(core: Core) -> Router {
682. 682 router_with_state(AppState::new(core))
683. 683}
684. 684
685. 685/// Unauthenticated router with the gateway force-enabled and no background
686. 686/// scheduler. For embedders that run their own supervision loop and need
687. 687/// clean teardown: dropping this router releases every session lock,
688. 688/// whereas `secured_router`'s scheduler pins handles until process exit.
689. 689pub fn gateway_router(core: Core) -> Router {
690. 690 let mut state = AppState::new(core);
691. 691 state.enable_gateway();
692. 692 router_with_state(state)
693. 693}
694. 694
695. 695fn router_with_state(state: AppState) -> Router {
696. 696 Router::new()
697. 697 .route("/health", get(health))
698. 698 .route("/sessions", get(list_sessions).post(create_session))
699. 699 .route("/sessions/{id}/attach", post(attach_session))
700. 700 .route("/sessions/{id}/diff", get(session_diff))
701. 701 .route("/sessions/{id}/receipts", get(session_receipts))
702. 702 .route(
703. 703 "/sessions/{id}/work",
704. 704 get(session_work).post(session_work_command),
705. 705 )
706. 706 .route("/sessions/{id}/work/confirm", post(session_work_confirm))
707. 707 .route("/sessions/{id}/work/revise", post(session_work_revise))
708. 708 .route(
709. 709 "/sessions/{id}/work/items/{item_id}/retry",
710. 710 post(session_work_retry),
711. 711 )
712. 712 .route(
713. 713 "/sessions/{id}/work/items/{item_id}/cancel",
714. 714 post(session_work_cancel_item),
715. 715 )
716. 716 .route(
717. 717 "/sessions/{id}/work/items/{item_id}/reassign",
718. 718 post(session_work_reassign),
719. 719 )
720. 720 .route("/flows", get(flows_list))
721. 721 .route("/flows/{name}/runs", get(flow_runs_list))
722. 722 .route("/flows/{name}/runs/{run}/graph", get(flow_run_graph))
723. 723 .route("/sessions/{id}/checkpoints", get(list_checkpoints))
724. 724 .route(
725. 725 "/sessions/{id}/checkpoints/{seq}/restore",
726. 726 post(restore_checkpoint),
727. 727 )
728. 728 .route("/sessions/{id}/archive", post(set_archived))
729. 729 .route("/sessions/archived", delete(delete_all_archived))
730. 730 .route("/sessions/{id}", delete(delete_session))
731. 731 .route("/sessions/{id}/restore", post(restore_session))
732. 732 .route("/skills", get(list_skills))
733. 733 .route("/commands", get(list_commands))
734. 734 .route("/plugins", get(list_plugins))
735. 735 .route("/plugins/catalog", get(plugin_catalog))
736. 736 .route("/plugins/audit", get(plugin_audit))
737. 737 .route("/plugins/invocations", get(plugin_invocations))
738. 738 .route(
739. 739 "/presentations",
740. 740 get(list_presentations).post(register_presentations),
741. 741 )
742. 742 .route(
743. 743 "/presentations/specs/{id}/{revision}",
744. 744 get(get_presentation_spec),
745. 745 )
746. 746 .route(
747. 747 "/presentations/revisions",
748. 748 post(propose_presentation_revision),
749. 749 )
750. 750 .route(
751. 751 "/sessions/{id}/presentation/proposals",
752. 752 post(propose_session_presentation_revision),
753. 753 )
754. 754 .route("/presentations/export", get(export_presentations))
755. 755 .route("/presentations/import", post(import_presentations))
756. 756 .route(
757. 757 "/presentations/activate-all",
758. 758 post(activate_all_presentations),
759. 759 )
760. 760 .route(
761. 761 "/presentations/deactivate-all",
762. 762 post(deactivate_all_presentations),
763. 763 )
764. 764 .route(
765. 765 "/presentations/{id}/{revision}/activate",
766. 766 post(activate_presentation),
767. 767 )
768. 768 .route(
769. 769 "/presentations/{id}/deactivate",
770. 770 post(deactivate_presentation),
771. 771 )
772. 772 .route("/presentations/{id}/reset", post(reset_presentation))
773. 773 .route(
774. 774 "/presentations/plugins/{plugin_id}",
775. 775 delete(revoke_presentations_plugin),
776. 776 )
777. 777 .route(
778. 778 "/plugins/retired",
779. 779 get(list_retired_plugins).delete(remove_retired_plugins),
780. 780 )
781. 781 .route(
782. 782 "/plugins/sources",
783. 783 get(plugin_sources).post(plugin_register_source),
784. 784 )
785. 785 .route("/plugins/sources/{id}/enable", post(plugin_source_enable))
786. 786 .route("/plugins/sources/{id}/disable", post(plugin_source_disable))
787. 787 .route("/plugins/keys/{id}/revoke", post(plugin_key_revoke))
788. 788 .route("/plugins/keys/{id}/restore", post(plugin_key_restore))
789. 789 .route("/plugins/install", post(plugin_install))
790. 790 .route("/plugins/update", post(plugin_update))
791. 791 .route("/plugins/{name}/enable", post(plugin_enable))
792. 792 .route("/plugins/{name}/disable", post(plugin_disable))
793. 793 .route("/plugins/{name}/rollback", post(plugin_rollback))
794. 794 .route("/plugins/{name}", delete(plugin_remove))
795. 795 .route("/sessions/{id}/pr", get(session_pr))
796. 796 .route("/sessions/{id}/pr/merge", post(pr_merge))
797. 797 .route("/tasks", get(list_tasks).post(create_task))
798. 798 .route(
799. 799 "/tasks/{id}",
800. 800 axum::routing::patch(patch_task).delete(delete_task),
801. 801 )
802. 802 .route("/tasks/{id}/run-now", post(run_task_now))
803. 803 .route("/tasks/{id}/retry-delivery", post(retry_task_delivery))
804. 804 .route("/sessions/{id}/launch", get(get_launch))
805. 805 .route("/sessions/{id}/launch/prepare", post(prepare_launch))
806. 806 .route("/sessions/{id}/launch/start", post(start_launch))
807. 807 .route("/sessions/{id}/launch/stop", post(stop_launch))
808. 808 .route("/sessions/{id}/launch/logs", get(launch_logs))
809. 809 .route("/sessions/{id}/run", post(run_prompt))
810. 810 .route("/sessions/{id}/steering", post(send_steering))
811. 811 .route("/sessions/{id}/cancel", post(cancel_run))
812. 812 .route("/sessions/{id}/pause", post(pause_run))
813. 813 .route("/sessions/{id}/resume", post(resume_run))
814. 814 .route("/sessions/{id}/control-state", get(control_state))
815. 815 .route("/sessions/{id}/plan-change", post(plan_change))
816. 816 .route("/sessions/{id}/workers", get(list_workers))
817. 817 .route("/sessions/{id}/workers/{child}/steer", post(steer_worker))
818. 818 .route("/sessions/{id}/workers/{child}/stop", post(stop_worker))
819. 819 // Backward-compatible aliases for the old `subagents` route names.
820. 820 .route("/sessions/{id}/subagents", get(list_workers))
821. 821 .route("/sessions/{id}/subagents/{child}/steer", post(steer_worker))
822. 822 .route("/sessions/{id}/subagents/{child}/stop", post(stop_worker))
823. 823 .route("/sessions/{id}/approvals/{req_id}", post(answer_approval))
824. 824 .route("/sessions/{id}/outcome-review", post(record_outcome_review))
825. 825 .route("/sessions/{id}/events", get(events_sse))
826. 826 .route(
827. 827 "/sessions/{id}/sandbox/executions",
828. 828 get(session_sandbox_executions),
829. 829 )
830. 830 .route(
831. 831 "/sessions/{id}/sandbox/records",
832. 832 get(list_session_sandbox_records),
833. 833 )
834. 834 .route(
835. 835 "/sessions/{id}/sandbox/candidates",
836. 836 post(export_sandbox_candidate),
837. 837 )
838. 838 .route(
839. 839 "/sessions/{id}/sandbox/candidates/{candidate_id}/files",
840. 840 get(read_sandbox_candidate_file),
841. 841 )
842. 842 .route(
843. 843 "/sessions/{id}/sandbox/candidates/{candidate_id}/files/raw",
844. 844 get(read_sandbox_candidate_file_raw),
845. 845 )
846. 846 .route(
847. 847 "/sessions/{id}/sandbox/candidates/{candidate_id}/office-review",
848. 848 get(read_sandbox_candidate_office_review),
849. 849 )
850. 850 .route(
851. 851 "/sessions/{id}/sandbox/candidates/{candidate_id}/office-narrow",
852. 852 post(narrow_sandbox_candidate_office),
853. 853 )
854. 854 .route(
855. 855 "/sessions/{id}/sandbox/candidates/{candidate_id}/office",
856. 856 get(read_sandbox_candidate_office_projection),
857. 857 )
858. 858 .route(
859. 859 "/sessions/{id}/sandbox/candidates/{candidate_id}/comments",
860. 860 get(list_sandbox_candidate_comments).post(comment_on_sandbox_candidate),
861. 861 )
862. 862 .route(
863. 863 "/sessions/{id}/sandbox/candidates/{candidate_id}/comments/{comment_id}/request-revision",
864. 864 post(request_revision_from_candidate_comment),
865. 865 )
866. 866 .route(
867. 867 "/sessions/{id}/sandbox/promote",
868. 868 post(promote_sandbox_candidate),
869. 869 )
870. 870 .route(
871. 871 "/sessions/{id}/sandbox/promotions/{candidate_id}/undo",
872. 872 post(undo_sandbox_promotion),
873. 873 )
874. 874 .route(
875. 875 "/sessions/{id}/sandbox/promotions/{candidate_id}/checks",
876. 876 post(run_sandbox_workspace_check),
877. 877 )
878. 878 .route("/sessions/{id}/presentation", get(presentation_snapshot))
879. 879 .route("/sessions/{id}/results/{result_id}", get(session_result))
880. 880 .route(
881. 881 "/sessions/{id}/presentation/feedback",
882. 882 post(presentation_feedback),
883. 883 )
884. 884 .route(
885. 885 "/sessions/{id}/presentation/select",
886. 886 post(select_presentation_for_session),
887. 887 )
888. 888 .route(
889. 889 "/sessions/{id}/presentation/events",
890. 890 get(presentation_events_sse),
891. 891 )
892. 892 .route("/sessions/{id}/transcript", get(transcript))
893. 893 .route("/sessions/{id}/transcript.md", get(transcript_markdown))
894. 894 .route(
895. 895 "/sessions/{id}/coworking/invitations",
896. 896 get(list_coworking_invitations).post(create_coworking_invitation),
897. 897 )
898. 898 .route("/sessions/{id}/coworking/me", get(coworking_me))
899. 899 .route(
900. 900 "/sessions/{id}/coworking/presence",
901. 901 get(coworking_presence),
902. 902 )
903. 903 .route(
904. 904 "/sessions/{id}/coworking/messages",
905. 905 post(create_coworking_message),
906. 906 )
907. 907 .route(
908. 908 "/sessions/{id}/coworking/approvals",
909. 909 get(list_coworking_approvals),
910. 910 )
911. 911 .route(
912. 912 "/sessions/{id}/coworking/approvals/{req_id}",
913. 913 post(answer_coworking_approval),
914. 914 )
915. 915 .route(
916. 916 "/sessions/{id}/coworking/approvals/{req_id}/delegate",
917. 917 post(delegate_coworking_approval),
918. 918 )
919. 919 .route("/sessions/{id}/coworking/updates", get(coworking_updates))
920. 920 .route("/sessions/{id}/office-workspaces", get(office_workspace::list).post(office_workspace::create))
921. 921 .route("/sessions/{id}/office-workspaces/{room_id}", post(office_workspace::mutate))
922. 922 .route("/sessions/{id}/office-workspaces/{room_id}/presence", post(office_workspace::focus))
923. 923 .route(
924. 924 "/sessions/{id}/coworking/invitations/{grant_id}/revoke",
925. 925 post(revoke_coworking_invitation),
926. 926 )
927. 927 .route("/sessions/{id}/side", post(side_chat))
928. 928 .route("/sessions/{id}/side/cancel", post(side_cancel_run))
929. 929 .route("/sessions/{id}/bestofn", post(start_bestofn))
930. 930 .route("/sessions/{id}/keep", post(keep_best_run))
931. 931 .route("/sessions/{id}/discard", post(discard_best_run))
932. 932 .route("/fs/file", get(read_file).put(write_file))
933. 933 .route("/fs/file/raw", get(read_file_raw))
934. 934 .route("/fs/office", get(read_office_projection))
935. 935 .route(
936. 936 "/fs/inbox",
937. 937 post(upload_to_inbox).layer(axum::extract::DefaultBodyLimit::max(
938. 938 inbox::UPLOAD_MAX_BYTES,
939. 939 )),
940. 940 )
941. 941 .route("/fs/preview/{*path}", get(preview_file))
942. 942 .route("/sandbox/records", get(list_sandbox_records))
943. 943 .route("/fs/tree", get(fs_tree))
944. 944 .route("/config", get(get_config).patch(patch_config))
945. 945 .route("/config/intent/evidence", post(patch_evidence_policy))
946. 946 .route(
947. 947 "/config/global",
948. 948 get(get_global_config_layer).patch(patch_global_config),
949. 949 )
950. 950 .route("/config/workspace", get(get_workspace_config_layer))
951. 951 .route("/config/project", get(get_workspace_config_layer))
952. 952 .route("/config/mode", post(set_permission_mode))
953. 953 .route(
954. 954 "/agent-network/capabilities",
955. 955 post(agent_network_capability),
956. 956 )
957. 957 .route(
958. 958 "/agent-network/messages",
959. 959 post(agent_network_send).get(agent_network_receive),
960. 960 )
961. 961 .route("/config/mcp", get(get_mcp_servers).put(put_mcp_servers))
962. 962 .route(
963. 963 "/config/mcp/global",
964. 964 get(get_global_mcp_servers).put(put_global_mcp_servers),
965. 965 )
966. 966 .route("/config/integrations", get(get_integration_catalog))
967. 967 .route(
968. 968 "/config/integrations/{id}",
969. 969 get(get_scoped_integration)
970. 970 .put(put_scoped_integration)
971. 971 .delete(delete_scoped_integration),
972. 972 )
973. 973 .route("/config/hooks", get(get_hooks).put(put_hooks))
974. 974 .route(
975. 975 "/config/prompts",
976. 976 get(get_prompt_layer).put(put_prompt_block),
977. 977 )
978. 978 .route("/config/prompts/effective", get(get_prompt_effective))
979. 979 .route("/config/prompts/preview", post(preview_prompt))
980. 980 .route("/config/prompts/roles", get(list_prompt_roles))
981. 981 .route("/agents", get(agent_chats::list))
982. 982 .route("/agents/{agent}/open", post(agent_chats::open))
983. 983 .route("/config/agents", get(get_agents).put(put_agents))
984. 984 .route(
985. 985 "/config/hooks/global",
986. 986 get(get_global_hooks).put(put_global_hooks),
987. 987 )
988. 988 .route(
989. 989 "/config/key",
990. 990 put(put_provider_key).delete(delete_provider_key),
991. 991 )
992. 992 // Distributed event bus status (vak-bus, docs/design/53).
993. 993 // Credentials are never returned; only the connection state and
994. 994 // metrics are exposed.
995. 995 .route(
996. 996 "/config/bus",
997. 997 get(get_bus_config)
998. 998 .put(put_bus_config)
999. 999 .delete(delete_bus_config),
1000. 1000 )