1. 1use crate::{AppState, agents, register_handle};
2. 2use axum::{
3. 3 Json,
4. 4 extract::{Path, State},
5. 5 http::StatusCode,
6. 6 response::{IntoResponse, Response},
7. 7};
8. 8use serde::Deserialize;
9. 9use std::io::BufRead;
10. 10use vak_session::types::{
11. 11 AgentIdentity, ConversationContext, ConversationOrigin, Entry, EntryPayload, SessionHeader,
12. 12};
13. 13
14. 14pub(crate) fn header(path: &std::path::Path) -> Result<SessionHeader, String> {
15. 15 let file = std::fs::File::open(path).map_err(|e| e.to_string())?;
16. 16 let line = std::io::BufReader::new(file)
17. 17 .lines()
18. 18 .next()
19. 19 .ok_or("empty session ledger")?
20. 20 .map_err(|e| e.to_string())?;
21. 21 match serde_json::from_str::<Entry>(&line)
22. 22 .map_err(|e| e.to_string())?
23. 23 .payload
24. 24 {
25. 25 EntryPayload::Header(header) => Ok(header),
26. 26 _ => Err("session ledger has no header".into()),
27. 27 }
28. 28}
29. 29
30. 30fn error(status: StatusCode, message: impl ToString) -> Response {
31. 31 (
32. 32 status,
33. 33 Json(serde_json::json!({"error": message.to_string()})),
34. 34 )
35. 35 .into_response()
36. 36}
37. 37
38. 38pub(crate) async fn list(State(state): State<AppState>) -> Response {
39. 39 match agents::effective(&state.active_core()) {
40. 40 Ok(agents) => Json(serde_json::json!({"agents": agents})).into_response(),
41. 41 Err(e) => error(StatusCode::INTERNAL_SERVER_ERROR, e),
42. 42 }
43. 43}
44. 44
45. 45/// Identity lookup plus admissibility check by id alone, without deriving a
46. 46/// workspace path. Shared by [`resolve_agent_core`] (a fresh or looked-up
47. 47/// conversation, which still needs to compute a workspace) and
48. 48/// `crate::resolve_core_for_header` (an existing session, which already
49. 49/// knows its own workspace from its header and must never let a re-derived
50. 50/// path disagree with it — see that function's doc comment).
51. 51#[allow(clippy::result_large_err)]
52. 52pub(crate) fn resolve_agent_identity(
53. 53 active: &vak_core::Core,
54. 54 id: &str,
55. 55) -> Result<AgentIdentity, Response> {
56. 56 if id == "vak" {
57. 57 return Ok(AgentIdentity {
58. 58 id: id.to_string(),
59. 59 revision: 1,
60. 60 name: "Vakyartha".into(),
61. 61 character: "vak".into(),
62. 62 personality: String::new(),
63. 63 animation: "subtle".into(),
64. 64 voice: "default".into(),
65. 65 behaviour: String::new(),
66. 66 responsibilities: String::new(),
67. 67 instructions: String::new(),
68. 68 });
69. 69 }
70. 70 let profiles =
71. 71 agents::effective(active).map_err(|e| error(StatusCode::INTERNAL_SERVER_ERROR, e))?;
72. 72 let Some(profile) = profiles.into_iter().find(|p| p.id == id) else {
73. 73 return Err(error(
74. 74 StatusCode::NOT_FOUND,
75. 75 "This agent is no longer available.",
76. 76 ));
77. 77 };
78. 78 if !profile.is_admissible() {
79. 79 return Err(error(
80. 80 StatusCode::CONFLICT,
81. 81 "This Agent is paused or archived.",
82. 82 ));
83. 83 }
84. 84 Ok(profile.identity())
85. 85}
86. 86
87. 87/// Resolve an Agent id to its identity and its own isolated `Core` — the
88. 88/// single place this resolution happens, so every endpoint that needs "this
89. 89/// Agent's own data" (workspace, sessions_home, runtime safety pins) goes
90. 90/// through the same logic `open` uses, rather than each one re-deriving or
91. 91/// (worse) silently falling back to the process's default workspace. Any
92. 92/// new endpoint scoped to a specific Agent should call this rather than
93. 93/// reading `state.core`/`state.active_core()` directly.
94. 94#[allow(clippy::result_large_err)]
95. 95pub(crate) fn resolve_agent_core(
96. 96 state: &AppState,
97. 97 id: &str,
98. 98) -> Result<(AgentIdentity, vak_core::Core), Response> {
99. 99 let active = state.active_core();
100. 100 let identity = resolve_agent_identity(&active, id)?;
101. 101 // Each user-created Agent gets its own isolated project workspace (files,
102. 102 // tool access, permissions) rather than sharing the process's default
103. 103 // workspace — resolved through the same `CorePool` a channel/gateway
104. 104 // workspace switch uses, so trust/permission/sandbox resolution is
105. 105 // identical to a local run rooted there. The built-in "vak" agent keeps
106. 106 // the process's own workspace for backward compatibility.
107. 107 //
108. 108 // The base workspace must match whichever root `agents::save` used to
109. 109 // persist this profile (`agents.rs` saves "user"-scope agents under
110. 110 // `default_workspace()`, "workspace"-scope under the saving request's
111. 111 // own cwd) — otherwise, whenever the active core points somewhere other
112. 112 // than `default_workspace()` (a browser workspace switch, a gateway
113. 113 // channel), a "user"-scope agent's precreated directory and its actual
114. 114 // runtime workspace would silently diverge. `agents::effective` already
115. 115 // gives the workspace layer precedence over the shared layer for a
116. 116 // duplicate id, so mirror that precedence here.
117. 117 let default_root = vak_config::paths::default_workspace();
118. 118 let base = if identity.id == "vak" {
119. 119 active.cwd().clone()
120. 120 } else {
121. 121 let is_workspace_scoped = active.cwd() != &default_root
122. 122 && agents::load(active.cwd())
123. 123 .unwrap_or_default()
124. 124 .iter()
125. 125 .any(|p| p.id == identity.id);
126. 126 if is_workspace_scoped {
127. 127 active.cwd().clone()
128. 128 } else {
129. 129 default_root
130. 130 }
131. 131 };
132. 132 let workspace = vak_config::paths::agent_workspace(&base, &identity.id);
133. 133 // `agents::save` already creates this directory once at agent-creation
134. 134 // time; opening an agent is idempotent and hit repeatedly (reload, tab
135. 135 // switch, reconnect), so skip the mkdir once it's confirmed to exist
136. 136 // rather than paying the syscalls on every open.
137. 137 if !workspace.is_dir()
138. 138 && let Err(e) = std::fs::create_dir_all(&workspace)
139. 139 {
140. 140 return Err(error(StatusCode::INTERNAL_SERVER_ERROR, e));
141. 141 }
142. 142 let core = pinned_core_for_workspace(state, &active, &identity, &workspace)
143. 143 .map_err(|e| error(StatusCode::INTERNAL_SERVER_ERROR, e))?;
144. 144 Ok((identity, core))
145. 145}
146. 146
147. 147/// Resolve a `Core` pinned exactly the way `/agents/{id}/open` pins one:
148. 148/// runtime permission-mode cap, sandbox backend override, provider instance
149. 149/// override, and shared sessions_home carried forward from `active`, plus
150. 150/// the one-time trust backstop for a workspace that predates `agents::save`
151. 151/// recording trust.
152. 152///
153. 153/// Takes the target `workspace` directly rather than re-deriving it from
154. 154/// `active.cwd()` (finding 6): `resolve_agent_core` still has to compute a
155. 155/// workspace path for a fresh-or-looked-up conversation, but a session that
156. 156/// already exists knows its own workspace from its header, and re-deriving
157. 157/// one from whatever happens to be the CURRENT active workspace can
158. 158/// disagree with it once the active workspace has moved on. Before this,
159. 159/// `/attach`, `/run` and the SSE endpoints resolved a plain pooled `Core`
160. 160/// with none of these pins, so the same session's security ceiling
161. 161/// depended on which endpoint touched it first.
162. 162pub(crate) fn pinned_core_for_workspace(
163. 163 state: &AppState,
164. 164 active: &vak_core::Core,
165. 165 identity: &AgentIdentity,
166. 166 workspace: &std::path::Path,
167. 167) -> Result<vak_core::Core, vak_core::CoreError> {
168. 168 // Backstop for an Agent whose workspace predates `agents::save` carrying
169. 169 // trust forward (or was created by some other path this fix missed):
170. 170 // without a trust marker here, `CorePool::resolve_at` below treats it as
171. 171 // untrusted and silently strips its own `permission_mode`, `hooks`,
172. 172 // `mcp.servers`, and other privileged config forever — the same gap
173. 173 // `agents::save` closes at creation time, applied retroactively the
174. 174 // first time this Agent is opened from a trusted context.
175. 175 if identity.id != "vak"
176. 176 && active.project_config_trusted()
177. 177 && !vak_core::trust::is_trusted(workspace)
178. 178 {
179. 179 let _ = vak_core::trust::mark_trusted(workspace);
180. 180 }
181. 181 let core = if workspace == active.cwd().as_path() {
182. 182 active.clone()
183. 183 } else {
184. 184 let resolved =
185. 185 match state
186. 186 .gateway
187. 187 .core_pool
188. 188 .resolve_at(workspace, None, std::time::Instant::now())
189. 189 {
190. 190 Ok(core) => core,
191. 191 // `resolve_at` failing (a transient permission-ceiling recheck
192. 192 // error on a cache hit, or `Core::new_with_trust`'s own IO/config
193. 193 // error) is not itself a trust decision — falling back to an
194. 194 // unconditional `true` here would let an operator-declined
195. 195 // workspace's hooks/MCP servers/secret scope apply anyway, exactly the
196. 196 // bypass `vak_core::trust` exists to close. Recompute trust the
197. 197 // same way `resolve_at` does rather than assuming it.
198. 198 Err(_) => vak_core::Core::new_with_trust(
199. 199 workspace.to_path_buf(),
200. 200 vak_core::trust::is_trusted(workspace),
201. 201 )?,
202. 202 };
203. 203 // A freshly-resolved Core has its own default sessions/data home
204. 204 // (real on-disk `data_home()`), which would silently diverge from
205. 205 // wherever this app/process's data actually lives if the active
206. 206 // Core was pointed at a non-default root (test isolation, or a
207. 207 // future custom data-home setting). Every agent's data must live
208. 208 // under the *same* root, just in its own agent-scoped subdirectory
209. 209 // (`Core::sessions_home` already layers that on top).
210. 210 resolved.set_sessions_home(active.shared_data_home());
211. 211 if let Some(provider) = active.provider_instance_override() {
212. 212 resolved.set_provider_instance(provider);
213. 213 }
214. 214 // A user-pinned safety ceiling (e.g. read-only mode, or a hardened
215. 215 // sandbox backend) is a this-session/this-app control, not a
216. 216 // per-project-directory config value — it must not silently loosen
217. 217 // the moment a different Agent's Core is resolved from that
218. 218 // workspace's own on-disk config. Carry the pin forward the same
219. 219 // way a persisted config value already is via `sessions_home`.
220. 220 if let Some(mode) = active.permission_mode_override_value() {
221. 221 // Cap against this workspace's own resolved ceiling, the same
222. 222 // way a per-channel override is capped in `core_pool.rs` — a
223. 223 // pin from a more-permissive workspace must never grant more
224. 224 // access than this agent's own config already allows.
225. 225 let ceiling = resolved.effective_permission_mode();
226. 226 resolved.set_permission_mode(mode.capped_by(ceiling));
227. 227 }
228. 228 if let Some(backend) = active.sandbox_backend_override_value() {
229. 229 resolved.set_sandbox_backend(Some(backend));
230. 230 }
231. 231 resolved
232. 232 };
233. 233 Ok(core.with_agent_identity(Some(identity.clone())))
234. 234}
235. 235
236. 236#[derive(Deserialize, Default)]
237. 237pub(crate) struct OpenAgentRequest {
238. 238 #[serde(default)]
239. 239 create_new: bool,
240. 240}
241. 241
242. 242/// One resolved conversation, cached so a repeated open (every reload, tab
243. 243/// switch, reconnect — this endpoint is hit constantly) can skip the
244. 244/// admission lock and directory scan entirely.
245. 245#[derive(Clone)]
246. 246struct CachedAgentSession {
247. 247 session_id: String,
248. 248 ledger_path: std::path::PathBuf,
249. 249 agent: AgentIdentity,
250. 250}
251. 251
252. 252/// Keyed by (workspace, conversation id) rather than just the Agent id: the
253. 253/// built-in `vak` agent alone can have one conversation per workspace, and
254. 254/// `create_new` mints a fresh conversation id every time so it never
255. 255/// collides with — or evicts — the durable one. Revalidated by file
256. 256/// existence on every read (below), so a stale entry pointing at a deleted
257. 257/// ledger is never trusted, only discarded.
258. 258static AGENT_SESSION_CACHE: std::sync::OnceLock<
259. 259 std::sync::Mutex<std::collections::HashMap<(std::path::PathBuf, String), CachedAgentSession>>,
260. 260> = std::sync::OnceLock::new();
261. 261
262. 262fn agent_session_cache() -> &'static std::sync::Mutex<
263. 263 std::collections::HashMap<(std::path::PathBuf, String), CachedAgentSession>,
264. 264> {
265. 265 AGENT_SESSION_CACHE.get_or_init(|| std::sync::Mutex::new(std::collections::HashMap::new()))
266. 266}
267. 267
268. 268/// Whether a ledger has any entry beyond its header — decided from file
269. 269/// size against the header line's own byte length, never by reading and
270. 270/// counting every line (that read the WHOLE file just to answer a yes/no
271. 271/// question, for every candidate, on every open).
272. 272fn has_entries_beyond_header(path: &std::path::Path) -> bool {
273. 273 let Ok(file) = std::fs::File::open(path) else {
274. 274 return false;
275. 275 };
276. 276 let mut reader = std::io::BufReader::new(file);
277. 277 let mut first_line = String::new();
278. 278 let Ok(read) = reader.read_line(&mut first_line) else {
279. 279 return false;
280. 280 };
281. 281 if read == 0 {
282. 282 return false;
283. 283 }
284. 284 let Ok(metadata) = std::fs::metadata(path) else {
285. 285 return false;
286. 286 };
287. 287 metadata.len() > read as u64
288. 288}
289. 289
290. 290/// The directory scan that used to run inline on the async handler
291. 291/// (blocking `std::fs` calls, one `header()` read per ledger). Runs inside
292. 292/// `spawn_blocking`; an unreadable or empty ledger is skipped with a
293. 293/// warning rather than failing the whole request — a fleet of conversations
294. 294/// must not go dark because one file next to them is corrupt.
295. 295fn scan_candidates(
296. 296 dir: std::path::PathBuf,
297. 297 cwd: std::path::PathBuf,
298. 298 agent_id: String,
299. 299 conversation: ConversationContext,
300. 300 trashed: std::collections::HashSet<String>,
301. 301) -> Vec<(SessionHeader, bool)> {
302. 302 let mut candidates = Vec::new();
303. 303 let Ok(entries) = std::fs::read_dir(&dir) else {
304. 304 return candidates;
305. 305 };
306. 306 for entry in entries.flatten() {
307. 307 if entry.path().extension().and_then(|s| s.to_str()) != Some("jsonl") {
308. 308 continue;
309. 309 }
310. 310 let h = match header(&entry.path()) {
311. 311 Ok(h) => h,
312. 312 Err(e) => {
313. 313 eprintln!(
314. 314 "[agents] skipping unreadable session ledger {}: {e}",
315. 315 entry.path().display()
316. 316 );
317. 317 continue;
318. 318 }
319. 319 };
320. 320 if h.cwd == cwd
321. 321 && !trashed.contains(&h.session_id)
322. 322 && h.parent_session_id.is_none()
323. 323 && h.agent.as_ref().is_some_and(|a| a.id == agent_id)
324. 324 && h.conversation.as_ref() == Some(&conversation)
325. 325 {
326. 326 let has_content = has_entries_beyond_header(&entry.path());
327. 327 candidates.push((h, has_content));
328. 328 }
329. 329 }
330. 330 candidates
331. 331}
332. 332
333. 333fn opened_response(session_id: &str, agent: &AgentIdentity, core: &vak_core::Core) -> Response {
334. 334 Json(serde_json::json!({"session_id": session_id, "agent": agent, "cwd": core.cwd()}))
335. 335 .into_response()
336. 336}
337. 337
338. 338/// Reopen an already-known session into the live handle map if it is not
339. 339/// there already. `Err` means the ledger could not actually be (re)opened
340. 340/// even though its file exists (e.g. removed a moment ago, or held
341. 341/// exclusively elsewhere in a way `open_session_read_only` also refuses) —
342. 342/// callers either fall back to a fresh scan or surface the reason, but
343. 343/// never silently answer 200 for a session that was not actually attached.
344. 344async fn ensure_registered(
345. 345 state: &AppState,
346. 346 core: &vak_core::Core,
347. 347 session_id: &str,
348. 348) -> Result<(), String> {
349. 349 if state.get(session_id).is_some() {
350. 350 return Ok(());
351. 351 }
352. 352 let session = match core.open_session(session_id).await {
353. 353 Ok(session) => session,
354. 354 Err(vak_core::CoreError::Session(vak_session::SessionError::Locked(_))) => {
355. 355 match core.open_session_read_only(session_id).await {
356. 356 Ok(session) => session,
357. 357 Err(e) => return Err(format!("session is locked elsewhere: {e}")),
358. 358 }
359. 359 }
360. 360 Err(e) => return Err(e.to_string()),
361. 361 };
362. 362 register_handle(
363. 363 state,
364. 364 session_id.to_string(),
365. 365 session,
366. 366 core.cwd().clone(),
367. 367 core.clone(),
368. 368 );
369. 369 Ok(())
370. 370}
371. 371
372. 372/// Opening an existing Agent conversation is idempotent. An explicit new
373. 373/// conversation gets its own identity and append-only session ledger.
374. 374pub(crate) async fn open(
375. 375 State(state): State<AppState>,
376. 376 Path(id): Path<String>,
377. 377 Json(request): Json<OpenAgentRequest>,
378. 378) -> Response {
379. 379 let (identity, core) = match resolve_agent_core(&state, &id) {
380. 380 Ok(pair) => pair,
381. 381 Err(response) => return response,
382. 382 };
383. 383 // The desktop surface has one durable conversation per selected Agent.
384. 384 // This is deliberately derived from the Agent identity, not from browser
385. 385 // storage or a transient session id, so reopening the same Agent resumes
386. 386 // the same conversation while another Agent gets an independent ledger.
387. 387 let conversation = ConversationContext {
388. 388 conversation_id: if request.create_new {
389. 389 format!("agent:{}:local:{}", identity.id, uuid::Uuid::now_v7())
390. 390 } else {
391. 391 format!("agent:{}:local", identity.id)
392. 392 },
393. 393 audience_id: "local".into(),
394. 394 origin: Some(ConversationOrigin {
395. 395 surface: "desktop".into(),
396. 396 address: "local".into(),
397. 397 bot_id: None,
398. 398 }),
399. 399 };
400. 400 let core = core.with_conversation_context(Some(conversation.clone()));
401. 401 let cache_key = (core.cwd().clone(), conversation.conversation_id.clone());
402. 402
403. 403 let cached = agent_session_cache()
404. 404 .lock()
405. 405 .unwrap_or_else(std::sync::PoisonError::into_inner)
406. 406 .get(&cache_key)
407. 407 .cloned();
408. 408 if let Some(cached) = cached
409. 409 && cached.ledger_path.is_file()
410. 410 {
411. 411 if ensure_registered(&state, &core, &cached.session_id)
412. 412 .await
413. 413 .is_ok()
414. 414 {
415. 415 return opened_response(&cached.session_id, &cached.agent, &core);
416. 416 }
417. 417 // The ledger existed a moment ago but could not actually be
418. 418 // (re)opened; drop the stale entry and fall through to a fresh scan.
419. 419 agent_session_cache()
420. 420 .lock()
421. 421 .unwrap_or_else(std::sync::PoisonError::into_inner)
422. 422 .remove(&cache_key);
423. 423 }
424. 424
425. 425 let dir = vak_session::SessionPath::sessions_dir(&core.sessions_home(), core.cwd());
426. 426 if let Err(e) = std::fs::create_dir_all(&dir) {
427. 427 return error(StatusCode::INTERNAL_SERVER_ERROR, e);
428. 428 }
429. 429 // One short cross-process admission lock per workspace. A contending
430. 430 // request retries explicitly; it cannot create a second conversation.
431. 431 let lock = match std::fs::OpenOptions::new()
432. 432 .create(true)
433. 433 .truncate(false)
434. 434 .read(true)
435. 435 .write(true)
436. 436 .open(dir.join("agent-admission.lock"))
437. 437 {
438. 438 Ok(file) => file,
439. 439 Err(e) => return error(StatusCode::INTERNAL_SERVER_ERROR, e),
440. 440 };
441. 441 if lock.try_lock().is_err() {
442. 442 return error(StatusCode::CONFLICT, "An agent is opening. Try again.");
443. 443 }
444. 444 let mut candidates = match tokio::task::spawn_blocking({
445. 445 let dir = dir.clone();
446. 446 let cwd = core.cwd().clone();
447. 447 let agent_id = identity.id.clone();
448. 448 let conversation = conversation.clone();
449. 449 let trashed = vak_core::trash::trashed(&core.shared_data_home());
450. 450 move || scan_candidates(dir, cwd, agent_id, conversation, trashed)
451. 451 })
452. 452 .await
453. 453 {
454. 454 Ok(candidates) => candidates,
455. 455 Err(e) => return error(StatusCode::INTERNAL_SERVER_ERROR, e),
456. 456 };
457. 457 candidates.sort_by_key(|(h, has_content)| (*has_content, h.created_at));
458. 458 if let Some((h, _)) = candidates.last() {
459. 459 let sid = h.session_id.clone();
460. 460 if let Err(e) = ensure_registered(&state, &core, &sid).await {
461. 461 return error(StatusCode::CONFLICT, e);
462. 462 }
463. 463 // The identity frozen when the conversation was admitted, not the
464. 464 // catalogue's current entry: editing an Agent never rewrites a
465. 465 // conversation it already owns (AGENTS.md invariant 37).
466. 466 let admitted = h.agent.clone().unwrap_or_else(|| identity.clone());
467. 467 agent_session_cache()
468. 468 .lock()
469. 469 .unwrap_or_else(std::sync::PoisonError::into_inner)
470. 470 .insert(
471. 471 cache_key,
472. 472 CachedAgentSession {
473. 473 session_id: sid.clone(),
474. 474 ledger_path: dir.join(format!("{sid}.jsonl")),
475. 475 agent: admitted.clone(),
476. 476 },
477. 477 );
478. 478 return opened_response(&sid, &admitted, &core);
479. 479 }
480. 480 let session = match core.start_session().await {
481. 481 Ok(session) => session,
482. 482 Err(e) => return error(StatusCode::INTERNAL_SERVER_ERROR, e),
483. 483 };
484. 484 let Some(h) = session.header() else {
485. 485 return error(StatusCode::INTERNAL_SERVER_ERROR, "Session has no header");
486. 486 };
487. 487 let sid = h.session_id.clone();
488. 488 register_handle(
489. 489 &state,
490. 490 sid.clone(),
491. 491 session,
492. 492 core.cwd().clone(),
493. 493 core.clone(),
494. 494 );
495. 495 agent_session_cache()
496. 496 .lock()
497. 497 .unwrap_or_else(std::sync::PoisonError::into_inner)
498. 498 .insert(
499. 499 cache_key,
500. 500 CachedAgentSession {
501. 501 session_id: sid.clone(),
502. 502 ledger_path: dir.join(format!("{sid}.jsonl")),
503. 503 agent: identity.clone(),
504. 504 },
505. 505 );
506. 506 opened_response(&sid, &identity, &core)
507. 507}
508. 508