Vakyartha documentation

2001 }
2002 "install" => match vak_ops::install(svc, &cfg) {
2003 Ok(()) => (serde_json::json!({ "ok": true, "action": "install" }), true),
2004 Err(e) => (serde_json::json!({ "ok": false, "error": e }), false),
2005 },
2006 "uninstall" => match vak_ops::uninstall(svc, &cfg) {
2007 Ok(()) => (
2008 serde_json::json!({ "ok": true, "action": "uninstall" }),
2009 true,
2010 ),
2011 Err(e) => (serde_json::json!({ "ok": false, "error": e }), false),
2012 },
2013 other => {
2014 return (
2015 StatusCode::BAD_REQUEST,
2016 Json(serde_json::json!({ "error": format!("unknown action '{other}'") })),
2017 )
2018 .into_response();
2019 }
2020 };
2021 let after = service_control::status(svc, cfg.clone())
2022 .await
2023 .map(|s| s.to_string())
2024 .unwrap_or_else(|e| e);
2025 let desired_reached = match action.as_str() {
2026 "start" | "restart" | "install" => after == "running",
2027 "stop" => matches!(after.as_str(), "stopped" | "not installed"),
2028 "uninstall" => after == "not installed",
2029 _ => false,
2030 };
2031 let verification_status = if !succeeded {
2032 "failed"
2033 } else if desired_reached {
2034 "verified"
2035 } else {
2036 "pending"
2037 };
2038 let verification_detail = if !succeeded {
2039 "The service manager rejected the requested operation; the post-action probe is authoritative."
2040 } else if desired_reached {
2041 "The post-action service-manager probe reached the requested state."
2042 } else {
2043 "The manager accepted the request but the desired state is not visible yet; keep the receipt and re-probe."
2044 };
2045 let mut receipt = operations::ActionReceipt {
2046 receipt_id: format!("OP-{}", uuid::Uuid::now_v7().simple()),
2047 service: service.clone(),
2048 action: action.clone(),
2049 requested_at,
2050 completed_at: Utc::now(),
2051 succeeded,
2052 verification: operations::ActionVerification {
2053 status: verification_status.to_string(),
2054 before,
2055 after,
2056 detail: verification_detail.to_string(),
2057 },
2058 persisted: false,
2059 };
2060 receipt.persisted = operations::record_action(&state.core.sessions_home(), &receipt).is_ok();
2061 let receipt_json = serde_json::to_value(&receipt).unwrap_or_else(|_| serde_json::json!({}));
2062 let mut result = result;
2063 if let Some(object) = result.as_object_mut() {
2064 object.insert(
2065 "receipt_id".to_string(),
2066 serde_json::json!(receipt.receipt_id),
2067 );
2068 object.insert(
2069 "verification".to_string(),
2070 receipt_json["verification"].clone(),
2071 );
2072 object.insert(
2073 "receipt_persisted".to_string(),
2074 serde_json::json!(receipt.persisted),
2075 );
2076 }
2077 if succeeded {
2078 vak_core::security_events::record(
2079 &state.core.sessions_home(),
2080 vak_core::security_events::EventKind::ConfigChange,
2081 "service_action",
2082 &format!("service={service} action={action}"),
2083 None,
2084 );
2085 state
2086 .hub
2087 .emit_config_changed("service_action", &format!("{service}:{action}"));
2088 (StatusCode::OK, Json(result)).into_response()
2089 } else {
2090 (StatusCode::CONFLICT, Json(result)).into_response()
2091 }
2092}
2093
2094fn note_payload(n: &vak_core::memory::NoteBlock, scope: &str) -> serde_json::Value {
2095 serde_json::json!({
2096 "id": n.id,
2097 "ts": n.ts.to_rfc3339(),
2098 "kind": n.kind,
2099 "tag": n.tag,
2100 "session_id": n.session_id,
2101 "text": n.text,
2102 "scope": scope,
2103 })
2104}
2105
2106/// Which Agent an endpoint scoped to "the currently open Agent's own data"
2107/// (memory, learning proposals) should resolve against. Absent means the
2108/// built-in "vak" Agent — the same default `agent_chats::open` uses.
2109#[derive(serde::Deserialize, Default)]
2110struct AgentScopeQuery {
2111 #[serde(default)]
2112 agent: Option<String>,
2113}
2114
2115/// Resolve the `Core` an Agent-scoped endpoint should read/write through.
2116///
2117/// A registered session already carries the exact `Core` it was opened
2118/// under (agent identity, isolated workspace, and now-agent-scoped
2119/// `sessions_home` all resolved once at `agent_chats::open` time) — reusing
2120/// it is cheaper and more precise than re-deriving identity from an id, so
2121/// `session_id` (when the caller already has one, e.g. `AppendMemoryBody`)
2122/// takes precedence over an explicit `agent` id.
2123///
2124/// This is the single place "which Agent's data does this endpoint mean"
2125/// gets decided, so a future endpoint scoped the same way calls this
2126/// instead of reading `state.core` directly and drifting out of sync with
2127/// `agent_chats::open` the way `list_sessions` once did (see commit
2128/// 7e6713c0 and its follow-up).
2129#[allow(clippy::result_large_err)]
2130fn resolve_scoped_core(
2131 state: &AppState,
2132 session_id: Option<&str>,
2133 agent: Option<&str>,
2134) -> Result<vak_core::Core, axum::response::Response> {
2135 if let Some(sid) = session_id
2136 && let Some(handle) = state.get(sid)
2137 {
2138 return Ok(handle.core.clone());
2139 }
2140 let id = agent.unwrap_or("vak");
2141 agent_chats::resolve_agent_core(state, id).map(|(_, core)| core)
2142}
2143
2144async fn list_memory(
2145 State(state): State<AppState>,
2146 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2147) -> axum::response::Response {
2148 use axum::response::IntoResponse;
2149 let core = scoped_core!(&state, None, q.agent.as_deref());
2150 // The resolved Agent's own (now agent-scoped) sessions_home is primary;
2151 // `state.core`'s plain, un-scoped home is kept as a fallback merge so
2152 // notes written before Agents carried their own sessions_home (or by an
2153 // older build) are not silently hidden.
2154 let mut homes = vec![core.sessions_home(), state.core.sessions_home()];
2155 let shared = state.core.shared_data_home();
2156 if !homes.contains(&shared) {
2157 homes.push(shared);
2158 }
2159 let mut blocks: Vec<serde_json::Value> = Vec::new();
2160 let mut seen = std::collections::HashSet::new();
2161 let mut seen_home = std::collections::HashSet::new();
2162 for home in homes {
2163 if !seen_home.insert(home.clone()) {
2164 continue;
2165 }
2166 for n in vak_core::memory::list_notes(&home, core.cwd()) {
2167 if seen.insert((n.kind.clone(), n.tag.clone(), n.text.clone())) {
2168 blocks.push(note_payload(&n, "workspace"));
2169 }
2170 }
2171 for n in vak_core::memory::list_profile_notes(&home) {
2172 if seen.insert((n.kind.clone(), n.tag.clone(), n.text.clone())) {
2173 blocks.push(note_payload(&n, "profile"));
2174 }
2175 }
2176 }
2177 Json(serde_json::json!({ "notes": blocks })).into_response()
2178}
2179
2180async fn cleanup_memory(
2181 State(state): State<AppState>,
2182 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2183) -> axum::response::Response {
2184 use axum::response::IntoResponse;
2185 let core = scoped_core!(&state, None, q.agent.as_deref());
2186 let mut homes = vec![core.sessions_home(), state.core.sessions_home()];
2187 let shared = state.core.shared_data_home();
2188 if !homes.contains(&shared) {
2189 homes.push(shared);
2190 }
2191 let mut report = vak_core::memory::CleanupReport::default();
2192 let mut seen_home = std::collections::HashSet::new();
2193 for home in homes {
2194 if !seen_home.insert(home.clone()) {
2195 continue;
2196 }
2197 let home_report =
2198 vak_core::memory::cleanup_artifacts(&home, std::time::Duration::from_secs(86_400));
2199 report.removed_locks += home_report.removed_locks;
2200 report.removed_temps += home_report.removed_temps;
2201 report.removed_empty_dirs += home_report.removed_empty_dirs;
2202 }
2203 vak_core::security_events::record(
2204 &core.sessions_home(),
2205 vak_core::security_events::EventKind::ConfigChange,
2206 "memory_cleanup",
2207 &format!(
2208 "locks={} temps={} empty_dirs={}",
2209 report.removed_locks, report.removed_temps, report.removed_empty_dirs
2210 ),
2211 None,
2212 );
2213 Json(serde_json::json!({
2214 "removed_locks": report.removed_locks,
2215 "removed_temps": report.removed_temps,
2216 "removed_empty_dirs": report.removed_empty_dirs,
2217 }))
2218 .into_response()
2219}
2220
2221async fn consolidate_memory_route(
2222 State(state): State<AppState>,
2223 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2224) -> axum::response::Response {
2225 use axum::response::IntoResponse;
2226 let core = scoped_core!(&state, None, q.agent.as_deref());
2227 match core.consolidate_memory() {
2228 Ok(report) => (
2229 StatusCode::OK,
2230 Json(serde_json::to_value(&report).unwrap_or_default()),
2231 )
2232 .into_response(),
2233 Err(err) => (
2234 StatusCode::INTERNAL_SERVER_ERROR,
2235 Json(serde_json::json!({ "error": err })),
2236 )
2237 .into_response(),
2238 }
2239}
2240
2241#[derive(serde::Deserialize)]
2242struct ListEntitiesQuery {
2243 #[serde(default)]
2244 q: Option<String>,
2245 #[serde(default)]
2246 scope: Option<String>,
2247}
2248
2249async fn list_entities_route(
2250 State(state): State<AppState>,
2251 axum::extract::Query(query): axum::extract::Query<ListEntitiesQuery>,
2252) -> axum::response::Response {
2253 use axum::response::IntoResponse;
2254 let home = state.core.sessions_home();
2255 let is_global = query.scope.as_deref() == Some("global");
2256 let cwd_buf = state.core.cwd();
2257 let cwd = if is_global {
2258 None
2259 } else {
2260 Some(cwd_buf.as_path())
2261 };
2262 let entities = if let Some(ref q) = query.q {
2263 vak_core::entities::search_entities(&home, cwd, q)
2264 } else {
2265 vak_core::entities::list_entities(&home, cwd)
2266 };
2267 (
2268 StatusCode::OK,
2269 Json(serde_json::json!({ "entities": entities })),
2270 )
2271 .into_response()
2272}
2273
2274async fn get_entity_route(
2275 State(state): State<AppState>,
2276 Path(id): Path<String>,
2277 axum::extract::Query(query): axum::extract::Query<ListEntitiesQuery>,
2278) -> axum::response::Response {
2279 use axum::response::IntoResponse;
2280 let home = state.core.sessions_home();
2281 let is_global = query.scope.as_deref() == Some("global");
2282 let cwd_buf = state.core.cwd();
2283 let cwd = if is_global {
2284 None
2285 } else {
2286 Some(cwd_buf.as_path())
2287 };
2288 if let Some(entity) = vak_core::entities::get_entity(&home, cwd, &id) {
2289 (
2290 StatusCode::OK,
2291 Json(serde_json::to_value(&entity).unwrap_or_default()),
2292 )
2293 .into_response()
2294 } else {
2295 (
2296 StatusCode::NOT_FOUND,
2297 Json(serde_json::json!({ "error": "entity not found" })),
2298 )
2299 .into_response()
2300 }
2301}
2302
2303#[derive(serde::Deserialize)]
2304struct UpsertEntityBody {
2305 #[serde(default)]
2306 id: Option<String>,
2307 name: String,
2308 entity_type: String,
2309 #[serde(default)]
2310 summary: String,
2311 #[serde(default)]
2312 attributes: std::collections::BTreeMap<String, String>,
2313 #[serde(default)]
2314 relations: Vec<vak_core::entities::EntityRelation>,
2315 #[serde(default)]
2316 scope: Option<String>,
2317}
2318
2319async fn upsert_entity_route(
2320 State(state): State<AppState>,
2321 Json(body): Json<UpsertEntityBody>,
2322) -> axum::response::Response {
2323 use axum::response::IntoResponse;
2324 let home = state.core.sessions_home();
2325 let is_global = body.scope.as_deref() == Some("global");
2326 let cwd_buf = state.core.cwd();
2327 let cwd = if is_global {
2328 None
2329 } else {
2330 Some(cwd_buf.as_path())
2331 };
2332 let id = body.id.unwrap_or_else(|| {
2333 let slug = body
2334 .name
2335 .to_ascii_lowercase()
2336 .chars()
2337 .map(|c| if c.is_alphanumeric() { c } else { '-' })
2338 .collect::<String>()
2339 .trim_matches('-')
2340 .to_string();
2341 if slug.is_empty() {
2342 uuid::Uuid::now_v7().to_string()
2343 } else {
2344 slug
2345 }
2346 });
2347
2348 let record = vak_core::entities::EntityRecord {
2349 id,
2350 name: body.name,
2351 entity_type: body.entity_type,
2352 summary: body.summary,
2353 attributes: body.attributes,
2354 relations: body.relations,
2355 updated_at: chrono::Utc::now(),
2356 };
2357
2358 match vak_core::entities::upsert_entity(&home, cwd, record) {
2359 Ok(saved) => (
2360 StatusCode::OK,
2361 Json(serde_json::to_value(&saved).unwrap_or_default()),
2362 )
2363 .into_response(),
2364 Err(err) => (
2365 StatusCode::INTERNAL_SERVER_ERROR,
2366 Json(serde_json::json!({ "error": err.to_string() })),
2367 )
2368 .into_response(),
2369 }
2370}
2371
2372async fn delete_entity_route(
2373 State(state): State<AppState>,
2374 Path(id): Path<String>,
2375 axum::extract::Query(query): axum::extract::Query<ListEntitiesQuery>,
2376) -> axum::response::Response {
2377 use axum::response::IntoResponse;
2378 let home = state.core.sessions_home();
2379 let is_global = query.scope.as_deref() == Some("global");
2380 let cwd_buf = state.core.cwd();
2381 let cwd = if is_global {
2382 None
2383 } else {
2384 Some(cwd_buf.as_path())
2385 };
2386 match vak_core::entities::delete_entity(&home, cwd, &id) {
2387 Ok(true) => (StatusCode::OK, Json(serde_json::json!({ "deleted": true }))).into_response(),
2388 Ok(false) => (
2389 StatusCode::NOT_FOUND,
2390 Json(serde_json::json!({ "deleted": false, "error": "entity not found" })),
2391 )
2392 .into_response(),
2393 Err(err) => (
2394 StatusCode::INTERNAL_SERVER_ERROR,
2395 Json(serde_json::json!({ "error": err.to_string() })),
2396 )
2397 .into_response(),
2398 }
2399}
2400
2401/// Resolve a note id to the markdown store it lives in. The workspace tier/// is per-cwd; the profile tier is global (`<home>/memory/user/USER.md`).
2402#[derive(serde::Deserialize)]
2403struct AppendMemoryBody {
2404 text: String,
2405 #[serde(default)]
2406 kind: Option<String>,
2407 #[serde(default)]
2408 tag: Option<String>,
2409 #[serde(default)]
2410 scope: Option<MemoryScope>,
2411 #[serde(default)]
2412 session_id: Option<String>,
2413 /// Which Agent this note belongs to; see `AgentScopeQuery`. Absent
2414 /// means "vak", unless `session_id` names a currently-registered
2415 /// session, whose own Agent takes precedence (see `resolve_scoped_core`).
2416 #[serde(default)]
2417 agent: Option<String>,
2418}
2419
2420/// Append a note to either tier. Keeps gateway/desktop/CLI symmetric —
2421/// every surface writes through the same validated core API.
2422async fn append_memory(
2423 State(state): State<AppState>,
2424 Json(body): Json<AppendMemoryBody>,
2425) -> axum::response::Response {
2426 use axum::response::IntoResponse;
2427 let core = scoped_core!(&state, body.session_id.as_deref(), body.agent.as_deref());
2428 let home = core.sessions_home();
2429 let scope = body.scope.unwrap_or(MemoryScope::Workspace);
2430 let kind = body.kind.unwrap_or_else(|| "fact".to_string());
2431 let tag = body.tag.unwrap_or_default();
2432 let session = body.session_id.unwrap_or_else(|| "http".to_string());
2433 let result = match scope {
2434 MemoryScope::Workspace => {
2435 vak_core::memory::append_note(&home, core.cwd(), &kind, &tag, &session, &body.text)
2436 }
2437 MemoryScope::Profile => {
2438 vak_core::memory::append_profile_note(&home, &kind, &tag, &body.text, &session)
2439 }
2440 };
2441 match result {
2442 Ok(note) => {
2443 let scope_str = match scope {
2444 MemoryScope::Workspace => "workspace",
2445 MemoryScope::Profile => "profile",
2446 };
2447 vak_core::security_events::record(
2448 &home,
2449 vak_core::security_events::EventKind::ConfigChange,
2450 "memory_append",
2451 &format!("scope={scope_str} note_id={}", note.id),
2452 None,
2453 );
2454 (StatusCode::CREATED, Json(note_payload(&note, scope_str))).into_response()
2455 }
2456 Err(e) => (
2457 StatusCode::BAD_REQUEST,
2458 Json(serde_json::json!({ "error": e })),
2459 )
2460 .into_response(),
2461 }
2462}
2463
2464fn memory_store_path(core: &vak_core::Core, scope: MemoryScope) -> PathBuf {
2465 let home = core.sessions_home();
2466 match scope {
2467 MemoryScope::Workspace => home
2468 .join("memory")
2469 .join(vak_core::memory::hash_cwd(core.cwd()))
2470 .join("MEMORY.md"),
2471 MemoryScope::Profile => vak_core::memory::profile_path(&home),
2472 }
2473}
2474
2475#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Deserialize, Default)]
2476#[serde(rename_all = "lowercase")]
2477enum MemoryScope {
2478 #[default]
2479 Workspace,
2480 Profile,
2481}
2482
2483async fn forget_memory_note(
2484 State(state): State<AppState>,
2485 Path(note_id): Path<String>,
2486 axum::extract::Query(q): axum::extract::Query<MemoryScopeQuery>,
2487) -> axum::response::Response {
2488 use axum::response::IntoResponse;
2489 let core = scoped_core!(&state, None, q.agent.as_deref());
2490 let path = memory_store_path(&core, q.scope.unwrap_or_default());
2491 let result = vak_core::memory::forget_note(&path, &note_id);
2492 // A note written before this Agent's data moved to its own sessions_home
2493 // subfolder still lives at the legacy, un-scoped path — fall back to it
2494 // the same way `promote_proposal`/`reject_proposal` already do, so an
2495 // old note surfaced by `list_memory`'s merged view can still be forgotten.
2496 let legacy_path = memory_store_path(&state.core, q.scope.unwrap_or_default());
2497 let result = match result {
2498 Err(_) if legacy_path != path => vak_core::memory::forget_note(&legacy_path, &note_id),
2499 other => other,
2500 };
2501 match result {
2502 Ok(bytes) => {
2503 vak_core::security_events::record(
2504 &core.sessions_home(),
2505 vak_core::security_events::EventKind::ConfigChange,
2506 "memory_forget",
2507 &format!("scope={:?} note_id={note_id}", q.scope.unwrap_or_default()),
2508 None,
2509 );
2510 (
2511 StatusCode::OK,
2512 Json(serde_json::json!({ "forgotten": note_id, "bytes": bytes })),
2513 )
2514 .into_response()
2515 }
2516 Err(e) => (
2517 StatusCode::NOT_FOUND,
2518 Json(serde_json::json!({ "error": e })),
2519 )
2520 .into_response(),
2521 }
2522}
2523
2524#[derive(serde::Deserialize)]
2525struct MemoryAmendBody {
2526 text: String,
2527 #[serde(default)]
2528 scope: Option<MemoryScope>,
2529 /// See `AgentScopeQuery`.
2530 #[serde(default)]
2531 agent: Option<String>,
2532}
2533
2534#[derive(serde::Deserialize, Default)]
2535struct MemoryScopeQuery {
2536 #[serde(default)]
2537 scope: Option<MemoryScope>,
2538 /// See `AgentScopeQuery`.
2539 #[serde(default)]
2540 agent: Option<String>,
2541}
2542
2543async fn amend_memory_note(
2544 State(state): State<AppState>,
2545 Path(note_id): Path<String>,
2546 Json(body): Json<MemoryAmendBody>,
2547) -> axum::response::Response {
2548 use axum::response::IntoResponse;
2549 if body.text.trim().is_empty() {
2550 return (
2551 StatusCode::BAD_REQUEST,
2552 Json(serde_json::json!({ "error": "note must not be empty" })),
2553 )
2554 .into_response();
2555 }
2556 let core = scoped_core!(&state, None, body.agent.as_deref());
2557 let path = memory_store_path(&core, body.scope.unwrap_or_default());
2558 let result = vak_core::memory::amend_note(&path, &note_id, &body.text);
2559 // See the identical fallback in `forget_memory_note`.
2560 let legacy_path = memory_store_path(&state.core, body.scope.unwrap_or_default());
2561 let result = match result {
2562 Err(_) if legacy_path != path => {
2563 vak_core::memory::amend_note(&legacy_path, &note_id, &body.text)
2564 }
2565 other => other,
2566 };
2567 match result {
2568 Ok(()) => {
2569 vak_core::security_events::record(
2570 &core.sessions_home(),
2571 vak_core::security_events::EventKind::ConfigChange,
2572 "memory_amend",
2573 &format!(
2574 "scope={:?} note_id={note_id}",
2575 body.scope.unwrap_or_default()
2576 ),
2577 None,
2578 );
2579 (
2580 StatusCode::OK,
2581 Json(serde_json::json!({ "amended": note_id })),
2582 )
2583 .into_response()
2584 }
2585 Err(e) => (
2586 StatusCode::NOT_FOUND,
2587 Json(serde_json::json!({ "error": e })),
2588 )
2589 .into_response(),
2590 }
2591}
2592
2593fn proposals_payload(core: &Core) -> Vec<serde_json::Value> {
2594 let mut proposals = vak_core::learning::list_proposals(&core.sessions_home(), core.cwd());
2595 if proposals.is_empty() && core.shared_data_home() != core.sessions_home() {
2596 proposals = vak_core::learning::list_proposals(&core.shared_data_home(), core.cwd());
2597 }
2598 proposals
2599 .iter()
2600 .map(|p| {
2601 serde_json::json!({
2602 "id": p.id,
2603 "name": p.name,
2604 "description": p.description,
2605 })
2606 })
2607 .collect()
2608}
2609
2610async fn list_proposals_route(
2611 State(state): State<AppState>,
2612 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2613) -> axum::response::Response {
2614 use axum::response::IntoResponse;
2615 let core = scoped_core!(&state, None, q.agent.as_deref());
2616 Json(serde_json::json!({ "proposals": proposals_payload(&core) })).into_response()
2617}
2618
2619async fn promote_proposal(
2620 State(state): State<AppState>,
2621 Path(id): Path<String>,
2622 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2623) -> axum::response::Response {
2624 use axum::response::IntoResponse;
2625 let core = scoped_core!(&state, None, q.agent.as_deref());
2626 let res = vak_core::learning::promote(&core.sessions_home(), core.cwd(), &id);
2627 let res = match res {
2628 Err(_) if state.core.shared_data_home() != core.sessions_home() => {
2629 vak_core::learning::promote(&state.core.shared_data_home(), core.cwd(), &id)
2630 }
2631 other => other,
2632 };
2633 match res {
2634 Ok(name) => (
2635 StatusCode::OK,
2636 Json(serde_json::json!({ "promoted": name })),
2637 )
2638 .into_response(),
2639 Err(e) => (
2640 StatusCode::NOT_FOUND,
2641 Json(serde_json::json!({ "error": e })),
2642 )
2643 .into_response(),
2644 }
2645}
2646
2647async fn reject_proposal(
2648 State(state): State<AppState>,
2649 Path(id): Path<String>,
2650 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2651) -> axum::response::Response {
2652 use axum::response::IntoResponse;
2653 let core = scoped_core!(&state, None, q.agent.as_deref());
2654 let res = vak_core::learning::reject(&core.sessions_home(), core.cwd(), &id);
2655 let res = match res {
2656 Err(_) if state.core.shared_data_home() != core.sessions_home() => {
2657 vak_core::learning::reject(&state.core.shared_data_home(), core.cwd(), &id)
2658 }
2659 other => other,
2660 };
2661 match res {
2662 Ok(()) => (StatusCode::OK, Json(serde_json::json!({ "rejected": id }))).into_response(),
2663 Err(e) => (
2664 StatusCode::NOT_FOUND,
2665 Json(serde_json::json!({ "error": e })),
2666 )
2667 .into_response(),
2668 }
2669}
2670
2671#[derive(serde::Deserialize)]
2672struct SearchQuery {
2673 q: String,
2674 #[serde(default)]
2675 limit: Option<usize>,
2676 /// Session id whose (already-in-context) content should be skipped.
2677 #[serde(default)]
2678 exclude: Option<String>,
2679 /// Cross-project recall: search every project's ledgers under the
2680 /// sessions home (docs/design/29-personal-os.md P1), not just this cwd.
2681 #[serde(default)]
2682 all: bool,
2683 /// The Agent whose ledgers and memory are searched (default `vak`).
2684 /// Each Agent's memory is private (AGENTS.md invariant 37), so a search
2685 /// resolves one Agent the way `/memory` does and reads only its home.
2686 #[serde(default)]
2687 agent: Option<String>,
2688}
2689
2690async fn search_sessions(
2691 State(state): State<AppState>,
2692 axum::extract::Query(q): axum::extract::Query<SearchQuery>,
2693) -> axum::response::Response {
2694 use axum::response::IntoResponse;
2695 let core = scoped_core!(&state, None, q.agent.as_deref());
2696 let home = core.sessions_home();
2697 let cwd = core.cwd().clone();
2698 let query = q.q.clone();
2699 let limit = q.limit.unwrap_or(vak_session::DEFAULT_LIMIT);
2700 let excluded =
2701 vak_core::trash::search_exclusions(&core.shared_data_home(), q.exclude.as_deref());
2702 let all = q.all;
2703 let mut extras = Vec::new();
2704 let mut workspace_notes = vak_core::memory::list_notes(&home, &cwd);
2705 if all {
2706 let root = home.join("memory");
2707 if let Ok(entries) = std::fs::read_dir(&root) {
2708 workspace_notes.clear();
2709 for entry in entries.flatten() {
2710 let path = entry.path().join("MEMORY.md");
2711 if let Ok(raw) = std::fs::read_to_string(path) {
2712 workspace_notes.extend(vak_core::memory::parse_blocks(&raw));
2713 }
2714 }
2715 }
2716 }
2717 for note in workspace_notes {
2718 let id = if note.tag.is_empty() {
2719 note.id.clone()
2720 } else {
2721 note.tag.clone()
2722 };
2723 extras.push(vak_session::ExternalDoc {
2724 id,
2725 text: format!("[{}] {}", note.kind, note.text),
2726 ts: Some(note.ts),
2727 role: Some("memory".into()),
2728 });
2729 }
2730 for note in vak_core::memory::list_profile_notes(&home) {
2731 let id = format!(
2732 "profile/{}",
2733 if note.tag.is_empty() {
2734 note.id.clone()
2735 } else {
2736 note.tag.clone()
2737 }
2738 );
2739 extras.push(vak_session::ExternalDoc {
2740 id,
2741 text: format!("[{}] {}", note.kind, note.text),
2742 ts: Some(note.ts),
2743 role: Some("profile".into()),
2744 });
2745 }
2746 match tokio::task::spawn_blocking(move || {
2747 // Both hit shapes are Serialize; the workspace path keeps its flat
2748 // SessionHit wire shape, cross-project adds the project_hash wrapper.
2749 let searched = if all {
2750 vak_session::search_all_extended(&home, &query, limit, &excluded, &extras)
2751 .map(|hits| serde_json::to_value(&hits).map_err(|e| e.to_string()))
2752 } else {
2753 vak_session::search_extended(&home, &cwd, &query, limit, &excluded, &extras)
2754 .map(|hits| serde_json::to_value(&hits).map_err(|e| e.to_string()))
2755 };
2756 match searched {
2757 Ok(inner) => inner,
2758 Err(e) => Err(e.to_string()),
2759 }
2760 })
2761 .await
2762 {
2763 Ok(Ok(hits)) => Json(serde_json::json!({ "all": all, "hits": hits })).into_response(),
2764 Ok(Err(e)) => (
2765 StatusCode::INTERNAL_SERVER_ERROR,
2766 Json(serde_json::json!({ "error": e })),
2767 )
2768 .into_response(),
2769 Err(e) => (
2770 StatusCode::INTERNAL_SERVER_ERROR,
2771 Json(serde_json::json!({ "error": e.to_string() })),
2772 )
2773 .into_response(),
2774 }
2775}
2776
2777/// The bearer token lives for the life of the process; embedders (desktop
2778/// shell, tests) need it to hand to their webview, so build the secured
2779/// stack here instead of inside `serve()`.
2780pub fn secured_router(core: Core) -> (Router, String) {
2781 secured_router_with(core, false)
2782}
2783
2784/// Same stack with a CLI-level gateway override (`serve --gateway`).
2785///
2786/// Token selection: when `VAK_GATEWAY_TOKEN` is set in the
2787/// environment, it is used verbatim so service-managed bridges and other
2788/// long-lived clients can survive process restarts. Otherwise a fresh
2789/// per-process token is minted as before. The variable is never logged.
2790pub fn secured_router_with(core: Core, force_gateway: bool) -> (Router, String) {
2791 secured_router_with_port(core, force_gateway, vak_ops::OpsConfig::detect().port)
2792}
2793
2794/// Same secured stack with the actual listener port carried into operational
2795/// probes. `serve_with` uses this so a non-default `--port` cannot make the
2796/// console probe a different process.
2797pub fn secured_router_with_port(core: Core, force_gateway: bool, port: u16) -> (Router, String) {
2798 // Tauri can use either its custom scheme or the loopback-style origin,
2799 // depending on the platform and WebView runtime, plus vite dev servers.
2800 let origins = [
2801 "tauri://localhost",
2802 "http://tauri.localhost",
2803 "https://tauri.localhost",
2804 "http://localhost:1420",
2805 "http://127.0.0.1:1420",
2806 "http://localhost:5173",
2807 "http://127.0.0.1:5173",
2808 ]
2809 .into_iter()
2810 .filter_map(|o| o.parse::<axum::http::HeaderValue>().ok())
2811 .collect::<Vec<_>>();
2812 let cors = tower_http::cors::CorsLayer::new()
2813 .allow_origin(origins)
2814 // Must cover every method the router exposes: PATCH (/config,
2815 // /sessions/:id/config) and DELETE are preflighted, so omitting them
2816 // makes the browser reject the request before it is ever sent.
2817 .allow_methods([
2818 axum::http::Method::GET,
2819 axum::http::Method::POST,
2820 axum::http::Method::PUT,
2821 axum::http::Method::PATCH,
2822 axum::http::Method::DELETE,
2823 ])
2824 .allow_headers([
2825 axum::http::header::AUTHORIZATION,
2826 axum::http::header::CONTENT_TYPE,
2827 ]);
2828 let mut state = AppState::new(core);
2829 state.ops_port = port;
2830 if force_gateway {
2831 state.enable_gateway();
2832 }
2833 let token = (*state.auth_token).clone();
2834 let rl_settings = state.core.config().gateway.rate_limit.clone();
2835 let rl_config = rate_limit::RateLimitConfig::from_settings(rl_settings);
2836 let limiter = rate_limit::RateLimiter::new(rl_config, state.core.sessions_home());
2837 let app = router_with_state(state.clone())
2838 .layer(axum::middleware::from_fn_with_state(
2839 limiter,
2840 rate_limit::rate_limit_layer,
2841 ))
2842 .layer(axum::middleware::from_fn_with_state(
2843 state.clone(),
2844 enforce_participant_audience,
2845 ))
2846 .layer(axum::middleware::from_fn_with_state(
2847 AuthPolicy {
2848 token: token.clone(),
2849 home: state.core.sessions_home(),
2850 trusted_hosts: state.core.config().server.trusted_hosts.clone(),
2851 },
2852 require_bearer,
2853 ))
2854 .layer(cors);
2855 #[cfg(unix)]
2856 {
2857 let broker = state.core.agent_network_broker();
2858 let policy_file = state
2859 .core
2860 .sessions_home()
2861 .join("agent-network/policies.json");
2862 if let Err(error) = broker.load_policies(&policy_file) {
2863 eprintln!("[agent-network] policy load failed: {error}");
2864 }
2865 let socket =
2866 vak_core::agent_network::AgentNetworkBroker::socket_path(&state.core.sessions_home());
2867 tokio::spawn(async move {
2868 if let Err(error) = broker.serve_unix(&socket).await {
2869 eprintln!("[agent-network] broker stopped: {error}");
2870 }
2871 });
2872 }
2873 // Local routines: fires due scheduled tasks while this server lives.
2874 start_scheduler(&state);
2875 delivery::start_replay(&state.core);
2876 // Capability discovery is NOT started here.
2877 //
2878 // It used to be, and the CLI did its own bounded wait, and the desktop
2879 // did neither — three surfaces answering "when may a prompt be frozen?"
2880 // three different ways, which is how an admitted, working MCP server
2881 // still produced a session that had never seen its catalog.
2882 // `Core::admitted_capabilities` owns that decision now, so every surface
2883 // gets the same packet whether it was reached from a terminal, this
2884 // server, or the desktop app.
2885 // Background index sync: keeps the admin console populated from the
2886 // very first boot. Idempotent; never blocks request handling.
2887 if let Some(store) = state.store.clone() {
2888 let home = state.core.sessions_home();
2889 tokio::spawn(async move {
2890 match store.rebuild(&home) {
2891 Ok(s) if s.files_scanned > 0 => eprintln!(
2892 "[store] indexed {} files / {} entries",
2893 s.files_scanned, s.entries_indexed
2894 ),
2895 Ok(_) => {}
2896 Err(e) => eprintln!("[store] startup rebuild failed: {e}"),
2897 }
2898 });
2899 }
2900 (app, token)
2901}
2902
2903fn participant_matches_conversation_audience(
2904 participant: &coworking::VerifiedPrincipal,
2905 conversation_id: &str,
2906 observed_audience: Option<&str>,
2907) -> bool {
2908 participant.conversation_id == conversation_id
2909 && observed_audience == Some(participant.audience_id.as_str())
2910}
2911
2912/// Recheck the live conversation audience after bearer authentication and
2913/// before any scoped handler runs. The route whitelist limits *where* a
2914/// participant may go; this boundary also proves the durable grant still
2915/// names the audience owned by that conversation.
2916async fn enforce_participant_audience(
2917 State(state): State<AppState>,
2918 req: axum::extract::Request,
2919 next: axum::middleware::Next,
2920) -> axum::response::Response {
2921 if let Some(AuthenticatedPrincipal::Participant(participant)) =
2922 req.extensions().get::<AuthenticatedPrincipal>()
2923 {
2924 let conversation_id = req
2925 .uri()
2926 .path()
2927 .trim_matches('/')
2928 .split('/')
2929 .nth(1)
2930 .unwrap_or_default();
2931 let audience = conversation_audience(&state, conversation_id);
2932 if !participant_matches_conversation_audience(
2933 participant,
2934 conversation_id,
2935 audience.as_deref(),
2936 ) {
2937 return StatusCode::FORBIDDEN.into_response();
2938 }
2939 }
2940 next.run(req).await
2941}
2942
2943/// Initialize the distributed event bus from config and attach it to the
2944/// global `EventHub`. Called from `serve_with` (async) so NATS connection
2945/// retries don't block request handling.
2946pub async fn init_server_bus(core: &Core) {
2947 let config = core.config().server.bus.clone();
2948 if config.nats_url.is_none() {
2949 return;
2950 }
2951 install_server_bus(core, &config).await;
2952}
2953
2954/// Builds the bus `config` describes, with its secrets from the secrets
2955/// chain, and makes it the live one.
2956async fn install_server_bus(core: &Core, config: &vak_config::BusResolved) {
2957 let sessions_home = core.sessions_home();
2958 let workspace_id = {
2959 use sha2::{Digest, Sha256};
2960 let mut hasher = Sha256::new();
2961 hasher.update(sessions_home.to_string_lossy().as_bytes());
2962 let result = hasher.finalize();
2963 // First 6 bytes → 12 hex chars, a compact workspace-scoped subject prefix.
2964 let prefix = &result[..6];
2965 let hex: String = prefix.iter().map(|b| format!("{b:02x}")).collect();
2966 format!("ws_{}", hex)
2967 };
2968 let bus = crate::bus::ServerBus::from_resolved(
2969 &workspace_id,
2970 config,
2971 &crate::bus::BusCredentials::of(core),
2972 )
2973 .await;
2974 if let Some(mut hub) = crate::events::global() {
2975 hub.set_server_bus(std::sync::Arc::new(bus));
2976 }
2977}
2978
2979pub async fn serve(core: Core, addr: std::net::SocketAddr) -> std::io::Result<()> {
2980 serve_with(core, addr, false).await
2981}
2982
2983/// `force_gateway` mirrors `serve --gateway`: enable routing regardless of
2984/// the (untrusted-stripped) project config.
2985/// Serve an already-bound listener with an already-built router.
2986///
2987/// The pieces `secured_router` returns, joined. `vak setup` binds its own
2988/// ephemeral loopback port so it can print the URL *before* serving, and
2989/// needs the token from the same call — which `serve_with` cannot give it,
2990/// because that mints and consumes the token internally. Exposed here so
2991/// axum stays a dependency of this crate rather than leaking into the CLI.
2992pub async fn serve_router(listener: tokio::net::TcpListener, app: Router) -> std::io::Result<()> {
2993 axum::serve(listener, app).await
2994}
2995
2996pub async fn serve_with(
2997 core: Core,
2998 addr: std::net::SocketAddr,
2999 force_gateway: bool,
3000) -> std::io::Result<()> {

2001 }
2002 "install" => match vak_ops::install(svc, &cfg) {
2003 Ok(()) => (serde_json::json!({ "ok": true, "action": "install" }), true),
2004 Err(e) => (serde_json::json!({ "ok": false, "error": e }), false),
2005 },
2006 "uninstall" => match vak_ops::uninstall(svc, &cfg) {
2007 Ok(()) => (
2008 serde_json::json!({ "ok": true, "action": "uninstall" }),
2009 true,
2010 ),
2011 Err(e) => (serde_json::json!({ "ok": false, "error": e }), false),
2012 },
2013 other => {
2014 return (
2015 StatusCode::BAD_REQUEST,
2016 Json(serde_json::json!({ "error": format!("unknown action '{other}'") })),
2017 )
2018 .into_response();
2019 }
2020 };
2021 let after = service_control::status(svc, cfg.clone())
2022 .await
2023 .map(|s| s.to_string())
2024 .unwrap_or_else(|e| e);
2025 let desired_reached = match action.as_str() {
2026 "start" | "restart" | "install" => after == "running",
2027 "stop" => matches!(after.as_str(), "stopped" | "not installed"),
2028 "uninstall" => after == "not installed",
2029 _ => false,
2030 };
2031 let verification_status = if !succeeded {
2032 "failed"
2033 } else if desired_reached {
2034 "verified"
2035 } else {
2036 "pending"
2037 };
2038 let verification_detail = if !succeeded {
2039 "The service manager rejected the requested operation; the post-action probe is authoritative."
2040 } else if desired_reached {
2041 "The post-action service-manager probe reached the requested state."
2042 } else {
2043 "The manager accepted the request but the desired state is not visible yet; keep the receipt and re-probe."
2044 };
2045 let mut receipt = operations::ActionReceipt {
2046 receipt_id: format!("OP-{}", uuid::Uuid::now_v7().simple()),
2047 service: service.clone(),
2048 action: action.clone(),
2049 requested_at,
2050 completed_at: Utc::now(),
2051 succeeded,
2052 verification: operations::ActionVerification {
2053 status: verification_status.to_string(),
2054 before,
2055 after,
2056 detail: verification_detail.to_string(),
2057 },
2058 persisted: false,
2059 };
2060 receipt.persisted = operations::record_action(&state.core.sessions_home(), &receipt).is_ok();
2061 let receipt_json = serde_json::to_value(&receipt).unwrap_or_else(|_| serde_json::json!({}));
2062 let mut result = result;
2063 if let Some(object) = result.as_object_mut() {
2064 object.insert(
2065 "receipt_id".to_string(),
2066 serde_json::json!(receipt.receipt_id),
2067 );
2068 object.insert(
2069 "verification".to_string(),
2070 receipt_json["verification"].clone(),
2071 );
2072 object.insert(
2073 "receipt_persisted".to_string(),
2074 serde_json::json!(receipt.persisted),
2075 );
2076 }
2077 if succeeded {
2078 vak_core::security_events::record(
2079 &state.core.sessions_home(),
2080 vak_core::security_events::EventKind::ConfigChange,
2081 "service_action",
2082 &format!("service={service} action={action}"),
2083 None,
2084 );
2085 state
2086 .hub
2087 .emit_config_changed("service_action", &format!("{service}:{action}"));
2088 (StatusCode::OK, Json(result)).into_response()
2089 } else {
2090 (StatusCode::CONFLICT, Json(result)).into_response()
2091 }
2092}
2093
2094fn note_payload(n: &vak_core::memory::NoteBlock, scope: &str) -> serde_json::Value {
2095 serde_json::json!({
2096 "id": n.id,
2097 "ts": n.ts.to_rfc3339(),
2098 "kind": n.kind,
2099 "tag": n.tag,
2100 "session_id": n.session_id,
2101 "text": n.text,
2102 "scope": scope,
2103 })
2104}
2105
2106/// Which Agent an endpoint scoped to "the currently open Agent's own data"
2107/// (memory, learning proposals) should resolve against. Absent means the
2108/// built-in "vak" Agent — the same default `agent_chats::open` uses.
2109#[derive(serde::Deserialize, Default)]
2110struct AgentScopeQuery {
2111 #[serde(default)]
2112 agent: Option<String>,
2113}
2114
2115/// Resolve the `Core` an Agent-scoped endpoint should read/write through.
2116///
2117/// A registered session already carries the exact `Core` it was opened
2118/// under (agent identity, isolated workspace, and now-agent-scoped
2119/// `sessions_home` all resolved once at `agent_chats::open` time) — reusing
2120/// it is cheaper and more precise than re-deriving identity from an id, so
2121/// `session_id` (when the caller already has one, e.g. `AppendMemoryBody`)
2122/// takes precedence over an explicit `agent` id.
2123///
2124/// This is the single place "which Agent's data does this endpoint mean"
2125/// gets decided, so a future endpoint scoped the same way calls this
2126/// instead of reading `state.core` directly and drifting out of sync with
2127/// `agent_chats::open` the way `list_sessions` once did (see commit
2128/// 7e6713c0 and its follow-up).
2129#[allow(clippy::result_large_err)]
2130fn resolve_scoped_core(
2131 state: &AppState,
2132 session_id: Option<&str>,
2133 agent: Option<&str>,
2134) -> Result<vak_core::Core, axum::response::Response> {
2135 if let Some(sid) = session_id
2136 && let Some(handle) = state.get(sid)
2137 {
2138 return Ok(handle.core.clone());
2139 }
2140 let id = agent.unwrap_or("vak");
2141 agent_chats::resolve_agent_core(state, id).map(|(_, core)| core)
2142}
2143
2144async fn list_memory(
2145 State(state): State<AppState>,
2146 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2147) -> axum::response::Response {
2148 use axum::response::IntoResponse;
2149 let core = scoped_core!(&state, None, q.agent.as_deref());
2150 // The resolved Agent's own (now agent-scoped) sessions_home is primary;
2151 // `state.core`'s plain, un-scoped home is kept as a fallback merge so
2152 // notes written before Agents carried their own sessions_home (or by an
2153 // older build) are not silently hidden.
2154 let mut homes = vec![core.sessions_home(), state.core.sessions_home()];
2155 let shared = state.core.shared_data_home();
2156 if !homes.contains(&shared) {
2157 homes.push(shared);
2158 }
2159 let mut blocks: Vec<serde_json::Value> = Vec::new();
2160 let mut seen = std::collections::HashSet::new();
2161 let mut seen_home = std::collections::HashSet::new();
2162 for home in homes {
2163 if !seen_home.insert(home.clone()) {
2164 continue;
2165 }
2166 for n in vak_core::memory::list_notes(&home, core.cwd()) {
2167 if seen.insert((n.kind.clone(), n.tag.clone(), n.text.clone())) {
2168 blocks.push(note_payload(&n, "workspace"));
2169 }
2170 }
2171 for n in vak_core::memory::list_profile_notes(&home) {
2172 if seen.insert((n.kind.clone(), n.tag.clone(), n.text.clone())) {
2173 blocks.push(note_payload(&n, "profile"));
2174 }
2175 }
2176 }
2177 Json(serde_json::json!({ "notes": blocks })).into_response()
2178}
2179
2180async fn cleanup_memory(
2181 State(state): State<AppState>,
2182 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2183) -> axum::response::Response {
2184 use axum::response::IntoResponse;
2185 let core = scoped_core!(&state, None, q.agent.as_deref());
2186 let mut homes = vec![core.sessions_home(), state.core.sessions_home()];
2187 let shared = state.core.shared_data_home();
2188 if !homes.contains(&shared) {
2189 homes.push(shared);
2190 }
2191 let mut report = vak_core::memory::CleanupReport::default();
2192 let mut seen_home = std::collections::HashSet::new();
2193 for home in homes {
2194 if !seen_home.insert(home.clone()) {
2195 continue;
2196 }
2197 let home_report =
2198 vak_core::memory::cleanup_artifacts(&home, std::time::Duration::from_secs(86_400));
2199 report.removed_locks += home_report.removed_locks;
2200 report.removed_temps += home_report.removed_temps;
2201 report.removed_empty_dirs += home_report.removed_empty_dirs;
2202 }
2203 vak_core::security_events::record(
2204 &core.sessions_home(),
2205 vak_core::security_events::EventKind::ConfigChange,
2206 "memory_cleanup",
2207 &format!(
2208 "locks={} temps={} empty_dirs={}",
2209 report.removed_locks, report.removed_temps, report.removed_empty_dirs
2210 ),
2211 None,
2212 );
2213 Json(serde_json::json!({
2214 "removed_locks": report.removed_locks,
2215 "removed_temps": report.removed_temps,
2216 "removed_empty_dirs": report.removed_empty_dirs,
2217 }))
2218 .into_response()
2219}
2220
2221async fn consolidate_memory_route(
2222 State(state): State<AppState>,
2223 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2224) -> axum::response::Response {
2225 use axum::response::IntoResponse;
2226 let core = scoped_core!(&state, None, q.agent.as_deref());
2227 match core.consolidate_memory() {
2228 Ok(report) => (
2229 StatusCode::OK,
2230 Json(serde_json::to_value(&report).unwrap_or_default()),
2231 )
2232 .into_response(),
2233 Err(err) => (
2234 StatusCode::INTERNAL_SERVER_ERROR,
2235 Json(serde_json::json!({ "error": err })),
2236 )
2237 .into_response(),
2238 }
2239}
2240
2241#[derive(serde::Deserialize)]
2242struct ListEntitiesQuery {
2243 #[serde(default)]
2244 q: Option<String>,
2245 #[serde(default)]
2246 scope: Option<String>,
2247}
2248
2249async fn list_entities_route(
2250 State(state): State<AppState>,
2251 axum::extract::Query(query): axum::extract::Query<ListEntitiesQuery>,
2252) -> axum::response::Response {
2253 use axum::response::IntoResponse;
2254 let home = state.core.sessions_home();
2255 let is_global = query.scope.as_deref() == Some("global");
2256 let cwd_buf = state.core.cwd();
2257 let cwd = if is_global {
2258 None
2259 } else {
2260 Some(cwd_buf.as_path())
2261 };
2262 let entities = if let Some(ref q) = query.q {
2263 vak_core::entities::search_entities(&home, cwd, q)
2264 } else {
2265 vak_core::entities::list_entities(&home, cwd)
2266 };
2267 (
2268 StatusCode::OK,
2269 Json(serde_json::json!({ "entities": entities })),
2270 )
2271 .into_response()
2272}
2273
2274async fn get_entity_route(
2275 State(state): State<AppState>,
2276 Path(id): Path<String>,
2277 axum::extract::Query(query): axum::extract::Query<ListEntitiesQuery>,
2278) -> axum::response::Response {
2279 use axum::response::IntoResponse;
2280 let home = state.core.sessions_home();
2281 let is_global = query.scope.as_deref() == Some("global");
2282 let cwd_buf = state.core.cwd();
2283 let cwd = if is_global {
2284 None
2285 } else {
2286 Some(cwd_buf.as_path())
2287 };
2288 if let Some(entity) = vak_core::entities::get_entity(&home, cwd, &id) {
2289 (
2290 StatusCode::OK,
2291 Json(serde_json::to_value(&entity).unwrap_or_default()),
2292 )
2293 .into_response()
2294 } else {
2295 (
2296 StatusCode::NOT_FOUND,
2297 Json(serde_json::json!({ "error": "entity not found" })),
2298 )
2299 .into_response()
2300 }
2301}
2302
2303#[derive(serde::Deserialize)]
2304struct UpsertEntityBody {
2305 #[serde(default)]
2306 id: Option<String>,
2307 name: String,
2308 entity_type: String,
2309 #[serde(default)]
2310 summary: String,
2311 #[serde(default)]
2312 attributes: std::collections::BTreeMap<String, String>,
2313 #[serde(default)]
2314 relations: Vec<vak_core::entities::EntityRelation>,
2315 #[serde(default)]
2316 scope: Option<String>,
2317}
2318
2319async fn upsert_entity_route(
2320 State(state): State<AppState>,
2321 Json(body): Json<UpsertEntityBody>,
2322) -> axum::response::Response {
2323 use axum::response::IntoResponse;
2324 let home = state.core.sessions_home();
2325 let is_global = body.scope.as_deref() == Some("global");
2326 let cwd_buf = state.core.cwd();
2327 let cwd = if is_global {
2328 None
2329 } else {
2330 Some(cwd_buf.as_path())
2331 };
2332 let id = body.id.unwrap_or_else(|| {
2333 let slug = body
2334 .name
2335 .to_ascii_lowercase()
2336 .chars()
2337 .map(|c| if c.is_alphanumeric() { c } else { '-' })
2338 .collect::<String>()
2339 .trim_matches('-')
2340 .to_string();
2341 if slug.is_empty() {
2342 uuid::Uuid::now_v7().to_string()
2343 } else {
2344 slug
2345 }
2346 });
2347
2348 let record = vak_core::entities::EntityRecord {
2349 id,
2350 name: body.name,
2351 entity_type: body.entity_type,
2352 summary: body.summary,
2353 attributes: body.attributes,
2354 relations: body.relations,
2355 updated_at: chrono::Utc::now(),
2356 };
2357
2358 match vak_core::entities::upsert_entity(&home, cwd, record) {
2359 Ok(saved) => (
2360 StatusCode::OK,
2361 Json(serde_json::to_value(&saved).unwrap_or_default()),
2362 )
2363 .into_response(),
2364 Err(err) => (
2365 StatusCode::INTERNAL_SERVER_ERROR,
2366 Json(serde_json::json!({ "error": err.to_string() })),
2367 )
2368 .into_response(),
2369 }
2370}
2371
2372async fn delete_entity_route(
2373 State(state): State<AppState>,
2374 Path(id): Path<String>,
2375 axum::extract::Query(query): axum::extract::Query<ListEntitiesQuery>,
2376) -> axum::response::Response {
2377 use axum::response::IntoResponse;
2378 let home = state.core.sessions_home();
2379 let is_global = query.scope.as_deref() == Some("global");
2380 let cwd_buf = state.core.cwd();
2381 let cwd = if is_global {
2382 None
2383 } else {
2384 Some(cwd_buf.as_path())
2385 };
2386 match vak_core::entities::delete_entity(&home, cwd, &id) {
2387 Ok(true) => (StatusCode::OK, Json(serde_json::json!({ "deleted": true }))).into_response(),
2388 Ok(false) => (
2389 StatusCode::NOT_FOUND,
2390 Json(serde_json::json!({ "deleted": false, "error": "entity not found" })),
2391 )
2392 .into_response(),
2393 Err(err) => (
2394 StatusCode::INTERNAL_SERVER_ERROR,
2395 Json(serde_json::json!({ "error": err.to_string() })),
2396 )
2397 .into_response(),
2398 }
2399}
2400
2401/// Resolve a note id to the markdown store it lives in. The workspace tier/// is per-cwd; the profile tier is global (`<home>/memory/user/USER.md`).
2402#[derive(serde::Deserialize)]
2403struct AppendMemoryBody {
2404 text: String,
2405 #[serde(default)]
2406 kind: Option<String>,
2407 #[serde(default)]
2408 tag: Option<String>,
2409 #[serde(default)]
2410 scope: Option<MemoryScope>,
2411 #[serde(default)]
2412 session_id: Option<String>,
2413 /// Which Agent this note belongs to; see `AgentScopeQuery`. Absent
2414 /// means "vak", unless `session_id` names a currently-registered
2415 /// session, whose own Agent takes precedence (see `resolve_scoped_core`).
2416 #[serde(default)]
2417 agent: Option<String>,
2418}
2419
2420/// Append a note to either tier. Keeps gateway/desktop/CLI symmetric —
2421/// every surface writes through the same validated core API.
2422async fn append_memory(
2423 State(state): State<AppState>,
2424 Json(body): Json<AppendMemoryBody>,
2425) -> axum::response::Response {
2426 use axum::response::IntoResponse;
2427 let core = scoped_core!(&state, body.session_id.as_deref(), body.agent.as_deref());
2428 let home = core.sessions_home();
2429 let scope = body.scope.unwrap_or(MemoryScope::Workspace);
2430 let kind = body.kind.unwrap_or_else(|| "fact".to_string());
2431 let tag = body.tag.unwrap_or_default();
2432 let session = body.session_id.unwrap_or_else(|| "http".to_string());
2433 let result = match scope {
2434 MemoryScope::Workspace => {
2435 vak_core::memory::append_note(&home, core.cwd(), &kind, &tag, &session, &body.text)
2436 }
2437 MemoryScope::Profile => {
2438 vak_core::memory::append_profile_note(&home, &kind, &tag, &body.text, &session)
2439 }
2440 };
2441 match result {
2442 Ok(note) => {
2443 let scope_str = match scope {
2444 MemoryScope::Workspace => "workspace",
2445 MemoryScope::Profile => "profile",
2446 };
2447 vak_core::security_events::record(
2448 &home,
2449 vak_core::security_events::EventKind::ConfigChange,
2450 "memory_append",
2451 &format!("scope={scope_str} note_id={}", note.id),
2452 None,
2453 );
2454 (StatusCode::CREATED, Json(note_payload(&note, scope_str))).into_response()
2455 }
2456 Err(e) => (
2457 StatusCode::BAD_REQUEST,
2458 Json(serde_json::json!({ "error": e })),
2459 )
2460 .into_response(),
2461 }
2462}
2463
2464fn memory_store_path(core: &vak_core::Core, scope: MemoryScope) -> PathBuf {
2465 let home = core.sessions_home();
2466 match scope {
2467 MemoryScope::Workspace => home
2468 .join("memory")
2469 .join(vak_core::memory::hash_cwd(core.cwd()))
2470 .join("MEMORY.md"),
2471 MemoryScope::Profile => vak_core::memory::profile_path(&home),
2472 }
2473}
2474
2475#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Deserialize, Default)]
2476#[serde(rename_all = "lowercase")]
2477enum MemoryScope {
2478 #[default]
2479 Workspace,
2480 Profile,
2481}
2482
2483async fn forget_memory_note(
2484 State(state): State<AppState>,
2485 Path(note_id): Path<String>,
2486 axum::extract::Query(q): axum::extract::Query<MemoryScopeQuery>,
2487) -> axum::response::Response {
2488 use axum::response::IntoResponse;
2489 let core = scoped_core!(&state, None, q.agent.as_deref());
2490 let path = memory_store_path(&core, q.scope.unwrap_or_default());
2491 let result = vak_core::memory::forget_note(&path, &note_id);
2492 // A note written before this Agent's data moved to its own sessions_home
2493 // subfolder still lives at the legacy, un-scoped path — fall back to it
2494 // the same way `promote_proposal`/`reject_proposal` already do, so an
2495 // old note surfaced by `list_memory`'s merged view can still be forgotten.
2496 let legacy_path = memory_store_path(&state.core, q.scope.unwrap_or_default());
2497 let result = match result {
2498 Err(_) if legacy_path != path => vak_core::memory::forget_note(&legacy_path, &note_id),
2499 other => other,
2500 };
2501 match result {
2502 Ok(bytes) => {
2503 vak_core::security_events::record(
2504 &core.sessions_home(),
2505 vak_core::security_events::EventKind::ConfigChange,
2506 "memory_forget",
2507 &format!("scope={:?} note_id={note_id}", q.scope.unwrap_or_default()),
2508 None,
2509 );
2510 (
2511 StatusCode::OK,
2512 Json(serde_json::json!({ "forgotten": note_id, "bytes": bytes })),
2513 )
2514 .into_response()
2515 }
2516 Err(e) => (
2517 StatusCode::NOT_FOUND,
2518 Json(serde_json::json!({ "error": e })),
2519 )
2520 .into_response(),
2521 }
2522}
2523
2524#[derive(serde::Deserialize)]
2525struct MemoryAmendBody {
2526 text: String,
2527 #[serde(default)]
2528 scope: Option<MemoryScope>,
2529 /// See `AgentScopeQuery`.
2530 #[serde(default)]
2531 agent: Option<String>,
2532}
2533
2534#[derive(serde::Deserialize, Default)]
2535struct MemoryScopeQuery {
2536 #[serde(default)]
2537 scope: Option<MemoryScope>,
2538 /// See `AgentScopeQuery`.
2539 #[serde(default)]
2540 agent: Option<String>,
2541}
2542
2543async fn amend_memory_note(
2544 State(state): State<AppState>,
2545 Path(note_id): Path<String>,
2546 Json(body): Json<MemoryAmendBody>,
2547) -> axum::response::Response {
2548 use axum::response::IntoResponse;
2549 if body.text.trim().is_empty() {
2550 return (
2551 StatusCode::BAD_REQUEST,
2552 Json(serde_json::json!({ "error": "note must not be empty" })),
2553 )
2554 .into_response();
2555 }
2556 let core = scoped_core!(&state, None, body.agent.as_deref());
2557 let path = memory_store_path(&core, body.scope.unwrap_or_default());
2558 let result = vak_core::memory::amend_note(&path, &note_id, &body.text);
2559 // See the identical fallback in `forget_memory_note`.
2560 let legacy_path = memory_store_path(&state.core, body.scope.unwrap_or_default());
2561 let result = match result {
2562 Err(_) if legacy_path != path => {
2563 vak_core::memory::amend_note(&legacy_path, &note_id, &body.text)
2564 }
2565 other => other,
2566 };
2567 match result {
2568 Ok(()) => {
2569 vak_core::security_events::record(
2570 &core.sessions_home(),
2571 vak_core::security_events::EventKind::ConfigChange,
2572 "memory_amend",
2573 &format!(
2574 "scope={:?} note_id={note_id}",
2575 body.scope.unwrap_or_default()
2576 ),
2577 None,
2578 );
2579 (
2580 StatusCode::OK,
2581 Json(serde_json::json!({ "amended": note_id })),
2582 )
2583 .into_response()
2584 }
2585 Err(e) => (
2586 StatusCode::NOT_FOUND,
2587 Json(serde_json::json!({ "error": e })),
2588 )
2589 .into_response(),
2590 }
2591}
2592
2593fn proposals_payload(core: &Core) -> Vec<serde_json::Value> {
2594 let mut proposals = vak_core::learning::list_proposals(&core.sessions_home(), core.cwd());
2595 if proposals.is_empty() && core.shared_data_home() != core.sessions_home() {
2596 proposals = vak_core::learning::list_proposals(&core.shared_data_home(), core.cwd());
2597 }
2598 proposals
2599 .iter()
2600 .map(|p| {
2601 serde_json::json!({
2602 "id": p.id,
2603 "name": p.name,
2604 "description": p.description,
2605 })
2606 })
2607 .collect()
2608}
2609
2610async fn list_proposals_route(
2611 State(state): State<AppState>,
2612 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2613) -> axum::response::Response {
2614 use axum::response::IntoResponse;
2615 let core = scoped_core!(&state, None, q.agent.as_deref());
2616 Json(serde_json::json!({ "proposals": proposals_payload(&core) })).into_response()
2617}
2618
2619async fn promote_proposal(
2620 State(state): State<AppState>,
2621 Path(id): Path<String>,
2622 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2623) -> axum::response::Response {
2624 use axum::response::IntoResponse;
2625 let core = scoped_core!(&state, None, q.agent.as_deref());
2626 let res = vak_core::learning::promote(&core.sessions_home(), core.cwd(), &id);
2627 let res = match res {
2628 Err(_) if state.core.shared_data_home() != core.sessions_home() => {
2629 vak_core::learning::promote(&state.core.shared_data_home(), core.cwd(), &id)
2630 }
2631 other => other,
2632 };
2633 match res {
2634 Ok(name) => (
2635 StatusCode::OK,
2636 Json(serde_json::json!({ "promoted": name })),
2637 )
2638 .into_response(),
2639 Err(e) => (
2640 StatusCode::NOT_FOUND,
2641 Json(serde_json::json!({ "error": e })),
2642 )
2643 .into_response(),
2644 }
2645}
2646
2647async fn reject_proposal(
2648 State(state): State<AppState>,
2649 Path(id): Path<String>,
2650 axum::extract::Query(q): axum::extract::Query<AgentScopeQuery>,
2651) -> axum::response::Response {
2652 use axum::response::IntoResponse;
2653 let core = scoped_core!(&state, None, q.agent.as_deref());
2654 let res = vak_core::learning::reject(&core.sessions_home(), core.cwd(), &id);
2655 let res = match res {
2656 Err(_) if state.core.shared_data_home() != core.sessions_home() => {
2657 vak_core::learning::reject(&state.core.shared_data_home(), core.cwd(), &id)
2658 }
2659 other => other,
2660 };
2661 match res {
2662 Ok(()) => (StatusCode::OK, Json(serde_json::json!({ "rejected": id }))).into_response(),
2663 Err(e) => (
2664 StatusCode::NOT_FOUND,
2665 Json(serde_json::json!({ "error": e })),
2666 )
2667 .into_response(),
2668 }
2669}
2670
2671#[derive(serde::Deserialize)]
2672struct SearchQuery {
2673 q: String,
2674 #[serde(default)]
2675 limit: Option<usize>,
2676 /// Session id whose (already-in-context) content should be skipped.
2677 #[serde(default)]
2678 exclude: Option<String>,
2679 /// Cross-project recall: search every project's ledgers under the
2680 /// sessions home (docs/design/29-personal-os.md P1), not just this cwd.
2681 #[serde(default)]
2682 all: bool,
2683 /// The Agent whose ledgers and memory are searched (default `vak`).
2684 /// Each Agent's memory is private (AGENTS.md invariant 37), so a search
2685 /// resolves one Agent the way `/memory` does and reads only its home.
2686 #[serde(default)]
2687 agent: Option<String>,
2688}
2689
2690async fn search_sessions(
2691 State(state): State<AppState>,
2692 axum::extract::Query(q): axum::extract::Query<SearchQuery>,
2693) -> axum::response::Response {
2694 use axum::response::IntoResponse;
2695 let core = scoped_core!(&state, None, q.agent.as_deref());
2696 let home = core.sessions_home();
2697 let cwd = core.cwd().clone();
2698 let query = q.q.clone();
2699 let limit = q.limit.unwrap_or(vak_session::DEFAULT_LIMIT);
2700 let excluded =
2701 vak_core::trash::search_exclusions(&core.shared_data_home(), q.exclude.as_deref());
2702 let all = q.all;
2703 let mut extras = Vec::new();
2704 let mut workspace_notes = vak_core::memory::list_notes(&home, &cwd);
2705 if all {
2706 let root = home.join("memory");
2707 if let Ok(entries) = std::fs::read_dir(&root) {
2708 workspace_notes.clear();
2709 for entry in entries.flatten() {
2710 let path = entry.path().join("MEMORY.md");
2711 if let Ok(raw) = std::fs::read_to_string(path) {
2712 workspace_notes.extend(vak_core::memory::parse_blocks(&raw));
2713 }
2714 }
2715 }
2716 }
2717 for note in workspace_notes {
2718 let id = if note.tag.is_empty() {
2719 note.id.clone()
2720 } else {
2721 note.tag.clone()
2722 };
2723 extras.push(vak_session::ExternalDoc {
2724 id,
2725 text: format!("[{}] {}", note.kind, note.text),
2726 ts: Some(note.ts),
2727 role: Some("memory".into()),
2728 });
2729 }
2730 for note in vak_core::memory::list_profile_notes(&home) {
2731 let id = format!(
2732 "profile/{}",
2733 if note.tag.is_empty() {
2734 note.id.clone()
2735 } else {
2736 note.tag.clone()
2737 }
2738 );
2739 extras.push(vak_session::ExternalDoc {
2740 id,
2741 text: format!("[{}] {}", note.kind, note.text),
2742 ts: Some(note.ts),
2743 role: Some("profile".into()),
2744 });
2745 }
2746 match tokio::task::spawn_blocking(move || {
2747 // Both hit shapes are Serialize; the workspace path keeps its flat
2748 // SessionHit wire shape, cross-project adds the project_hash wrapper.
2749 let searched = if all {
2750 vak_session::search_all_extended(&home, &query, limit, &excluded, &extras)
2751 .map(|hits| serde_json::to_value(&hits).map_err(|e| e.to_string()))
2752 } else {
2753 vak_session::search_extended(&home, &cwd, &query, limit, &excluded, &extras)
2754 .map(|hits| serde_json::to_value(&hits).map_err(|e| e.to_string()))
2755 };
2756 match searched {
2757 Ok(inner) => inner,
2758 Err(e) => Err(e.to_string()),
2759 }
2760 })
2761 .await
2762 {
2763 Ok(Ok(hits)) => Json(serde_json::json!({ "all": all, "hits": hits })).into_response(),
2764 Ok(Err(e)) => (
2765 StatusCode::INTERNAL_SERVER_ERROR,
2766 Json(serde_json::json!({ "error": e })),
2767 )
2768 .into_response(),
2769 Err(e) => (
2770 StatusCode::INTERNAL_SERVER_ERROR,
2771 Json(serde_json::json!({ "error": e.to_string() })),
2772 )
2773 .into_response(),
2774 }
2775}
2776
2777/// The bearer token lives for the life of the process; embedders (desktop
2778/// shell, tests) need it to hand to their webview, so build the secured
2779/// stack here instead of inside `serve()`.
2780pub fn secured_router(core: Core) -> (Router, String) {
2781 secured_router_with(core, false)
2782}
2783
2784/// Same stack with a CLI-level gateway override (`serve --gateway`).
2785///
2786/// Token selection: when `VAK_GATEWAY_TOKEN` is set in the
2787/// environment, it is used verbatim so service-managed bridges and other
2788/// long-lived clients can survive process restarts. Otherwise a fresh
2789/// per-process token is minted as before. The variable is never logged.
2790pub fn secured_router_with(core: Core, force_gateway: bool) -> (Router, String) {
2791 secured_router_with_port(core, force_gateway, vak_ops::OpsConfig::detect().port)
2792}
2793
2794/// Same secured stack with the actual listener port carried into operational
2795/// probes. `serve_with` uses this so a non-default `--port` cannot make the
2796/// console probe a different process.
2797pub fn secured_router_with_port(core: Core, force_gateway: bool, port: u16) -> (Router, String) {
2798 // Tauri can use either its custom scheme or the loopback-style origin,
2799 // depending on the platform and WebView runtime, plus vite dev servers.
2800 let origins = [
2801 "tauri://localhost",
2802 "http://tauri.localhost",
2803 "https://tauri.localhost",
2804 "http://localhost:1420",
2805 "http://127.0.0.1:1420",
2806 "http://localhost:5173",
2807 "http://127.0.0.1:5173",
2808 ]
2809 .into_iter()
2810 .filter_map(|o| o.parse::<axum::http::HeaderValue>().ok())
2811 .collect::<Vec<_>>();
2812 let cors = tower_http::cors::CorsLayer::new()
2813 .allow_origin(origins)
2814 // Must cover every method the router exposes: PATCH (/config,
2815 // /sessions/:id/config) and DELETE are preflighted, so omitting them
2816 // makes the browser reject the request before it is ever sent.
2817 .allow_methods([
2818 axum::http::Method::GET,
2819 axum::http::Method::POST,
2820 axum::http::Method::PUT,
2821 axum::http::Method::PATCH,
2822 axum::http::Method::DELETE,
2823 ])
2824 .allow_headers([
2825 axum::http::header::AUTHORIZATION,
2826 axum::http::header::CONTENT_TYPE,
2827 ]);
2828 let mut state = AppState::new(core);
2829 state.ops_port = port;
2830 if force_gateway {
2831 state.enable_gateway();
2832 }
2833 let token = (*state.auth_token).clone();
2834 let rl_settings = state.core.config().gateway.rate_limit.clone();
2835 let rl_config = rate_limit::RateLimitConfig::from_settings(rl_settings);
2836 let limiter = rate_limit::RateLimiter::new(rl_config, state.core.sessions_home());
2837 let app = router_with_state(state.clone())
2838 .layer(axum::middleware::from_fn_with_state(
2839 limiter,
2840 rate_limit::rate_limit_layer,
2841 ))
2842 .layer(axum::middleware::from_fn_with_state(
2843 state.clone(),
2844 enforce_participant_audience,
2845 ))
2846 .layer(axum::middleware::from_fn_with_state(
2847 AuthPolicy {
2848 token: token.clone(),
2849 home: state.core.sessions_home(),
2850 trusted_hosts: state.core.config().server.trusted_hosts.clone(),
2851 },
2852 require_bearer,
2853 ))
2854 .layer(cors);
2855 #[cfg(unix)]
2856 {
2857 let broker = state.core.agent_network_broker();
2858 let policy_file = state
2859 .core
2860 .sessions_home()
2861 .join("agent-network/policies.json");
2862 if let Err(error) = broker.load_policies(&policy_file) {
2863 eprintln!("[agent-network] policy load failed: {error}");
2864 }
2865 let socket =
2866 vak_core::agent_network::AgentNetworkBroker::socket_path(&state.core.sessions_home());
2867 tokio::spawn(async move {
2868 if let Err(error) = broker.serve_unix(&socket).await {
2869 eprintln!("[agent-network] broker stopped: {error}");
2870 }
2871 });
2872 }
2873 // Local routines: fires due scheduled tasks while this server lives.
2874 start_scheduler(&state);
2875 delivery::start_replay(&state.core);
2876 // Capability discovery is NOT started here.
2877 //
2878 // It used to be, and the CLI did its own bounded wait, and the desktop
2879 // did neither — three surfaces answering "when may a prompt be frozen?"
2880 // three different ways, which is how an admitted, working MCP server
2881 // still produced a session that had never seen its catalog.
2882 // `Core::admitted_capabilities` owns that decision now, so every surface
2883 // gets the same packet whether it was reached from a terminal, this
2884 // server, or the desktop app.
2885 // Background index sync: keeps the admin console populated from the
2886 // very first boot. Idempotent; never blocks request handling.
2887 if let Some(store) = state.store.clone() {
2888 let home = state.core.sessions_home();
2889 tokio::spawn(async move {
2890 match store.rebuild(&home) {
2891 Ok(s) if s.files_scanned > 0 => eprintln!(
2892 "[store] indexed {} files / {} entries",
2893 s.files_scanned, s.entries_indexed
2894 ),
2895 Ok(_) => {}
2896 Err(e) => eprintln!("[store] startup rebuild failed: {e}"),
2897 }
2898 });
2899 }
2900 (app, token)
2901}
2902
2903fn participant_matches_conversation_audience(
2904 participant: &coworking::VerifiedPrincipal,
2905 conversation_id: &str,
2906 observed_audience: Option<&str>,
2907) -> bool {
2908 participant.conversation_id == conversation_id
2909 && observed_audience == Some(participant.audience_id.as_str())
2910}
2911
2912/// Recheck the live conversation audience after bearer authentication and
2913/// before any scoped handler runs. The route whitelist limits *where* a
2914/// participant may go; this boundary also proves the durable grant still
2915/// names the audience owned by that conversation.
2916async fn enforce_participant_audience(
2917 State(state): State<AppState>,
2918 req: axum::extract::Request,
2919 next: axum::middleware::Next,
2920) -> axum::response::Response {
2921 if let Some(AuthenticatedPrincipal::Participant(participant)) =
2922 req.extensions().get::<AuthenticatedPrincipal>()
2923 {
2924 let conversation_id = req
2925 .uri()
2926 .path()
2927 .trim_matches('/')
2928 .split('/')
2929 .nth(1)
2930 .unwrap_or_default();
2931 let audience = conversation_audience(&state, conversation_id);
2932 if !participant_matches_conversation_audience(
2933 participant,
2934 conversation_id,
2935 audience.as_deref(),
2936 ) {
2937 return StatusCode::FORBIDDEN.into_response();
2938 }
2939 }
2940 next.run(req).await
2941}
2942
2943/// Initialize the distributed event bus from config and attach it to the
2944/// global `EventHub`. Called from `serve_with` (async) so NATS connection
2945/// retries don't block request handling.
2946pub async fn init_server_bus(core: &Core) {
2947 let config = core.config().server.bus.clone();
2948 if config.nats_url.is_none() {
2949 return;
2950 }
2951 install_server_bus(core, &config).await;
2952}
2953
2954/// Builds the bus `config` describes, with its secrets from the secrets
2955/// chain, and makes it the live one.
2956async fn install_server_bus(core: &Core, config: &vak_config::BusResolved) {
2957 let sessions_home = core.sessions_home();
2958 let workspace_id = {
2959 use sha2::{Digest, Sha256};
2960 let mut hasher = Sha256::new();
2961 hasher.update(sessions_home.to_string_lossy().as_bytes());
2962 let result = hasher.finalize();
2963 // First 6 bytes → 12 hex chars, a compact workspace-scoped subject prefix.
2964 let prefix = &result[..6];
2965 let hex: String = prefix.iter().map(|b| format!("{b:02x}")).collect();
2966 format!("ws_{}", hex)
2967 };
2968 let bus = crate::bus::ServerBus::from_resolved(
2969 &workspace_id,
2970 config,
2971 &crate::bus::BusCredentials::of(core),
2972 )
2973 .await;
2974 if let Some(mut hub) = crate::events::global() {
2975 hub.set_server_bus(std::sync::Arc::new(bus));
2976 }
2977}
2978
2979pub async fn serve(core: Core, addr: std::net::SocketAddr) -> std::io::Result<()> {
2980 serve_with(core, addr, false).await
2981}
2982
2983/// `force_gateway` mirrors `serve --gateway`: enable routing regardless of
2984/// the (untrusted-stripped) project config.
2985/// Serve an already-bound listener with an already-built router.
2986///
2987/// The pieces `secured_router` returns, joined. `vak setup` binds its own
2988/// ephemeral loopback port so it can print the URL *before* serving, and
2989/// needs the token from the same call — which `serve_with` cannot give it,
2990/// because that mints and consumes the token internally. Exposed here so
2991/// axum stays a dependency of this crate rather than leaking into the CLI.
2992pub async fn serve_router(listener: tokio::net::TcpListener, app: Router) -> std::io::Result<()> {
2993 axum::serve(listener, app).await
2994}
2995
2996pub async fn serve_with(
2997 core: Core,
2998 addr: std::net::SocketAddr,
2999 force_gateway: bool,
3000) -> std::io::Result<()> {

Lib

Indexing the workspace…

Lib