Vakyartha documentation

1//! Admin console API endpoints (docs/design/29-personal-os.md): session
2//! catalog, transcripts, live event SSE, security audit log, and store
3//! management. Mounted under `/admin/api` in `router_with_state`.
4
5use std::path::PathBuf;
6
7use axum::Json;
8use axum::extract::{Path, Query, State};
9use axum::http::StatusCode;
10use axum::response::sse::{Event, KeepAlive, Sse};
11use axum::response::{IntoResponse, Response};
12use serde::{Deserialize, Serialize};
13use std::collections::HashMap;
14use tokio_stream::StreamExt;
15
16use crate::AppState;
17
18pub(crate) const SESSION_COOKIE: &str = "vak_session";
19
20// ---- GET /admin/api/sessions ----------------------------------------------
21
22#[derive(Debug, Deserialize)]
23pub(crate) struct SessionListQuery {
24 pub limit: Option<usize>,
25 pub project: Option<String>,
26 pub agent: Option<String>,
27}
28
29#[derive(Debug, Serialize)]
30pub(crate) struct SessionListItem {
31 pub session_id: String,
32 pub project_hash: String,
33 pub entry_count: usize,
34 pub first_ts: String,
35 pub last_ts: String,
36 /// From the shared `archive.json` (keyed by session id, not scoped to a
37 /// workspace — the same map `/sessions/{id}/archive` reads and writes).
38 pub archived: bool,
39 #[serde(default, skip_serializing_if = "Option::is_none")]
40 pub agent_id: Option<String>,
41}
42
43fn map_session_agents(shared: &std::path::Path) -> HashMap<String, String> {
44 let mut map = HashMap::new();
45 if let Ok(agents) = std::fs::read_dir(shared.join("agents")) {
46 for agent in agents.flatten() {
47 let agent_id = agent.file_name().to_string_lossy().into_owned();
48 let agent_sessions = agent.path().join("sessions");
49 if agent_sessions.exists() {
50 for e in walkdir::WalkDir::new(&agent_sessions)
51 .max_depth(3)
52 .into_iter()
53 .flatten()
54 {
55 let path = e.path();
56 let maybe_stem = (e.file_type().is_file()
57 && path.extension().is_some_and(|ext| ext == "jsonl"))
58 .then(|| path.file_stem().and_then(|s| s.to_str()))
59 .flatten();
60 if let Some(stem) = maybe_stem {
61 map.insert(stem.to_string(), agent_id.clone());
62 }
63 }
64 }
65 }
66 }
67 map
68}
69
70pub(crate) async fn list_sessions_admin(
71 State(state): State<AppState>,
72 Query(q): Query<SessionListQuery>,
73) -> Response {
74 let Some(store) = &state.store else {
75 return (
76 StatusCode::SERVICE_UNAVAILABLE,
77 Json(serde_json::json!({ "error": "store not available" })),
78 )
79 .into_response();
80 };
81 let limit = q.limit.unwrap_or(50).clamp(1, 200);
82 // Both maps live under the shared vak home (`sessions_home`), not a
83 // per-workspace directory, so they apply across every project this
84 // store indexes — unlike archive/delete *mutation*, which only reaches
85 // a ledger file under this process's own workspace (see
86 // `workspace_project_hash` on `/admin/api/config`).
87 let archive_map = crate::read_archive(&state.core);
88 let shared = state.core.shared_data_home();
89 let trashed = vak_core::trash::trashed(&shared);
90 let agent_map = map_session_agents(&shared);
91
92 match store.list_sessions() {
93 Ok(all) => {
94 let visible = all.into_iter().filter(|s| {
95 let s_agent = agent_map
96 .get(&s.session_id)
97 .map(String::as_str)
98 .unwrap_or("vak");
99 q.project.as_ref().is_none_or(|p| &s.project_hash == p)
100 && q.agent.as_ref().is_none_or(|a| a == "all" || s_agent == a)
101 && !trashed.contains(&s.session_id)
102 });
103 let visible: Vec<_> = visible.collect();
104 let total = visible.len();
105 let items: Vec<SessionListItem> = visible
106 .into_iter()
107 .take(limit)
108 .map(|s| {
109 let agent_id = agent_map
110 .get(&s.session_id)
111 .cloned()
112 .or_else(|| Some("vak".to_string()));
113 SessionListItem {
114 archived: archive_map.get(&s.session_id).copied().unwrap_or(false),
115 session_id: s.session_id,
116 project_hash: s.project_hash,
117 entry_count: s.entry_count,
118 first_ts: s.first_ts,
119 last_ts: s.last_ts,
120 agent_id,
121 }
122 })
123 .collect();
124 Json(serde_json::json!({ "sessions": items, "total": total })).into_response()
125 }
126 Err(e) => (
127 StatusCode::INTERNAL_SERVER_ERROR,
128 Json(serde_json::json!({ "error": e.to_string() })),
129 )
130 .into_response(),
131 }
132}
133
134// ---- GET /admin/api/approvals ----------------------------------------------
135
136#[derive(Debug, Serialize)]
137pub(crate) struct PendingApproval {
138 pub session_id: String,
139 pub request_id: String,
140 pub tool: String,
141 pub args_json: String,
142 pub reason: String,
143 pub requested_at: String,
144}
145
146/// Aggregated pending approval gates across every live session. Answering
147/// still goes through the per-session endpoint — listing never crosses a
148/// session's approval scope, only displays it.
149pub(crate) async fn list_pending_approvals(
150 State(state): State<AppState>,
151) -> Json<serde_json::Value> {
152 let mut items: Vec<PendingApproval> = Vec::new();
153 for handle in state.live_handles() {
154 let pending = handle
155 .pending
156 .lock()
157 .unwrap_or_else(std::sync::PoisonError::into_inner);
158 for req in pending.values() {
159 items.push(PendingApproval {
160 session_id: handle.id.clone(),
161 request_id: req.id.clone(),
162 tool: req.tool.clone(),
163 args_json: truncate_chars(&req.args_json, 400),
164 reason: req.reason.clone(),
165 requested_at: req.requested_at.to_rfc3339(),
166 });
167 }
168 }
169 // Oldest first — the gate that has been waiting longest is the most urgent.
170 items.sort_by(|a, b| a.requested_at.cmp(&b.requested_at));
171 let total = items.len();
172 Json(serde_json::json!({ "approvals": items, "total": total }))
173}
174
175// ---- GET /admin/api/sessions/:id/transcript --------------------------------
176
177#[derive(Debug, Deserialize)]
178pub(crate) struct TranscriptQuery {
179 pub limit: Option<usize>,
180 pub offset: Option<usize>,
181 pub kind: Option<String>,
182 pub role: Option<String>,
183 /// Re-import this session's JSONL into the index before reading, so
184 /// entries appended by an active run become visible immediately.
185 pub refresh: Option<bool>,
186}
187
188/// Byte-safe truncation: never splits a multi-byte UTF-8 sequence.
189fn truncate_chars(s: &str, max_bytes: usize) -> String {
190 if s.len() <= max_bytes {
191 return s.to_string();
192 }
193 let mut end = max_bytes;
194 while end > 0 && !s.is_char_boundary(end) {
195 end -= 1;
196 }
197 format!("{}…", &s[..end])
198}
199
200/// Which ledger entries of a session are runtime-authored control messages,
201/// by entry id (empty when the session cannot be read).
202fn control_kinds_by_entry(
203 state: &AppState,
204 session_id: &str,
205) -> std::collections::HashMap<String, vak_intent::control::ControlKind> {
206 let from = |session: &vak_session::SessionLog| {
207 session
208 .derive_transcript()
209 .into_iter()
210 .filter_map(|item| item.control.map(|kind| (item.entry_id, kind)))
211 .collect()
212 };
213 if let Some(handle) = state.get(session_id)
214 && let Some(session) = handle
215 .session
216 .lock()
217 .unwrap_or_else(std::sync::PoisonError::into_inner)
218 .as_ref()
219 {
220 return from(session);
221 }
222 crate::open_historical_session(state, session_id)
223 .map(|session| from(&session))
224 .unwrap_or_default()
225}
226
227pub(crate) async fn session_transcript_admin(
228 State(state): State<AppState>,
229 Path(session_id): Path<String>,
230 Query(q): Query<TranscriptQuery>,
231) -> Json<serde_json::Value> {
232 let Some(store) = &state.store else {
233 return Json(serde_json::json!({ "error": "store not available" }));
234 };
235 if vak_core::trash::is_trashed(&state.core.shared_data_home(), &session_id) {
236 return Json(serde_json::json!({ "error": "session is in the trash" }));
237 }
238 // Fetch offset+limit so we can report whether more pages exist.
239 let limit = q.limit.unwrap_or(100).clamp(1, 500);
240 let offset = q.offset.unwrap_or(0);
241 if q.refresh == Some(true) {
242 crate::import_session_sync(store, &state.core.sessions_home(), &session_id);
243 }
244 let filter = vak_store::query::SearchFilter {
245 session_id: Some(session_id.clone()),
246 kind: q.kind,
247 role: q.role,
248 ..Default::default()
249 };
250 match store.query_page(&filter, limit, offset, true) {
251 Ok((entries, total)) => {
252 let has_more = offset.saturating_add(entries.len()) < total;
253 // The search index stores text only, so it cannot say which user
254 // rows the runtime authored. The ledger can: tag each row from it.
255 let controls = control_kinds_by_entry(&state, &session_id);
256 let page: Vec<serde_json::Value> = entries
257 .iter()
258 .map(|e| {
259 serde_json::json!({
260 "entry_id": e.entry_id,
261 "ts": e.ts,
262 "kind": e.kind.as_str(),
263 "role": e.role,
264 "tool_name": e.tool_name,
265 "is_error": e.is_error,
266 "content": truncate_chars(&e.content_text, 16000),
267 "control": controls.get(&e.entry_id),
268 })
269 })
270 .collect();
271 let contract = state
272 .get(&session_id)
273 .and_then(|handle| {
274 handle
275 .session
276 .lock()
277 .unwrap_or_else(std::sync::PoisonError::into_inner)
278 .as_ref()
279 .and_then(|session| session.header().map(|header| header.contract.clone()))
280 })
281 .or_else(|| {
282 crate::open_historical_session(&state, &session_id)
283 .and_then(|session| session.header().map(|header| header.contract.clone()))
284 });
285 let configuration_mismatch = false; // Per-turn routing: contract snapshot != mismatch
286 Json(serde_json::json!({
287 "session_id": session_id,
288 "entries": page,
289 "offset": offset,
290 "total": total,
291 "has_more": has_more,
292 "contract": contract,
293 "configuration_mismatch": configuration_mismatch,
294 }))
295 }
296 Err(e) => Json(serde_json::json!({ "error": e.to_string() })),
297 }
298}
299
300// ---- GET /admin/api/search -------------------------------------------------
301
302#[derive(Debug, Deserialize)]
303pub(crate) struct SearchQuery {
304 pub q: String,
305 pub limit: Option<usize>,
306 pub project: Option<String>,
307 pub role: Option<String>,
308 pub kind: Option<String>,
309 pub exclude_session: Option<String>,
310}
311
312pub(crate) async fn search_admin(
313 State(state): State<AppState>,
314 Query(q): Query<SearchQuery>,
315) -> Json<serde_json::Value> {
316 let Some(store) = &state.store else {
317 return Json(serde_json::json!({ "error": "store not available" }));
318 };
319 let limit = q.limit.unwrap_or(20).clamp(1, 100);
320 let filter = vak_store::query::SearchFilter {
321 project_hash: q.project,
322 role: q.role,
323 kind: q.kind,
324 excluded_sessions: vak_core::trash::search_exclusions(
325 &state.core.shared_data_home(),
326 q.exclude_session.as_deref(),
327 )
328 .into_iter()
329 .collect(),
330 ..Default::default()
331 };
332 match store.search(&q.q, limit, &filter) {
333 Ok(result) => Json(serde_json::json!({
334 "hits": result.entries,
335 "total": result.total,
336 })),
337 Err(e) => Json(serde_json::json!({ "error": e.to_string() })),
338 }
339}
340
341// ---- GET /admin/api/events (SSE) ------------------------------------------
342
343pub(crate) async fn admin_events_sse(
344 State(state): State<AppState>,
345) -> Sse<impl tokio_stream::Stream<Item = Result<Event, std::convert::Infallible>>> {
346 use tokio_stream::wrappers::BroadcastStream;
347
348 let rx = state.hub.subscribe();
349 let stream = BroadcastStream::new(rx).filter_map(|result| match result {
350 Ok(event) => {
351 let json = serde_json::to_string(&event).unwrap_or_default();
352 Some(Ok(Event::default().data(json)))
353 }
354 Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(n)) => {
355 let lagged = serde_json::json!({
356 "type": "Lagged",
357 "data": { "missed": n }
358 });
359 let json = serde_json::to_string(&lagged).unwrap_or_default();
360 Some(Ok(Event::default().data(json)))
361 }
362 });
363 Sse::new(stream).keep_alive(KeepAlive::default())
364}
365
366// ---- GET /admin/api/security -----------------------------------------------
367
368#[derive(Debug, Deserialize)]
369pub(crate) struct SecurityQuery {
370 pub limit: Option<usize>,
371 pub kind: Option<String>,
372}
373
374#[derive(Debug, Serialize)]
375struct SecurityEventEntry {
376 ts: String,
377 kind: String,
378 label: String,
379 detail: String,
380 ip: Option<String>,
381}
382
383pub(crate) async fn list_security_events(
384 State(state): State<AppState>,
385 Query(q): Query<SecurityQuery>,
386) -> Json<serde_json::Value> {
387 let limit = q.limit.unwrap_or(100).clamp(1, 500);
388 let home = state.core.sessions_home();
389 let events = vak_core::security_events::list(&home, limit);
390 let filtered: Vec<SecurityEventEntry> = events
391 .into_iter()
392 .filter(|e| {
393 q.kind.as_deref().is_none_or(|k| {
394 serde_json::to_value(&e.kind)
395 .ok()
396 .and_then(|v| v.as_str().map(String::from))
397 .as_deref()
398 == Some(k)
399 })
400 })
401 .map(|e| SecurityEventEntry {
402 ts: e.ts.to_rfc3339(),
403 kind: serde_json::to_value(&e.kind)
404 .ok()
405 .and_then(|v| v.as_str().map(String::from))
406 .unwrap_or_else(|| "unknown".into()),
407 label: e.label,
408 detail: e.detail,
409 ip: e.ip,
410 })
411 .collect();
412 Json(serde_json::json!({
413 "events": filtered,
414 "total": filtered.len(),
415 }))
416}
417
418// ---- POST /admin/api/store/rebuild ----------------------------------------
419
420pub(crate) async fn rebuild_store(State(state): State<AppState>) -> Json<serde_json::Value> {
421 let Some(store) = &state.store else {
422 return Json(serde_json::json!({ "ok": false, "error": "store not available" }));
423 };
424 let home = state.core.sessions_home();
425 match store.rebuild(&home) {
426 Ok(stats) => Json(serde_json::json!({
427 "ok": true,
428 "files_scanned": stats.files_scanned,
429 "entries_indexed": stats.entries_indexed,
430 "fts_rows": stats.fts_rows,
431 })),
432 Err(e) => Json(serde_json::json!({
433 "ok": false,
434 "error": e.to_string(),
435 })),
436 }
437}
438
439// ---- GET /admin/api/store/import/:session_id -------------------------------
440
441pub(crate) async fn import_session_store(
442 State(state): State<AppState>,
443 Path(session_id): Path<String>,
444) -> Json<serde_json::Value> {
445 let Some(store) = &state.store else {
446 return Json(serde_json::json!({ "ok": false, "error": "store not available" }));
447 };
448 let home = state.core.sessions_home();
449 let shared = state.core.shared_data_home();
450
451 let mut candidate: Option<(std::path::PathBuf, std::path::PathBuf)> = None;
452
453 let direct = home.join("sessions");
454 if direct.exists() {
455 for entry in walkdir::WalkDir::new(&direct)
456 .min_depth(2)
457 .max_depth(2)
458 .into_iter()
459 .filter_entry(|e| e.file_type().is_file())
460 .flatten()
461 {
462 let path = entry.path();
463 if path.extension().and_then(|e| e.to_str()) == Some("jsonl")
464 && path.file_stem().and_then(|s| s.to_str()) == Some(&session_id)
465 {
466 candidate = Some((home.clone(), path.to_path_buf()));
467 break;
468 }
469 }
470 }
471
472 if let Some(agents) = candidate
473 .is_none()
474 .then(|| std::fs::read_dir(shared.join("agents")).ok())
475 .flatten()
476 {
477 for agent in agents.flatten() {
478 let agent_home = agent.path();
479 let agent_sessions = agent_home.join("sessions");
480 if agent_sessions.exists() {
481 for entry in walkdir::WalkDir::new(&agent_sessions)
482 .min_depth(2)
483 .max_depth(2)
484 .into_iter()
485 .filter_entry(|e| e.file_type().is_file())
486 .flatten()
487 {
488 let path = entry.path();
489 if path.extension().and_then(|e| e.to_str()) == Some("jsonl")
490 && path.file_stem().and_then(|s| s.to_str()) == Some(&session_id)
491 {
492 candidate = Some((agent_home.clone(), path.to_path_buf()));
493 break;
494 }
495 }
496 }
497 if candidate.is_some() {
498 break;
499 }
500 }
501 }
502
503 if let Some((agent_home, path)) = candidate {
504 match store.import_session(&agent_home, &path) {
505 Ok(stats) => {
506 return Json(serde_json::json!({
507 "ok": true,
508 "entries_indexed": stats.entries_indexed,
509 "fts_rows": stats.fts_rows,
510 "skipped": stats.skipped,
511 }));
512 }
513 Err(e) => {
514 return Json(serde_json::json!({
515 "ok": false,
516 "error": e.to_string(),
517 }));
518 }
519 }
520 }
521 Json(serde_json::json!({ "error": format!("session {session_id} not found") }))
522}
523
524// ---- GET /admin/api/bestofn ------------------------------------------------
525
526#[derive(Debug, Serialize)]
527struct BestOfNRun {
528 session_id: String,
529 repo: String,
530 branch: String,
531}
532
533/// Active best-of-N candidate runs keyed by child session.
534pub(crate) async fn list_bestofn(State(state): State<AppState>) -> Json<serde_json::Value> {
535 let map = state
536 .best_runs
537 .lock()
538 .unwrap_or_else(std::sync::PoisonError::into_inner);
539 let runs: Vec<BestOfNRun> = map
540 .iter()
541 .map(|(id, meta)| BestOfNRun {
542 session_id: id.clone(),
543 repo: meta.repo.display().to_string(),
544 branch: meta.branch.clone(),
545 })
546 .collect();
547 let total = runs.len();
548 Json(serde_json::json!({ "runs": runs, "total": total }))
549}
550
551// ---- GET /admin/api/config ------------------------------------------------
552
553pub(crate) async fn get_config_admin(State(state): State<AppState>) -> Json<serde_json::Value> {
554 // Report the EFFECTIVE provider, model, turns, mode, and theme so runtime
555 // overrides applied by PATCH /config and POST /config/mode are accurately returned.
556 crate::refresh_control_plane(&state);
557 let route = state.core.effective_route();
558 let cfg = state.core.config();
559 let work = state.core.effective_work();
560 // The lists the engine actually evaluates, not the ones loaded at
561 // startup: `PUT /config/permissions` changes them without a restart, and
562 // a console showing the stale set would be reporting rules no run uses.
563 let permission_rules = state.core.effective_permission_rules();
564 Json(serde_json::json!({
565 "provider": route.provider,
566 "model": route.model,
567 "provider_source": route.provider_source,
568 "model_source": route.model_source,
569 "route_revision": route.revision,
570 "max_turns": state.core.effective_max_turns(),
571 "permission_mode": format!("{:?}", state.core.effective_permission_mode()),
572 // These three were consumed by the console and never sent. The
573 // console's `ConfigInfo` declared all of them, so nothing caught it:
574 // the approval-mode picker could not show which mode was in force,
575 // "Effective sandbox" rendered its loading placeholder forever, and
576 // the workers toggle rendered unchecked whatever the real value
577 // was — so the first click wrote the opposite of what was displayed.
578 "approval_mode": state.core.effective_approval_mode().as_str(),
579 "sandbox": state.core.effective_sandbox_name(),
580 "workers": state.core.effective_workers(),
581 "theme": state.core.effective_theme(),
582 "voice": {
583 "enabled": state.core.effective_voice().enabled,
584 "max_session_secs": state.core.effective_voice().max_session_secs,
585 "max_concurrent": state.core.effective_voice().max_concurrent,
586 "max_audio_bytes": state.core.effective_voice().max_audio_bytes,
587 // Keep quota semantics explicit for operators. These values are
588 // the live workspace admission limits; narrower bot/chat pins
589 // are reported by their binding endpoints and never merged here.
590 "quota": {
591 "session_seconds": state.core.effective_voice().max_session_secs,
592 "concurrent_sessions": state.core.effective_voice().max_concurrent,
593 "inbound_audio_bytes": state.core.effective_voice().max_audio_bytes,
594 "scope": "workspace",
595 "source": "effective",
596 },
597 "source": "effective",
598 },
599 "work": {
600 "enabled": work.enabled,
601 "default_mode": work.default_mode,
602 "max_items": work.max_items,
603 "max_revisions": work.max_revisions,
604 "max_parallel": work.max_parallel,
605 "confirmation": work.confirmation,
606 },
607 "memory": {
608 "search_enabled": cfg.memory.search_enabled,
609 "write_enabled": cfg.memory.write_enabled,
610 "skill_proposals": cfg.memory.skill_proposals,
611 "reflection": cfg.memory.reflection,
612 },
613 // The session list at `/admin/api/sessions` spans every project the
614 // store indexes, but archive/delete/run/steer on a session only
615 // reach a ledger file under *this* process's own workspace
616 // (`sessions_home/sessions/<hash(cwd)>/`). This is that same hash,
617 // matching `project_hash` on each session row — the console uses it
618 // to tell which rows those actions can actually reach.
619 "workspace_project_hash": vak_core::memory::hash_cwd(state.core.cwd()),
620 // The resolved rule lists the permission engine actually evaluates
621 // (vak_permission::Rule syntax: `Tool`, `Tool(glob)`, with a
622 // `+`/`?`/`-` prefix for allow/ask/deny). The admin console shows
623 // these verbatim and derives per-extension scope from them, so a
624 // reader can see what an MCP server or a hook is permitted to do
625 // rather than only that it is configured.
626 "permissions": {
627 "allow": permission_rules.0,
628 "ask": permission_rules.1,
629 "deny": permission_rules.2,
630 },
631 }))
632}
633
634// ---- GET /admin/api/gateway/status ----------------------------------------
635
636pub(crate) async fn gateway_status_admin(State(state): State<AppState>) -> Json<serde_json::Value> {
637 let gw = &state.gateway;
638 crate::refresh_control_plane(&state);
639 let default_route = state.core.effective_route();
640 let mut bindings = Vec::new();
641 let mut bound_targets = std::collections::HashSet::new();
642 for (target, binding) in gw.bindings_snapshot() {
643 bound_targets.insert(target.clone());
644 let configured_workspace = gw.workspace_override_for_entry(&target);
645 let effective_workspace = configured_workspace
646 .clone()
647 .or_else(|| binding.workspace.clone())
648 .unwrap_or_else(|| state.core.cwd().to_path_buf());
649 let channel_override = binding.provider.clone().zip(binding.model.clone());
650 let (provider, model, source, revision) = match &channel_override {
651 Some((provider, model)) => (
652 provider.clone(),
653 model.clone(),
654 "channel_override",
655 format!("channel:{provider}:{model}"),
656 ),
657 None => (
658 default_route.provider.clone(),
659 default_route.model.clone(),
660 "workspace_default",
661 default_route.revision.clone(),
662 ),
663 };
664 let contract = binding.session_id.as_ref().and_then(|session_id| {
665 state
666 .get(session_id)
667 .and_then(|handle| {
668 handle
669 .session
670 .lock()
671 .unwrap_or_else(std::sync::PoisonError::into_inner)
672 .as_ref()
673 .and_then(|session| session.header().cloned())
674 })
675 .or_else(|| {
676 crate::read_historical_header(&state, session_id, binding.workspace.as_deref())
677 })
678 });
679 let stale_reasons = contract
680 .as_ref()
681 .map(|header| {
682 let mut reasons = Vec::new();
683 if header.cwd != effective_workspace {
684 reasons.push("workspace_changed");
685 }
686 // Per-turn routing: provider/model are resolved fresh each turn
687 // from effective_route(), so the header's initial snapshot no
688 // longer constitutes a stale reason. WorkReceipt records actual
689 // per-turn dispatch for audit.
690 reasons
691 })
692 .unwrap_or_else(|| {
693 binding
694 .session_id
695 .as_ref()
696 .map(|_| vec!["session_missing"])
697 .unwrap_or_default()
698 });
699 let paused = binding
700 .session_id
701 .as_deref()
702 .and_then(|session_id| state.get(session_id))
703 .is_some_and(|handle| handle.steering.is_paused());
704 bindings.push(serde_json::json!({
705 "target": target,
706 "session_id": binding.session_id,
707 "workspace": effective_workspace,
708 "configured_workspace": configured_workspace,
709 "override": channel_override.map(|(provider, model)| serde_json::json!({
710 "provider": provider,
711 "model": model,
712 })),
713 "effective_route": {
714 "provider": provider,
715 "model": model,
716 "source": source,
717 "revision": revision,
718 },
719 "session_contract": contract.map(|header| serde_json::json!({
720 "provider": header.contract.provider,
721 "model": header.contract.model,
722 "workspace": header.cwd,
723 "app_version": header.contract.app_version,
724 })),
725 "stale": !stale_reasons.is_empty(),
726 "stale_reasons": stale_reasons,
727 "paused": paused,
728 }));
729 }
730 // An approved channel does not acquire a runtime binding until its first
731 // accepted message creates a session. Keep that approved-but-cold channel
732 // visible in the same table so the Admin UI does not claim it disappeared.
733 for entry in gw.allowlist_snapshot() {
734 if entry.status != crate::gateway::AllowlistStatus::Allowed
735 || bound_targets.contains(&entry.key)
736 {
737 continue;
738 }
739 let channel_override = entry
740 .route
741 .as_ref()
742 .map(|route| (route.provider.clone(), route.model.clone()));
743 let (provider, model, source, revision) = match &channel_override {
744 Some((provider, model)) => (
745 provider.clone(),
746 model.clone(),
747 "channel_override",
748 format!("channel:{provider}:{model}"),
749 ),
750 None => (
751 default_route.provider.clone(),
752 default_route.model.clone(),
753 "workspace_default",
754 default_route.revision.clone(),
755 ),
756 };
757 let workspace = gw.workspace_for_entry(&state.core, &entry.key);
758 bindings.push(serde_json::json!({
759 "target": entry.key,
760 "session_id": null,
761 "workspace": workspace,
762 "configured_workspace": gw.workspace_override_for_entry(&entry.key),
763 "override": channel_override.map(|(provider, model)| serde_json::json!({
764 "provider": provider,
765 "model": model,
766 })),
767 "effective_route": {
768 "provider": provider,
769 "model": model,
770 "source": source,
771 "revision": revision,
772 },
773 "session_contract": null,
774 "stale": false,
775 "stale_reasons": [],
776 }));
777 }
778 bindings.sort_by(|a, b| a["target"].as_str().cmp(&b["target"].as_str()));
779 // docs/design/34 Phase 2: which workspaces currently have a pooled Core
780 // running (warm) vs. cold (will lazily start on next inbound message),
781 // so the workspace picker in the approve flow isn't guessing.
782 let now = std::time::Instant::now();
783 let core_pool: Vec<serde_json::Value> = gw
784 .core_pool
785 .snapshot_at(now)
786 .into_iter()
787 .map(|entry| {
788 serde_json::json!({
789 "workspace": entry.workspace,
790 "is_default": entry.is_default,
791 "state": "warm",
792 "idle_secs": entry.idle_secs,
793 "permission_override": entry.permission_override,
794 "effective_permission_mode": entry.effective_permission_mode,
795 })
796 })
797 .collect();
798 Json(serde_json::json!({
799 "enabled": gw.enabled,
800 "workspace": state.core.cwd(),
801 "canonical_default_workspace": vak_config::paths::default_workspace(),
802 "default_route": default_route,
803 "bindings": bindings,
804 "chat_allowlist": state.core.config().gateway.chat_allowlist,
805 "chat_allowlist_open": state.core.config().gateway.chat_allowlist_open,
806 "core_pool": {
807 "max": state.core.config().gateway.core_pool_max,
808 "idle_secs": state.core.config().gateway.core_pool_idle_secs,
809 "entries": core_pool,
810 },
811 // docs/design/34 open question 4: the workspace field is a picker
812 // of workspaces vak has actually run in, not unconstrained free
813 // text — a typo'd path is caught at entry time by offering
814 // known-good options first.
815 "known_workspaces": known_workspaces(&state),
816 "workspace_catalog": workspace_catalog(&state),
817 }))
818}
819
820#[derive(Debug, Deserialize)]
821pub(crate) struct GatewayWorkspaceBody {
822 pub workspace: Option<String>,
823}
824
825pub(crate) async fn patch_gateway_workspace(
826 State(state): State<AppState>,
827 Json(body): Json<GatewayWorkspaceBody>,
828) -> Response {
829 let workspace = body
830 .workspace
831 .as_deref()
832 .map(str::trim)
833 .filter(|value| !value.is_empty());
834 let selected = match workspace {
835 None => None,
836 Some(value) => {
837 let path = PathBuf::from(value);
838 if !path.is_absolute() || !path.is_dir() {
839 return (
840 StatusCode::BAD_REQUEST,
841 Json(serde_json::json!({
842 "error": "workspace must be an existing absolute directory"
843 })),
844 )
845 .into_response();
846 }
847 match std::fs::canonicalize(path) {
848 Ok(path) => Some(path),
849 Err(error) => {
850 return (
851 StatusCode::BAD_REQUEST,
852 Json(serde_json::json!({ "error": error.to_string() })),
853 )
854 .into_response();
855 }
856 }
857 }
858 };
859 let data_home = state.core.sessions_home();
860 if let Err(error) =
861 vak_config::paths::persist_gateway_workspace_at(&data_home, selected.as_deref())
862 {
863 return (
864 StatusCode::INTERNAL_SERVER_ERROR,
865 Json(serde_json::json!({ "error": error.to_string() })),
866 )
867 .into_response();
868 }
869 let effective = vak_config::paths::gateway_workspace_at(
870 &data_home,
871 &vak_config::paths::default_workspace(),
872 );
873 state.hub.emit_config_changed(
874 "gateway_workspace_changed",
875 &effective.display().to_string(),
876 );
877 Json(serde_json::json!({
878 "workspace": effective,
879 "restart_required": effective.as_path() != state.core.cwd().as_path(),
880 }))
881 .into_response()
882}
883
884#[derive(Debug, Deserialize)]
885pub(crate) struct WorkspaceNamePatch {
886 pub path: String,
887 pub name: String,
888}
889
890fn workspace_names_path(state: &AppState) -> PathBuf {
891 state.core.sessions_home().join("workspace-names.json")
892}
893
894fn read_workspace_names(state: &AppState) -> HashMap<String, String> {
895 let path = workspace_names_path(state);
896 std::fs::read_to_string(path)
897 .ok()
898 .and_then(|text| serde_json::from_str(&text).ok())
899 .unwrap_or_default()
900}
901
902async fn patch_workspace_name(
903 State(state): State<AppState>,
904 Json(body): Json<WorkspaceNamePatch>,
905) -> Result<impl IntoResponse, (StatusCode, String)> {
906 let path = std::path::PathBuf::from(body.path.trim());
907 if !path.is_absolute() || !path.is_dir() {
908 return Err((
909 StatusCode::BAD_REQUEST,
910 "workspace must be an existing absolute directory".into(),
911 ));
912 }
913 let name = body.name.trim();
914 if name.is_empty() || name.len() > 80 {
915 return Err((
916 StatusCode::BAD_REQUEST,
917 "workspace name must be 1–80 characters".into(),
918 ));
919 }
920 let key = path.to_string_lossy().to_string();
921 let mut names = read_workspace_names(&state);
922 names.insert(key.clone(), name.to_string());
923 let destination = workspace_names_path(&state);
924 let temp = destination.with_extension("json.tmp");
925 let content = serde_json::to_vec_pretty(&names).map_err(|e| {
926 (
927 StatusCode::INTERNAL_SERVER_ERROR,
928 format!("could not encode workspace names: {e}"),
929 )
930 })?;
931 if let Some(parent) = destination.parent() {
932 tokio::fs::create_dir_all(parent).await.map_err(|e| {
933 (
934 StatusCode::INTERNAL_SERVER_ERROR,
935 format!("could not create workspace registry: {e}"),
936 )
937 })?;
938 }
939 tokio::fs::write(&temp, content).await.map_err(|e| {
940 (
941 StatusCode::INTERNAL_SERVER_ERROR,
942 format!("could not write workspace registry: {e}"),
943 )
944 })?;
945 tokio::fs::rename(&temp, &destination).await.map_err(|e| {
946 (
947 StatusCode::INTERNAL_SERVER_ERROR,
948 format!("could not commit workspace registry: {e}"),
949 )
950 })?;
951 Ok(Json(serde_json::json!({ "path": key, "name": name })))
952}
953
954/// Workspaces vak has session ledgers for, newest-first, plus the
955/// gateway's own cwd and any currently pooled workspace.
956///
957/// Sessions are stored per-cwd (`<home>/sessions/<hash>/<id>.jsonl`) and
958/// the real path lives in each ledger's header, so this reads only the
959/// first line of one file per project directory — no store rebuild, no
960/// `Core` start.
961fn known_workspaces(state: &AppState) -> Vec<String> {
962 use std::io::BufRead;
963 let mut seen: Vec<String> = vec![vak_config::paths::default_workspace().display().to_string()];
964 let gateway_workspace = vak_config::paths::gateway_workspace_at(
965 &state.core.sessions_home(),
966 &vak_config::paths::default_workspace(),
967 );
968 let gateway_workspace_text = gateway_workspace.display().to_string();
969 if gateway_workspace_text != seen[0] {
970 seen.push(gateway_workspace_text);
971 }
972 let current = state.core.cwd().display().to_string();
973 if !seen.contains(&current) {
974 seen.push(current);
975 }
976 let mut push = |path: String| {
977 if !path.is_empty() && !seen.contains(&path) && !is_scratch_workspace(&path) {
978 seen.push(path);
979 }
980 };
981 for entry in state
982 .gateway
983 .core_pool
984 .snapshot_at(std::time::Instant::now())
985 {
986 push(entry.workspace.display().to_string());
987 }
988 let shared = state.core.shared_data_home();
989 let mut scan_sessions_root = |sessions_root: std::path::PathBuf| {
990 let Ok(projects) = std::fs::read_dir(&sessions_root) else {
991 return;
992 };
993 for project in projects.flatten() {
994 let Ok(files) = std::fs::read_dir(project.path()) else {
995 continue;
996 };
997 for file in files.flatten() {
998 let path = file.path();
999 if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
1000 continue;

1//! Admin console API endpoints (docs/design/29-personal-os.md): session
2//! catalog, transcripts, live event SSE, security audit log, and store
3//! management. Mounted under `/admin/api` in `router_with_state`.
4
5use std::path::PathBuf;
6
7use axum::Json;
8use axum::extract::{Path, Query, State};
9use axum::http::StatusCode;
10use axum::response::sse::{Event, KeepAlive, Sse};
11use axum::response::{IntoResponse, Response};
12use serde::{Deserialize, Serialize};
13use std::collections::HashMap;
14use tokio_stream::StreamExt;
15
16use crate::AppState;
17
18pub(crate) const SESSION_COOKIE: &str = "vak_session";
19
20// ---- GET /admin/api/sessions ----------------------------------------------
21
22#[derive(Debug, Deserialize)]
23pub(crate) struct SessionListQuery {
24 pub limit: Option<usize>,
25 pub project: Option<String>,
26 pub agent: Option<String>,
27}
28
29#[derive(Debug, Serialize)]
30pub(crate) struct SessionListItem {
31 pub session_id: String,
32 pub project_hash: String,
33 pub entry_count: usize,
34 pub first_ts: String,
35 pub last_ts: String,
36 /// From the shared `archive.json` (keyed by session id, not scoped to a
37 /// workspace — the same map `/sessions/{id}/archive` reads and writes).
38 pub archived: bool,
39 #[serde(default, skip_serializing_if = "Option::is_none")]
40 pub agent_id: Option<String>,
41}
42
43fn map_session_agents(shared: &std::path::Path) -> HashMap<String, String> {
44 let mut map = HashMap::new();
45 if let Ok(agents) = std::fs::read_dir(shared.join("agents")) {
46 for agent in agents.flatten() {
47 let agent_id = agent.file_name().to_string_lossy().into_owned();
48 let agent_sessions = agent.path().join("sessions");
49 if agent_sessions.exists() {
50 for e in walkdir::WalkDir::new(&agent_sessions)
51 .max_depth(3)
52 .into_iter()
53 .flatten()
54 {
55 let path = e.path();
56 let maybe_stem = (e.file_type().is_file()
57 && path.extension().is_some_and(|ext| ext == "jsonl"))
58 .then(|| path.file_stem().and_then(|s| s.to_str()))
59 .flatten();
60 if let Some(stem) = maybe_stem {
61 map.insert(stem.to_string(), agent_id.clone());
62 }
63 }
64 }
65 }
66 }
67 map
68}
69
70pub(crate) async fn list_sessions_admin(
71 State(state): State<AppState>,
72 Query(q): Query<SessionListQuery>,
73) -> Response {
74 let Some(store) = &state.store else {
75 return (
76 StatusCode::SERVICE_UNAVAILABLE,
77 Json(serde_json::json!({ "error": "store not available" })),
78 )
79 .into_response();
80 };
81 let limit = q.limit.unwrap_or(50).clamp(1, 200);
82 // Both maps live under the shared vak home (`sessions_home`), not a
83 // per-workspace directory, so they apply across every project this
84 // store indexes — unlike archive/delete *mutation*, which only reaches
85 // a ledger file under this process's own workspace (see
86 // `workspace_project_hash` on `/admin/api/config`).
87 let archive_map = crate::read_archive(&state.core);
88 let shared = state.core.shared_data_home();
89 let trashed = vak_core::trash::trashed(&shared);
90 let agent_map = map_session_agents(&shared);
91
92 match store.list_sessions() {
93 Ok(all) => {
94 let visible = all.into_iter().filter(|s| {
95 let s_agent = agent_map
96 .get(&s.session_id)
97 .map(String::as_str)
98 .unwrap_or("vak");
99 q.project.as_ref().is_none_or(|p| &s.project_hash == p)
100 && q.agent.as_ref().is_none_or(|a| a == "all" || s_agent == a)
101 && !trashed.contains(&s.session_id)
102 });
103 let visible: Vec<_> = visible.collect();
104 let total = visible.len();
105 let items: Vec<SessionListItem> = visible
106 .into_iter()
107 .take(limit)
108 .map(|s| {
109 let agent_id = agent_map
110 .get(&s.session_id)
111 .cloned()
112 .or_else(|| Some("vak".to_string()));
113 SessionListItem {
114 archived: archive_map.get(&s.session_id).copied().unwrap_or(false),
115 session_id: s.session_id,
116 project_hash: s.project_hash,
117 entry_count: s.entry_count,
118 first_ts: s.first_ts,
119 last_ts: s.last_ts,
120 agent_id,
121 }
122 })
123 .collect();
124 Json(serde_json::json!({ "sessions": items, "total": total })).into_response()
125 }
126 Err(e) => (
127 StatusCode::INTERNAL_SERVER_ERROR,
128 Json(serde_json::json!({ "error": e.to_string() })),
129 )
130 .into_response(),
131 }
132}
133
134// ---- GET /admin/api/approvals ----------------------------------------------
135
136#[derive(Debug, Serialize)]
137pub(crate) struct PendingApproval {
138 pub session_id: String,
139 pub request_id: String,
140 pub tool: String,
141 pub args_json: String,
142 pub reason: String,
143 pub requested_at: String,
144}
145
146/// Aggregated pending approval gates across every live session. Answering
147/// still goes through the per-session endpoint — listing never crosses a
148/// session's approval scope, only displays it.
149pub(crate) async fn list_pending_approvals(
150 State(state): State<AppState>,
151) -> Json<serde_json::Value> {
152 let mut items: Vec<PendingApproval> = Vec::new();
153 for handle in state.live_handles() {
154 let pending = handle
155 .pending
156 .lock()
157 .unwrap_or_else(std::sync::PoisonError::into_inner);
158 for req in pending.values() {
159 items.push(PendingApproval {
160 session_id: handle.id.clone(),
161 request_id: req.id.clone(),
162 tool: req.tool.clone(),
163 args_json: truncate_chars(&req.args_json, 400),
164 reason: req.reason.clone(),
165 requested_at: req.requested_at.to_rfc3339(),
166 });
167 }
168 }
169 // Oldest first — the gate that has been waiting longest is the most urgent.
170 items.sort_by(|a, b| a.requested_at.cmp(&b.requested_at));
171 let total = items.len();
172 Json(serde_json::json!({ "approvals": items, "total": total }))
173}
174
175// ---- GET /admin/api/sessions/:id/transcript --------------------------------
176
177#[derive(Debug, Deserialize)]
178pub(crate) struct TranscriptQuery {
179 pub limit: Option<usize>,
180 pub offset: Option<usize>,
181 pub kind: Option<String>,
182 pub role: Option<String>,
183 /// Re-import this session's JSONL into the index before reading, so
184 /// entries appended by an active run become visible immediately.
185 pub refresh: Option<bool>,
186}
187
188/// Byte-safe truncation: never splits a multi-byte UTF-8 sequence.
189fn truncate_chars(s: &str, max_bytes: usize) -> String {
190 if s.len() <= max_bytes {
191 return s.to_string();
192 }
193 let mut end = max_bytes;
194 while end > 0 && !s.is_char_boundary(end) {
195 end -= 1;
196 }
197 format!("{}…", &s[..end])
198}
199
200/// Which ledger entries of a session are runtime-authored control messages,
201/// by entry id (empty when the session cannot be read).
202fn control_kinds_by_entry(
203 state: &AppState,
204 session_id: &str,
205) -> std::collections::HashMap<String, vak_intent::control::ControlKind> {
206 let from = |session: &vak_session::SessionLog| {
207 session
208 .derive_transcript()
209 .into_iter()
210 .filter_map(|item| item.control.map(|kind| (item.entry_id, kind)))
211 .collect()
212 };
213 if let Some(handle) = state.get(session_id)
214 && let Some(session) = handle
215 .session
216 .lock()
217 .unwrap_or_else(std::sync::PoisonError::into_inner)
218 .as_ref()
219 {
220 return from(session);
221 }
222 crate::open_historical_session(state, session_id)
223 .map(|session| from(&session))
224 .unwrap_or_default()
225}
226
227pub(crate) async fn session_transcript_admin(
228 State(state): State<AppState>,
229 Path(session_id): Path<String>,
230 Query(q): Query<TranscriptQuery>,
231) -> Json<serde_json::Value> {
232 let Some(store) = &state.store else {
233 return Json(serde_json::json!({ "error": "store not available" }));
234 };
235 if vak_core::trash::is_trashed(&state.core.shared_data_home(), &session_id) {
236 return Json(serde_json::json!({ "error": "session is in the trash" }));
237 }
238 // Fetch offset+limit so we can report whether more pages exist.
239 let limit = q.limit.unwrap_or(100).clamp(1, 500);
240 let offset = q.offset.unwrap_or(0);
241 if q.refresh == Some(true) {
242 crate::import_session_sync(store, &state.core.sessions_home(), &session_id);
243 }
244 let filter = vak_store::query::SearchFilter {
245 session_id: Some(session_id.clone()),
246 kind: q.kind,
247 role: q.role,
248 ..Default::default()
249 };
250 match store.query_page(&filter, limit, offset, true) {
251 Ok((entries, total)) => {
252 let has_more = offset.saturating_add(entries.len()) < total;
253 // The search index stores text only, so it cannot say which user
254 // rows the runtime authored. The ledger can: tag each row from it.
255 let controls = control_kinds_by_entry(&state, &session_id);
256 let page: Vec<serde_json::Value> = entries
257 .iter()
258 .map(|e| {
259 serde_json::json!({
260 "entry_id": e.entry_id,
261 "ts": e.ts,
262 "kind": e.kind.as_str(),
263 "role": e.role,
264 "tool_name": e.tool_name,
265 "is_error": e.is_error,
266 "content": truncate_chars(&e.content_text, 16000),
267 "control": controls.get(&e.entry_id),
268 })
269 })
270 .collect();
271 let contract = state
272 .get(&session_id)
273 .and_then(|handle| {
274 handle
275 .session
276 .lock()
277 .unwrap_or_else(std::sync::PoisonError::into_inner)
278 .as_ref()
279 .and_then(|session| session.header().map(|header| header.contract.clone()))
280 })
281 .or_else(|| {
282 crate::open_historical_session(&state, &session_id)
283 .and_then(|session| session.header().map(|header| header.contract.clone()))
284 });
285 let configuration_mismatch = false; // Per-turn routing: contract snapshot != mismatch
286 Json(serde_json::json!({
287 "session_id": session_id,
288 "entries": page,
289 "offset": offset,
290 "total": total,
291 "has_more": has_more,
292 "contract": contract,
293 "configuration_mismatch": configuration_mismatch,
294 }))
295 }
296 Err(e) => Json(serde_json::json!({ "error": e.to_string() })),
297 }
298}
299
300// ---- GET /admin/api/search -------------------------------------------------
301
302#[derive(Debug, Deserialize)]
303pub(crate) struct SearchQuery {
304 pub q: String,
305 pub limit: Option<usize>,
306 pub project: Option<String>,
307 pub role: Option<String>,
308 pub kind: Option<String>,
309 pub exclude_session: Option<String>,
310}
311
312pub(crate) async fn search_admin(
313 State(state): State<AppState>,
314 Query(q): Query<SearchQuery>,
315) -> Json<serde_json::Value> {
316 let Some(store) = &state.store else {
317 return Json(serde_json::json!({ "error": "store not available" }));
318 };
319 let limit = q.limit.unwrap_or(20).clamp(1, 100);
320 let filter = vak_store::query::SearchFilter {
321 project_hash: q.project,
322 role: q.role,
323 kind: q.kind,
324 excluded_sessions: vak_core::trash::search_exclusions(
325 &state.core.shared_data_home(),
326 q.exclude_session.as_deref(),
327 )
328 .into_iter()
329 .collect(),
330 ..Default::default()
331 };
332 match store.search(&q.q, limit, &filter) {
333 Ok(result) => Json(serde_json::json!({
334 "hits": result.entries,
335 "total": result.total,
336 })),
337 Err(e) => Json(serde_json::json!({ "error": e.to_string() })),
338 }
339}
340
341// ---- GET /admin/api/events (SSE) ------------------------------------------
342
343pub(crate) async fn admin_events_sse(
344 State(state): State<AppState>,
345) -> Sse<impl tokio_stream::Stream<Item = Result<Event, std::convert::Infallible>>> {
346 use tokio_stream::wrappers::BroadcastStream;
347
348 let rx = state.hub.subscribe();
349 let stream = BroadcastStream::new(rx).filter_map(|result| match result {
350 Ok(event) => {
351 let json = serde_json::to_string(&event).unwrap_or_default();
352 Some(Ok(Event::default().data(json)))
353 }
354 Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(n)) => {
355 let lagged = serde_json::json!({
356 "type": "Lagged",
357 "data": { "missed": n }
358 });
359 let json = serde_json::to_string(&lagged).unwrap_or_default();
360 Some(Ok(Event::default().data(json)))
361 }
362 });
363 Sse::new(stream).keep_alive(KeepAlive::default())
364}
365
366// ---- GET /admin/api/security -----------------------------------------------
367
368#[derive(Debug, Deserialize)]
369pub(crate) struct SecurityQuery {
370 pub limit: Option<usize>,
371 pub kind: Option<String>,
372}
373
374#[derive(Debug, Serialize)]
375struct SecurityEventEntry {
376 ts: String,
377 kind: String,
378 label: String,
379 detail: String,
380 ip: Option<String>,
381}
382
383pub(crate) async fn list_security_events(
384 State(state): State<AppState>,
385 Query(q): Query<SecurityQuery>,
386) -> Json<serde_json::Value> {
387 let limit = q.limit.unwrap_or(100).clamp(1, 500);
388 let home = state.core.sessions_home();
389 let events = vak_core::security_events::list(&home, limit);
390 let filtered: Vec<SecurityEventEntry> = events
391 .into_iter()
392 .filter(|e| {
393 q.kind.as_deref().is_none_or(|k| {
394 serde_json::to_value(&e.kind)
395 .ok()
396 .and_then(|v| v.as_str().map(String::from))
397 .as_deref()
398 == Some(k)
399 })
400 })
401 .map(|e| SecurityEventEntry {
402 ts: e.ts.to_rfc3339(),
403 kind: serde_json::to_value(&e.kind)
404 .ok()
405 .and_then(|v| v.as_str().map(String::from))
406 .unwrap_or_else(|| "unknown".into()),
407 label: e.label,
408 detail: e.detail,
409 ip: e.ip,
410 })
411 .collect();
412 Json(serde_json::json!({
413 "events": filtered,
414 "total": filtered.len(),
415 }))
416}
417
418// ---- POST /admin/api/store/rebuild ----------------------------------------
419
420pub(crate) async fn rebuild_store(State(state): State<AppState>) -> Json<serde_json::Value> {
421 let Some(store) = &state.store else {
422 return Json(serde_json::json!({ "ok": false, "error": "store not available" }));
423 };
424 let home = state.core.sessions_home();
425 match store.rebuild(&home) {
426 Ok(stats) => Json(serde_json::json!({
427 "ok": true,
428 "files_scanned": stats.files_scanned,
429 "entries_indexed": stats.entries_indexed,
430 "fts_rows": stats.fts_rows,
431 })),
432 Err(e) => Json(serde_json::json!({
433 "ok": false,
434 "error": e.to_string(),
435 })),
436 }
437}
438
439// ---- GET /admin/api/store/import/:session_id -------------------------------
440
441pub(crate) async fn import_session_store(
442 State(state): State<AppState>,
443 Path(session_id): Path<String>,
444) -> Json<serde_json::Value> {
445 let Some(store) = &state.store else {
446 return Json(serde_json::json!({ "ok": false, "error": "store not available" }));
447 };
448 let home = state.core.sessions_home();
449 let shared = state.core.shared_data_home();
450
451 let mut candidate: Option<(std::path::PathBuf, std::path::PathBuf)> = None;
452
453 let direct = home.join("sessions");
454 if direct.exists() {
455 for entry in walkdir::WalkDir::new(&direct)
456 .min_depth(2)
457 .max_depth(2)
458 .into_iter()
459 .filter_entry(|e| e.file_type().is_file())
460 .flatten()
461 {
462 let path = entry.path();
463 if path.extension().and_then(|e| e.to_str()) == Some("jsonl")
464 && path.file_stem().and_then(|s| s.to_str()) == Some(&session_id)
465 {
466 candidate = Some((home.clone(), path.to_path_buf()));
467 break;
468 }
469 }
470 }
471
472 if let Some(agents) = candidate
473 .is_none()
474 .then(|| std::fs::read_dir(shared.join("agents")).ok())
475 .flatten()
476 {
477 for agent in agents.flatten() {
478 let agent_home = agent.path();
479 let agent_sessions = agent_home.join("sessions");
480 if agent_sessions.exists() {
481 for entry in walkdir::WalkDir::new(&agent_sessions)
482 .min_depth(2)
483 .max_depth(2)
484 .into_iter()
485 .filter_entry(|e| e.file_type().is_file())
486 .flatten()
487 {
488 let path = entry.path();
489 if path.extension().and_then(|e| e.to_str()) == Some("jsonl")
490 && path.file_stem().and_then(|s| s.to_str()) == Some(&session_id)
491 {
492 candidate = Some((agent_home.clone(), path.to_path_buf()));
493 break;
494 }
495 }
496 }
497 if candidate.is_some() {
498 break;
499 }
500 }
501 }
502
503 if let Some((agent_home, path)) = candidate {
504 match store.import_session(&agent_home, &path) {
505 Ok(stats) => {
506 return Json(serde_json::json!({
507 "ok": true,
508 "entries_indexed": stats.entries_indexed,
509 "fts_rows": stats.fts_rows,
510 "skipped": stats.skipped,
511 }));
512 }
513 Err(e) => {
514 return Json(serde_json::json!({
515 "ok": false,
516 "error": e.to_string(),
517 }));
518 }
519 }
520 }
521 Json(serde_json::json!({ "error": format!("session {session_id} not found") }))
522}
523
524// ---- GET /admin/api/bestofn ------------------------------------------------
525
526#[derive(Debug, Serialize)]
527struct BestOfNRun {
528 session_id: String,
529 repo: String,
530 branch: String,
531}
532
533/// Active best-of-N candidate runs keyed by child session.
534pub(crate) async fn list_bestofn(State(state): State<AppState>) -> Json<serde_json::Value> {
535 let map = state
536 .best_runs
537 .lock()
538 .unwrap_or_else(std::sync::PoisonError::into_inner);
539 let runs: Vec<BestOfNRun> = map
540 .iter()
541 .map(|(id, meta)| BestOfNRun {
542 session_id: id.clone(),
543 repo: meta.repo.display().to_string(),
544 branch: meta.branch.clone(),
545 })
546 .collect();
547 let total = runs.len();
548 Json(serde_json::json!({ "runs": runs, "total": total }))
549}
550
551// ---- GET /admin/api/config ------------------------------------------------
552
553pub(crate) async fn get_config_admin(State(state): State<AppState>) -> Json<serde_json::Value> {
554 // Report the EFFECTIVE provider, model, turns, mode, and theme so runtime
555 // overrides applied by PATCH /config and POST /config/mode are accurately returned.
556 crate::refresh_control_plane(&state);
557 let route = state.core.effective_route();
558 let cfg = state.core.config();
559 let work = state.core.effective_work();
560 // The lists the engine actually evaluates, not the ones loaded at
561 // startup: `PUT /config/permissions` changes them without a restart, and
562 // a console showing the stale set would be reporting rules no run uses.
563 let permission_rules = state.core.effective_permission_rules();
564 Json(serde_json::json!({
565 "provider": route.provider,
566 "model": route.model,
567 "provider_source": route.provider_source,
568 "model_source": route.model_source,
569 "route_revision": route.revision,
570 "max_turns": state.core.effective_max_turns(),
571 "permission_mode": format!("{:?}", state.core.effective_permission_mode()),
572 // These three were consumed by the console and never sent. The
573 // console's `ConfigInfo` declared all of them, so nothing caught it:
574 // the approval-mode picker could not show which mode was in force,
575 // "Effective sandbox" rendered its loading placeholder forever, and
576 // the workers toggle rendered unchecked whatever the real value
577 // was — so the first click wrote the opposite of what was displayed.
578 "approval_mode": state.core.effective_approval_mode().as_str(),
579 "sandbox": state.core.effective_sandbox_name(),
580 "workers": state.core.effective_workers(),
581 "theme": state.core.effective_theme(),
582 "voice": {
583 "enabled": state.core.effective_voice().enabled,
584 "max_session_secs": state.core.effective_voice().max_session_secs,
585 "max_concurrent": state.core.effective_voice().max_concurrent,
586 "max_audio_bytes": state.core.effective_voice().max_audio_bytes,
587 // Keep quota semantics explicit for operators. These values are
588 // the live workspace admission limits; narrower bot/chat pins
589 // are reported by their binding endpoints and never merged here.
590 "quota": {
591 "session_seconds": state.core.effective_voice().max_session_secs,
592 "concurrent_sessions": state.core.effective_voice().max_concurrent,
593 "inbound_audio_bytes": state.core.effective_voice().max_audio_bytes,
594 "scope": "workspace",
595 "source": "effective",
596 },
597 "source": "effective",
598 },
599 "work": {
600 "enabled": work.enabled,
601 "default_mode": work.default_mode,
602 "max_items": work.max_items,
603 "max_revisions": work.max_revisions,
604 "max_parallel": work.max_parallel,
605 "confirmation": work.confirmation,
606 },
607 "memory": {
608 "search_enabled": cfg.memory.search_enabled,
609 "write_enabled": cfg.memory.write_enabled,
610 "skill_proposals": cfg.memory.skill_proposals,
611 "reflection": cfg.memory.reflection,
612 },
613 // The session list at `/admin/api/sessions` spans every project the
614 // store indexes, but archive/delete/run/steer on a session only
615 // reach a ledger file under *this* process's own workspace
616 // (`sessions_home/sessions/<hash(cwd)>/`). This is that same hash,
617 // matching `project_hash` on each session row — the console uses it
618 // to tell which rows those actions can actually reach.
619 "workspace_project_hash": vak_core::memory::hash_cwd(state.core.cwd()),
620 // The resolved rule lists the permission engine actually evaluates
621 // (vak_permission::Rule syntax: `Tool`, `Tool(glob)`, with a
622 // `+`/`?`/`-` prefix for allow/ask/deny). The admin console shows
623 // these verbatim and derives per-extension scope from them, so a
624 // reader can see what an MCP server or a hook is permitted to do
625 // rather than only that it is configured.
626 "permissions": {
627 "allow": permission_rules.0,
628 "ask": permission_rules.1,
629 "deny": permission_rules.2,
630 },
631 }))
632}
633
634// ---- GET /admin/api/gateway/status ----------------------------------------
635
636pub(crate) async fn gateway_status_admin(State(state): State<AppState>) -> Json<serde_json::Value> {
637 let gw = &state.gateway;
638 crate::refresh_control_plane(&state);
639 let default_route = state.core.effective_route();
640 let mut bindings = Vec::new();
641 let mut bound_targets = std::collections::HashSet::new();
642 for (target, binding) in gw.bindings_snapshot() {
643 bound_targets.insert(target.clone());
644 let configured_workspace = gw.workspace_override_for_entry(&target);
645 let effective_workspace = configured_workspace
646 .clone()
647 .or_else(|| binding.workspace.clone())
648 .unwrap_or_else(|| state.core.cwd().to_path_buf());
649 let channel_override = binding.provider.clone().zip(binding.model.clone());
650 let (provider, model, source, revision) = match &channel_override {
651 Some((provider, model)) => (
652 provider.clone(),
653 model.clone(),
654 "channel_override",
655 format!("channel:{provider}:{model}"),
656 ),
657 None => (
658 default_route.provider.clone(),
659 default_route.model.clone(),
660 "workspace_default",
661 default_route.revision.clone(),
662 ),
663 };
664 let contract = binding.session_id.as_ref().and_then(|session_id| {
665 state
666 .get(session_id)
667 .and_then(|handle| {
668 handle
669 .session
670 .lock()
671 .unwrap_or_else(std::sync::PoisonError::into_inner)
672 .as_ref()
673 .and_then(|session| session.header().cloned())
674 })
675 .or_else(|| {
676 crate::read_historical_header(&state, session_id, binding.workspace.as_deref())
677 })
678 });
679 let stale_reasons = contract
680 .as_ref()
681 .map(|header| {
682 let mut reasons = Vec::new();
683 if header.cwd != effective_workspace {
684 reasons.push("workspace_changed");
685 }
686 // Per-turn routing: provider/model are resolved fresh each turn
687 // from effective_route(), so the header's initial snapshot no
688 // longer constitutes a stale reason. WorkReceipt records actual
689 // per-turn dispatch for audit.
690 reasons
691 })
692 .unwrap_or_else(|| {
693 binding
694 .session_id
695 .as_ref()
696 .map(|_| vec!["session_missing"])
697 .unwrap_or_default()
698 });
699 let paused = binding
700 .session_id
701 .as_deref()
702 .and_then(|session_id| state.get(session_id))
703 .is_some_and(|handle| handle.steering.is_paused());
704 bindings.push(serde_json::json!({
705 "target": target,
706 "session_id": binding.session_id,
707 "workspace": effective_workspace,
708 "configured_workspace": configured_workspace,
709 "override": channel_override.map(|(provider, model)| serde_json::json!({
710 "provider": provider,
711 "model": model,
712 })),
713 "effective_route": {
714 "provider": provider,
715 "model": model,
716 "source": source,
717 "revision": revision,
718 },
719 "session_contract": contract.map(|header| serde_json::json!({
720 "provider": header.contract.provider,
721 "model": header.contract.model,
722 "workspace": header.cwd,
723 "app_version": header.contract.app_version,
724 })),
725 "stale": !stale_reasons.is_empty(),
726 "stale_reasons": stale_reasons,
727 "paused": paused,
728 }));
729 }
730 // An approved channel does not acquire a runtime binding until its first
731 // accepted message creates a session. Keep that approved-but-cold channel
732 // visible in the same table so the Admin UI does not claim it disappeared.
733 for entry in gw.allowlist_snapshot() {
734 if entry.status != crate::gateway::AllowlistStatus::Allowed
735 || bound_targets.contains(&entry.key)
736 {
737 continue;
738 }
739 let channel_override = entry
740 .route
741 .as_ref()
742 .map(|route| (route.provider.clone(), route.model.clone()));
743 let (provider, model, source, revision) = match &channel_override {
744 Some((provider, model)) => (
745 provider.clone(),
746 model.clone(),
747 "channel_override",
748 format!("channel:{provider}:{model}"),
749 ),
750 None => (
751 default_route.provider.clone(),
752 default_route.model.clone(),
753 "workspace_default",
754 default_route.revision.clone(),
755 ),
756 };
757 let workspace = gw.workspace_for_entry(&state.core, &entry.key);
758 bindings.push(serde_json::json!({
759 "target": entry.key,
760 "session_id": null,
761 "workspace": workspace,
762 "configured_workspace": gw.workspace_override_for_entry(&entry.key),
763 "override": channel_override.map(|(provider, model)| serde_json::json!({
764 "provider": provider,
765 "model": model,
766 })),
767 "effective_route": {
768 "provider": provider,
769 "model": model,
770 "source": source,
771 "revision": revision,
772 },
773 "session_contract": null,
774 "stale": false,
775 "stale_reasons": [],
776 }));
777 }
778 bindings.sort_by(|a, b| a["target"].as_str().cmp(&b["target"].as_str()));
779 // docs/design/34 Phase 2: which workspaces currently have a pooled Core
780 // running (warm) vs. cold (will lazily start on next inbound message),
781 // so the workspace picker in the approve flow isn't guessing.
782 let now = std::time::Instant::now();
783 let core_pool: Vec<serde_json::Value> = gw
784 .core_pool
785 .snapshot_at(now)
786 .into_iter()
787 .map(|entry| {
788 serde_json::json!({
789 "workspace": entry.workspace,
790 "is_default": entry.is_default,
791 "state": "warm",
792 "idle_secs": entry.idle_secs,
793 "permission_override": entry.permission_override,
794 "effective_permission_mode": entry.effective_permission_mode,
795 })
796 })
797 .collect();
798 Json(serde_json::json!({
799 "enabled": gw.enabled,
800 "workspace": state.core.cwd(),
801 "canonical_default_workspace": vak_config::paths::default_workspace(),
802 "default_route": default_route,
803 "bindings": bindings,
804 "chat_allowlist": state.core.config().gateway.chat_allowlist,
805 "chat_allowlist_open": state.core.config().gateway.chat_allowlist_open,
806 "core_pool": {
807 "max": state.core.config().gateway.core_pool_max,
808 "idle_secs": state.core.config().gateway.core_pool_idle_secs,
809 "entries": core_pool,
810 },
811 // docs/design/34 open question 4: the workspace field is a picker
812 // of workspaces vak has actually run in, not unconstrained free
813 // text — a typo'd path is caught at entry time by offering
814 // known-good options first.
815 "known_workspaces": known_workspaces(&state),
816 "workspace_catalog": workspace_catalog(&state),
817 }))
818}
819
820#[derive(Debug, Deserialize)]
821pub(crate) struct GatewayWorkspaceBody {
822 pub workspace: Option<String>,
823}
824
825pub(crate) async fn patch_gateway_workspace(
826 State(state): State<AppState>,
827 Json(body): Json<GatewayWorkspaceBody>,
828) -> Response {
829 let workspace = body
830 .workspace
831 .as_deref()
832 .map(str::trim)
833 .filter(|value| !value.is_empty());
834 let selected = match workspace {
835 None => None,
836 Some(value) => {
837 let path = PathBuf::from(value);
838 if !path.is_absolute() || !path.is_dir() {
839 return (
840 StatusCode::BAD_REQUEST,
841 Json(serde_json::json!({
842 "error": "workspace must be an existing absolute directory"
843 })),
844 )
845 .into_response();
846 }
847 match std::fs::canonicalize(path) {
848 Ok(path) => Some(path),
849 Err(error) => {
850 return (
851 StatusCode::BAD_REQUEST,
852 Json(serde_json::json!({ "error": error.to_string() })),
853 )
854 .into_response();
855 }
856 }
857 }
858 };
859 let data_home = state.core.sessions_home();
860 if let Err(error) =
861 vak_config::paths::persist_gateway_workspace_at(&data_home, selected.as_deref())
862 {
863 return (
864 StatusCode::INTERNAL_SERVER_ERROR,
865 Json(serde_json::json!({ "error": error.to_string() })),
866 )
867 .into_response();
868 }
869 let effective = vak_config::paths::gateway_workspace_at(
870 &data_home,
871 &vak_config::paths::default_workspace(),
872 );
873 state.hub.emit_config_changed(
874 "gateway_workspace_changed",
875 &effective.display().to_string(),
876 );
877 Json(serde_json::json!({
878 "workspace": effective,
879 "restart_required": effective.as_path() != state.core.cwd().as_path(),
880 }))
881 .into_response()
882}
883
884#[derive(Debug, Deserialize)]
885pub(crate) struct WorkspaceNamePatch {
886 pub path: String,
887 pub name: String,
888}
889
890fn workspace_names_path(state: &AppState) -> PathBuf {
891 state.core.sessions_home().join("workspace-names.json")
892}
893
894fn read_workspace_names(state: &AppState) -> HashMap<String, String> {
895 let path = workspace_names_path(state);
896 std::fs::read_to_string(path)
897 .ok()
898 .and_then(|text| serde_json::from_str(&text).ok())
899 .unwrap_or_default()
900}
901
902async fn patch_workspace_name(
903 State(state): State<AppState>,
904 Json(body): Json<WorkspaceNamePatch>,
905) -> Result<impl IntoResponse, (StatusCode, String)> {
906 let path = std::path::PathBuf::from(body.path.trim());
907 if !path.is_absolute() || !path.is_dir() {
908 return Err((
909 StatusCode::BAD_REQUEST,
910 "workspace must be an existing absolute directory".into(),
911 ));
912 }
913 let name = body.name.trim();
914 if name.is_empty() || name.len() > 80 {
915 return Err((
916 StatusCode::BAD_REQUEST,
917 "workspace name must be 1–80 characters".into(),
918 ));
919 }
920 let key = path.to_string_lossy().to_string();
921 let mut names = read_workspace_names(&state);
922 names.insert(key.clone(), name.to_string());
923 let destination = workspace_names_path(&state);
924 let temp = destination.with_extension("json.tmp");
925 let content = serde_json::to_vec_pretty(&names).map_err(|e| {
926 (
927 StatusCode::INTERNAL_SERVER_ERROR,
928 format!("could not encode workspace names: {e}"),
929 )
930 })?;
931 if let Some(parent) = destination.parent() {
932 tokio::fs::create_dir_all(parent).await.map_err(|e| {
933 (
934 StatusCode::INTERNAL_SERVER_ERROR,
935 format!("could not create workspace registry: {e}"),
936 )
937 })?;
938 }
939 tokio::fs::write(&temp, content).await.map_err(|e| {
940 (
941 StatusCode::INTERNAL_SERVER_ERROR,
942 format!("could not write workspace registry: {e}"),
943 )
944 })?;
945 tokio::fs::rename(&temp, &destination).await.map_err(|e| {
946 (
947 StatusCode::INTERNAL_SERVER_ERROR,
948 format!("could not commit workspace registry: {e}"),
949 )
950 })?;
951 Ok(Json(serde_json::json!({ "path": key, "name": name })))
952}
953
954/// Workspaces vak has session ledgers for, newest-first, plus the
955/// gateway's own cwd and any currently pooled workspace.
956///
957/// Sessions are stored per-cwd (`<home>/sessions/<hash>/<id>.jsonl`) and
958/// the real path lives in each ledger's header, so this reads only the
959/// first line of one file per project directory — no store rebuild, no
960/// `Core` start.
961fn known_workspaces(state: &AppState) -> Vec<String> {
962 use std::io::BufRead;
963 let mut seen: Vec<String> = vec![vak_config::paths::default_workspace().display().to_string()];
964 let gateway_workspace = vak_config::paths::gateway_workspace_at(
965 &state.core.sessions_home(),
966 &vak_config::paths::default_workspace(),
967 );
968 let gateway_workspace_text = gateway_workspace.display().to_string();
969 if gateway_workspace_text != seen[0] {
970 seen.push(gateway_workspace_text);
971 }
972 let current = state.core.cwd().display().to_string();
973 if !seen.contains(&current) {
974 seen.push(current);
975 }
976 let mut push = |path: String| {
977 if !path.is_empty() && !seen.contains(&path) && !is_scratch_workspace(&path) {
978 seen.push(path);
979 }
980 };
981 for entry in state
982 .gateway
983 .core_pool
984 .snapshot_at(std::time::Instant::now())
985 {
986 push(entry.workspace.display().to_string());
987 }
988 let shared = state.core.shared_data_home();
989 let mut scan_sessions_root = |sessions_root: std::path::PathBuf| {
990 let Ok(projects) = std::fs::read_dir(&sessions_root) else {
991 return;
992 };
993 for project in projects.flatten() {
994 let Ok(files) = std::fs::read_dir(project.path()) else {
995 continue;
996 };
997 for file in files.flatten() {
998 let path = file.path();
999 if path.extension().and_then(|e| e.to_str()) != Some("jsonl") {
1000 continue;

Indexing the workspace…