2154. 2154 .unwrap_or_else(std::sync::PoisonError::into_inner)
2155. 2155 .remove(&id);
2156. 2156 eprintln!("[gateway] approval {short} timed out; denied");
2157. 2157 false
2158. 2158 }
2159. 2159 }
2160. 2160 }
2161. 2161}
2162. 2162
2163. 2163/// "yes"/"no" vocabulary for chat replies, optionally addressed to one
2164. 2164/// gate: "yes ab12cd34". Deliberately small and strict — casual chatter
2165. 2165/// from the approver chat must not resolve gates. Returns the verdict and
2166. 2166/// the gate-id prefix when one was supplied.
2167. 2167fn parse_verdict(text: &str) -> Option<(bool, Option<String>)> {
2168. 2168 let mut tokens = text.split_whitespace();
2169. 2169 let head = tokens.next()?.to_lowercase();
2170. 2170 let verdict = match head.as_str() {
2171. 2171 "y" | "yes" | "approve" | "approved" | "ok" | "allow" => true,
2172. 2172 "n" | "no" | "deny" | "denied" | "block" => false,
2173. 2173 _ => return None,
2174. 2174 };
2175. 2175 // Extra prose after a bare verdict is ignored; exactly one short token
2176. 2176 // is treated as a gate id.
2177. 2177 let id = match tokens.next() {
2178. 2178 Some(t)
2179. 2179 if tokens.next().is_none()
2180. 2180 && t.len() >= 4
2181. 2181 && t.chars().all(|c| c.is_ascii_alphanumeric()) =>
2182. 2182 {
2183. 2183 Some(t.to_lowercase())
2184. 2184 }
2185. 2185 _ => None,
2186. 2186 };
2187. 2187 Some((verdict, id))
2188. 2188}
2189. 2189
2190. 2190async fn gateway_inbound(
2191. 2191 State(state): State<AppState>,
2192. 2192 Json(body): Json<InboundBody>,
2193. 2193) -> axum::response::Response {
2194. 2194 crate::refresh_control_plane(&state);
2195. 2195 if !state.gateway.enabled {
2196. 2196 return (
2197. 2197 StatusCode::CONFLICT,
2198. 2198 Json(serde_json::json!({
2199. 2199 "error": "gateway disabled: set [gateway] enabled = true (trusted config) or pass serve --gateway"
2200. 2200 })),
2201. 2201 )
2202. 2202 .into_response();
2203. 2203 }
2204. 2204 let text = body.text.trim().to_string();
2205. 2205 let has_attachment = body.attachments.iter().any(|a| !a.data.trim().is_empty());
2206. 2206 if body.surface.trim().is_empty()
2207. 2207 || body.chat.trim().is_empty()
2208. 2208 || (text.is_empty() && !has_attachment)
2209. 2209 {
2210. 2210 return (
2211. 2211 StatusCode::BAD_REQUEST,
2212. 2212 Json(serde_json::json!({"error": "surface, chat and text are required"})),
2213. 2213 )
2214. 2214 .into_response();
2215. 2215 }
2216. 2216 // Multi-bot-per-channel: a chat's key is scoped to the bot that
2217. 2217 // delivered the message whenever the bridge knows its own bot id
2218. 2218 // (`--bot-id`), so the same physical chat served by several bots gets
2219. 2219 // one independent allowlist entry, session, and policy per bot instead
2220. 2220 // of all of them colliding onto one shared conversation. Legacy/
2221. 2221 // single-bot bridges (no bot id) keep the original two-part key
2222. 2222 // unchanged. See `legacy_key_for` for the one-time migration this
2223. 2223 // implies for an already-approved chat or a `chat_allowlist` row.
2224. 2224 let key = match body
2225. 2225 .bot_id
2226. 2226 .as_deref()
2227. 2227 .map(str::trim)
2228. 2228 .filter(|b| !b.is_empty())
2229. 2229 {
2230. 2230 Some(bot_id) => format!("{}:{}:{bot_id}", body.surface.trim(), body.chat.trim()),
2231. 2231 None => format!("{}:{}", body.surface.trim(), body.chat.trim()),
2232. 2232 };
2233. 2233 // 0c-01/0c-02/docs/design/34: chat allowlist — reject messages from
2234. 2234 // unknown chats, but record a reviewable *pending* entry instead of a
2235. 2235 // flat rejection so the operator has a forward path to "let it
2236. 2236 // through" that isn't a hand-edited config file + process restart.
2237. 2237 // `chat_allowlist_open = true` still bypasses the store entirely.
2238. 2238 if !state.gateway.chat_allowlist_open() {
2239. 2239 let decision = state.gateway.allowlist_resolve_inbound(
2240. 2240 &state.core,
2241. 2241 &key,
2242. 2242 &text,
2243. 2243 body.bot_id.as_deref(),
2244. 2244 );
2245. 2245 match decision {
2246. 2246 AllowlistDecision::Allowed => {}
2247. 2247 AllowlistDecision::Denied => {
2248. 2248 vak_core::security_events::record(
2249. 2249 &state.core.sessions_home(),
2250. 2250 vak_core::security_events::EventKind::ChatDenied,
2251. 2251 "chat_denied",
2252. 2252 &format!("key={key}"),
2253. 2253 None,
2254. 2254 );
2255. 2255 return (
2256. 2256 StatusCode::FORBIDDEN,
2257. 2257 Json(serde_json::json!({
2258. 2258 "error": format!("chat '{key}' rejected: denied by operator"),
2259. 2259 "state": "denied",
2260. 2260 })),
2261. 2261 )
2262. 2262 .into_response();
2263. 2263 }
2264. 2264 AllowlistDecision::NewlyPending => {
2265. 2265 vak_core::security_events::record(
2266. 2266 &state.core.sessions_home(),
2267. 2267 vak_core::security_events::EventKind::ChatPending,
2268. 2268 "chat_pending",
2269. 2269 &format!("key={key}"),
2270. 2270 None,
2271. 2271 );
2272. 2272 return (
2273. 2273 StatusCode::FORBIDDEN,
2274. 2274 Json(serde_json::json!({
2275. 2275 "error": format!(
2276. 2276 "chat '{key}' rejected: awaiting operator approval in the admin console"
2277. 2277 ),
2278. 2278 "state": "pending",
2279. 2279 })),
2280. 2280 )
2281. 2281 .into_response();
2282. 2282 }
2283. 2283 AllowlistDecision::StillPending => {
2284. 2284 // A lighter, non-security-event log line: this is expected
2285. 2285 // repeat traffic from an already-reviewable key, not a
2286. 2286 // fresh incident to append to the audit trail each time.
2287. 2287 eprintln!("[gateway] chat '{key}' still pending operator review");
2288. 2288 return (
2289. 2289 StatusCode::FORBIDDEN,
2290. 2290 Json(serde_json::json!({
2291. 2291 "error": format!(
2292. 2292 "chat '{key}' rejected: still awaiting operator approval"
2293. 2293 ),
2294. 2294 "state": "pending",
2295. 2295 })),
2296. 2296 )
2297. 2297 .into_response();
2298. 2298 }
2299. 2299 }
2300. 2300 }
2301. 2301
2302. 2302 // Approval replies from the designated approver surface resolve the
2303. 2303 // addressed gate (or the oldest one) instead of becoming conversation
2304. 2304 // input. Any non-verdict text from that chat falls through to normal
2305. 2305 // routing.
2306. 2306 if state.gateway.forward_mode()
2307. 2307 && state.gateway.approver_target().as_deref() == Some(key.as_str())
2308. 2308 && let Some((verdict, gate_id)) = parse_verdict(&text)
2309. 2309 {
2310. 2310 return match state.gateway.resolve_gate(verdict, gate_id.as_deref()) {
2311. 2311 Ok(resolved) => {
2312. 2312 // A chat "no" is observable here and nowhere else, so the
2313. 2313 // durable record of the denial is written at the same beat.
2314. 2314 if !verdict {
2315. 2315 let short = resolved.id.get(..8).unwrap_or(resolved.id.as_str());
2316. 2316 let _ = vak_core::inbox::record(
2317. 2317 &state.core.shared_data_home(),
2318. 2318 vak_core::inbox::Kind::ApprovalDenied,
2319. 2319 &format!("approval denied [{short}]"),
2320. 2320 &format!(
2321. 2321 "session {} denied forwarded gate {} ({} pending)",
2322. 2322 resolved.session_id, resolved.id, resolved.remaining
2323. 2323 ),
2324. 2324 Some(&resolved.session_id),
2325. 2325 None,
2326. 2326 );
2327. 2327 }
2328. 2328 (
2329. 2329 StatusCode::OK,
2330. 2330 Json(serde_json::json!({
2331. 2331 "state": "approval_resolved",
2332. 2332 "approved": verdict,
2333. 2333 "gate": resolved.id,
2334. 2334 "session_id": resolved.session_id,
2335. 2335 "remaining": resolved.remaining,
2336. 2336 })),
2337. 2337 )
2338. 2338 .into_response()
2339. 2339 }
2340. 2340 Err(()) => (
2341. 2341 StatusCode::OK,
2342. 2342 Json(serde_json::json!({ "state": "no_pending_approvals" })),
2343. 2343 )
2344. 2344 .into_response(),
2345. 2345 };
2346. 2346 }
2347. 2347
2348. 2348 // docs/design/34 Phase 2: run this key's entry through its own
2349. 2349 // workspace's Core (sandbox, permission mode, session ledger) — not
2350. 2350 // just its provider/model — when the entry names a workspace other
2351. 2351 // than the gateway's own. Falls back to the gateway's default Core
2352. 2352 // when the entry has no workspace override, exactly as before.
2353. 2353 let core = match state.gateway.core_for_entry(&state.core, &key) {
2354. 2354 Ok(core) => core,
2355. 2355 Err(e) => {
2356. 2356 return (
2357. 2357 StatusCode::INTERNAL_SERVER_ERROR,
2358. 2358 Json(serde_json::json!({"error": format!("workspace core unavailable: {e}")})),
2359. 2359 )
2360. 2360 .into_response();
2361. 2361 }
2362. 2362 };
2363. 2363
2364. 2364 // Admission owns the conversation context. Stamp it before creating or
2365. 2365 // reopening the bound session so the ledger, prompt contract, and every
2366. 2366 // later delivery can identify the authorized audience and originating
2367. 2367 // transport without reverse-engineering mutable gateway state.
2368. 2368 let conversation_context = vak_session::ConversationContext {
2369. 2369 conversation_id: key.clone(),
2370. 2370 audience_id: key.clone(),
2371. 2371 origin: Some(vak_session::ConversationOrigin {
2372. 2372 surface: body.surface.trim().to_string(),
2373. 2373 address: body.chat.trim().to_string(),
2374. 2374 bot_id: body.bot_id.clone(),
2375. 2375 }),
2376. 2376 };
2377. 2377 let core = core.with_conversation_context(Some(conversation_context));
2378. 2378
2379. 2379 let handle = match resolve_session(&state, &core, &key).await {
2380. 2380 Ok(h) => h,
2381. 2381 Err(e) => {
2382. 2382 return (
2383. 2383 StatusCode::INTERNAL_SERVER_ERROR,
2384. 2384 Json(serde_json::json!({"error": e})),
2385. 2385 )
2386. 2386 .into_response();
2387. 2387 }
2388. 2388 };
2389. 2389
2390. 2390 let request_id = body
2391. 2391 .request_id
2392. 2392 .as_deref()
2393. 2393 .map(str::trim)
2394. 2394 .filter(|value| !value.is_empty())
2395. 2395 .map(ToOwned::to_owned)
2396. 2396 .unwrap_or_else(|| format!("gateway-{}", uuid::Uuid::now_v7()));
2397. 2397 let already_admitted = handle
2398. 2398 .admissions
2399. 2399 .lock()
2400. 2400 .unwrap_or_else(std::sync::PoisonError::into_inner)
2401. 2401 .contains(&request_id)
2402. 2402 || handle
2403. 2403 .session
2404. 2404 .lock()
2405. 2405 .ok()
2406. 2406 .and_then(|guard| {
2407. 2407 guard
2408. 2408 .as_ref()
2409. 2409 .map(|log| log.has_request_admission(&request_id))
2410. 2410 })
2411. 2411 .unwrap_or(false);
2412. 2412 if already_admitted {
2413. 2413 return (
2414. 2414 StatusCode::ACCEPTED,
2415. 2415 Json(serde_json::json!({
2416. 2416 "request_id": request_id,
2417. 2417 "state": "already_admitted",
2418. 2418 "decision": "duplicate",
2419. 2419 "session_id": binding_session(&state, &key),
2420. 2420 })),
2421. 2421 )
2422. 2422 .into_response();
2423. 2423 }
2424. 2424 handle
2425. 2425 .admissions
2426. 2426 .lock()
2427. 2427 .unwrap_or_else(std::sync::PoisonError::into_inner)
2428. 2428 .insert(request_id.clone());
2429. 2429 // Voice notes are transcribed only now — after allowlist admission and
2430. 2430 // request de-duplication, so an unknown chat or a retried request never
2431. 2431 // spends a provider call — through this chat's bot → chat voice tiers.
2432. 2432 // Spoken words are steering text, never control: commands are parsed from
2433. 2433 // what was typed.
2434. 2434 let mut voice_notes = Vec::new();
2435. 2435 for attachment in body.attachments.iter().filter(|a| a.kind == "audio") {
2436. 2436 voice_notes.push(
2437. 2437 crate::voice::transcribe_voice_note(
2438. 2438 &state,
2439. 2439 &core,
2440. 2440 &key,
2441. 2441 &attachment.mime,
2442. 2442 &attachment.data,
2443. 2443 attachment.error.as_deref(),
2444. 2444 )
2445. 2445 .await,
2446. 2446 );
2447. 2447 }
2448. 2448 let expanded_text = std::iter::once(text.clone())
2449. 2449 .filter(|typed| !typed.is_empty())
2450. 2450 .chain(voice_notes.iter().map(crate::voice::VoiceNote::prompt_line))
2451. 2451 .collect::<Vec<_>>()
2452. 2452 .join("\n");
2453. 2453 let mut admission_data = std::collections::BTreeMap::from([
2454. 2454 ("request_id".into(), request_id.clone()),
2455. 2455 ("sender".into(), body.sender.clone().unwrap_or_default()),
2456. 2456 ("origin_surface".into(), body.surface.trim().to_string()),
2457. 2457 ("origin_address".into(), body.chat.trim().to_string()),
2458. 2458 (
2459. 2459 "agent_id".into(),
2460. 2460 core.agent_identity()
2461. 2461 .map(|agent| agent.id.clone())
2462. 2462 .unwrap_or_else(|| "vak".into()),
2463. 2463 ),
2464. 2464 ]);
2465. 2465 if let Some(context) = core.conversation_context() {
2466. 2466 admission_data.insert("audience_id".into(), context.audience_id.clone());
2467. 2467 admission_data.insert("conversation_id".into(), context.conversation_id.clone());
2468. 2468 if let Some(origin) = &context.origin {
2469. 2469 admission_data.insert("bot_id".into(), origin.bot_id.clone().unwrap_or_default());
2470. 2470 }
2471. 2471 }
2472. 2472 crate::record_activity_or_buffer(
2473. 2473 &handle,
2474. 2474 vak_session::ActivityRecord {
2475. 2475 activity_id: format!("admission-{request_id}"),
2476. 2476 turn: None,
2477. 2477 kind: vak_session::ActivityKind::Run,
2478. 2478 status: vak_session::ActivityStatus::Running,
2479. 2479 label: "Gateway request accepted".into(),
2480. 2480 detail: Some(expanded_text.clone()),
2481. 2481 data: admission_data,
2482. 2482 },
2483. 2483 );
2484. 2484
2485. 2485 // Busy? Queue as logged steering input; the running loop consumes it
2486. 2486 // between model steps, and any leftovers run as a continuation turn.
2487. 2487 // The full composed message (text + images) is queued so nothing the
2488. 2488 // sender supplied is degraded to bare text.
2489. 2489 let who = body
2490. 2490 .sender
2491. 2491 .as_deref()
2492. 2492 .map(str::trim)
2493. 2493 .filter(|s| !s.is_empty())
2494. 2494 .unwrap_or("unknown");
2495. 2495 let preview: String = expanded_text.chars().take(80).collect();
2496. 2496 state.hub.emit_gateway_inbound(&body.surface, who, &preview);
2497. 2497 // Only an explicit command is control; everything else a person types
2498. 2498 // while the run is busy is steering text (docs/design/47, control
2499. 2499 // plane). "Stop using semicolons" steers; "/stop" or a bare "stop"
2500. 2500 // cancels.
2501. 2501 let command = vak_intent::parse_command(&text);
2502. 2502 let intervention = command
2503. 2503 .as_ref()
2504. 2504 .map(vak_intent::Command::intervention_kind)
2505. 2505 .unwrap_or(vak_intent::InterventionKind::Steer);
2506. 2506 let session_id = binding_session(&state, &key);
2507. 2507 if matches!(
2508. 2508 intervention,
2509. 2509 vak_intent::InterventionKind::Replan
2510. 2510 | vak_intent::InterventionKind::Reprioritize
2511. 2511 | vak_intent::InterventionKind::AddRequirement
2512. 2512 | vak_intent::InterventionKind::RemoveRequirement
2513. 2513 ) && let Some(id) = session_id.clone()
2514. 2514 {
2515. 2515 return crate::plan_change(
2516. 2516 axum::extract::State(state.clone()),
2517. 2517 axum::extract::Path(id),
2518. 2518 axum::Json(crate::PlanChangeBody {
2519. 2519 text: text.clone(),
2520. 2520 source: "human".into(),
2521. 2521 target_revision: None,
2522. 2522 }),
2523. 2523 )
2524. 2524 .await;
2525. 2525 }
2526. 2526 match intervention {
2527. 2527 vak_intent::InterventionKind::Status => {
2528. 2528 crate::record_control_activity(&handle, "Run status requested", "status");
2529. 2529 let paused = handle.steering.is_paused();
2530. 2530 let running = handle
2531. 2531 .session
2532. 2532 .lock()
2533. 2533 .unwrap_or_else(std::sync::PoisonError::into_inner)
2534. 2534 .is_none();
2535. 2535 return (
2536. 2536 StatusCode::OK,
2537. 2537 Json(serde_json::json!({
2538. 2538 "state": "status",
2539. 2539 "session_id": session_id,
2540. 2540 "running": running,
2541. 2541 "paused": paused,
2542. 2542 })),
2543. 2543 )
2544. 2544 .into_response();
2545. 2545 }
2546. 2546 vak_intent::InterventionKind::Pause => {
2547. 2547 handle.steering.pause();
2548. 2548 crate::record_control_activity(&handle, "Run paused", "pause");
2549. 2549 return (
2550. 2550 StatusCode::ACCEPTED,
2551. 2551 Json(serde_json::json!({
2552. 2552 "state": "paused",
2553. 2553 "session_id": session_id,
2554. 2554 })),
2555. 2555 )
2556. 2556 .into_response();
2557. 2557 }
2558. 2558 vak_intent::InterventionKind::Resume => {
2559. 2559 handle.steering.resume();
2560. 2560 crate::record_control_activity(&handle, "Run resumed", "resume");
2561. 2561 return (
2562. 2562 StatusCode::ACCEPTED,
2563. 2563 Json(serde_json::json!({
2564. 2564 "state": "resumed",
2565. 2565 "session_id": session_id,
2566. 2566 })),
2567. 2567 )
2568. 2568 .into_response();
2569. 2569 }
2570. 2570 vak_intent::InterventionKind::Cancel => {
2571. 2571 handle
2572. 2572 .cancel
2573. 2573 .lock()
2574. 2574 .unwrap_or_else(std::sync::PoisonError::into_inner)
2575. 2575 .cancel();
2576. 2576 crate::record_control_activity(&handle, "Run cancelled", "cancel");
2577. 2577 return (
2578. 2578 StatusCode::ACCEPTED,
2579. 2579 Json(serde_json::json!({
2580. 2580 "state": "cancelled",
2581. 2581 "session_id": session_id,
2582. 2582 })),
2583. 2583 )
2584. 2584 .into_response();
2585. 2585 }
2586. 2586 _ => {}
2587. 2587 }
2588. 2588 let busy = handle
2589. 2589 .session
2590. 2590 .lock()
2591. 2591 .unwrap_or_else(std::sync::PoisonError::into_inner)
2592. 2592 .is_none();
2593. 2593 if busy {
2594. 2594 let attributed = match body.sender.as_deref().map(str::trim) {
2595. 2595 Some(who) if !who.is_empty() => format!("[from {who}] {expanded_text}"),
2596. 2596 _ => expanded_text.clone(),
2597. 2597 };
2598. 2598 handle.steering.push_steering_message(compose_prompt(
2599. 2599 &attributed,
2600. 2600 &body.attachments,
2601. 2601 core.cwd(),
2602. 2602 ));
2603. 2603 return (
2604. 2604 StatusCode::ACCEPTED,
2605. 2605 Json(serde_json::json!({
2606. 2606 "state": "steering_queued",
2607. 2607 "session_id": session_id.unwrap_or_default(),
2608. 2608 })),
2609. 2609 )
2610. 2610 .into_response();
2611. 2611 }
2612. 2612
2613. 2613 if core.provider().is_err() {
2614. 2614 return (
2615. 2615 StatusCode::SERVICE_UNAVAILABLE,
2616. 2616 Json(serde_json::json!({"error": "no provider credential configured"})),
2617. 2617 )
2618. 2618 .into_response();
2619. 2619 }
2620. 2620
2621. 2621 let (reply_tx, reply_rx) = oneshot::channel::<ChatReply>();
2622. 2622 let want_reply = body.wait;
2623. 2623 // 0c-02: attribute the sender identity to the prompt text.
2624. 2624 let attributed = match body.sender.as_deref().map(str::trim) {
2625. 2625 Some(who) if !who.is_empty() => format!("[from {who}] {expanded_text}"),
2626. 2626 _ => expanded_text,
2627. 2627 };
2628. 2628 let prompt = compose_prompt(&attributed, &body.attachments, core.cwd());
2629. 2629 // Each heard voice note is a durable transcript activity on the
2630. 2630 // append-only ledger, the same evidence the web voice socket records.
2631. 2631 for note in &voice_notes {
2632. 2632 if let crate::voice::VoiceNote::Heard(transcript) = note
2633. 2633 && let Ok(mut session) = handle.session.lock()
2634. 2634 && let Some(session) = session.as_mut()
2635. 2635 {
2636. 2636 let _ = session.append_voice_transcript(
2637. 2637 format!("voice:{request_id}"),
2638. 2638 transcript.clone(),
2639. 2639 true,
2640. 2640 );
2641. 2641 }
2642. 2642 }
2643. 2643 // Bind this turn's `tasks` tool default (`Core::with_default_deliver_to`)
2644. 2644 // to the exact chat/bot destination it is running in. Chat surfaces use
2645. 2645 // the three-part target whenever a bot identity is known; this prevents a
2646. 2646 // scheduled result from falling back to another bot's credentials.
2647. 2647 // So "remind me every morning at 8" typed (or spoken, via Gemini Live
2648. 2648 // transcription feeding the same turn) into this chat reports back into
2649. 2649 // this same chat unless the model is told to route it elsewhere.
2650. 2650 // Same clone-and-stamp carries the surface, so the system prompt tells
2651. 2651 // the model its reply is read as a chat message rather than printed in a
2652. 2652 // terminal (docs/design/07-prompt.md). The pooled core underneath is
2653. 2653 // stamped `Server`; this narrows it to the actual transport.
2654. 2654 let core_for_turn = core
2655. 2655 .clone()
2656. 2656 .with_default_deliver_to(Some(format!(
2657. 2657 "{}:{}{}",
2658. 2658 body.surface.trim(),
2659. 2659 body.chat.trim(),
2660. 2660 body.bot_id
2661. 2661 .as_deref()
2662. 2662 .filter(|id| !id.trim().is_empty())
2663. 2663 .map(|id| format!(":{id}"))
2664. 2664 .unwrap_or_default()
2665. 2665 )))
2666. 2666 .with_surface(vak_core::Surface::Chat {
2667. 2667 channel: body.surface.trim().to_string(),
2668. 2668 })
2669. 2669 // Must agree with the approver `execute_turn_chain` installs below:
2670. 2670 // `GatewayApprover` in forward mode, `AutoDeny` otherwise. Stamped
2671. 2671 // here, before the session's prompt is composed, so the prompt can
2672. 2672 // decline to advertise a capability this chat could never use.
2673. 2673 .with_approver_answerable(state.gateway.forward_mode())
2674. 2674 .with_prompt_overlays(state.gateway.resolve_prompt_overlays(&key));
2675. 2675 start_turn_chain(
2676. 2676 &state,
2677. 2677 &core_for_turn,
2678. 2678 handle,
2679. 2679 prompt,
2680. 2680 want_reply.then_some(reply_tx),
2681. 2681 );
2682. 2682
2683. 2683 if !want_reply {
2684. 2684 return (
2685. 2685 StatusCode::ACCEPTED,
2686. 2686 Json(serde_json::json!({ "state": "started", "request_id": request_id })),
2687. 2687 )
2688. 2688 .into_response();
2689. 2689 }
2690. 2690 match tokio::time::timeout(WAIT_TIMEOUT, reply_rx).await {
2691. 2691 Ok(Ok(reply)) => {
2692. 2692 let accepts_files = body
2693. 2693 .capabilities
2694. 2694 .as_ref()
2695. 2695 .and_then(|capabilities| capabilities.accepts_files)
2696. 2696 .unwrap_or(false);
2697. 2697 let (text, files) =
2698. 2698 return_drafts(&core, reply.text, &reply.drafts, accepts_files).await;
2699. 2699 // The turn's delivery posture, from its intent entry: the
2700. 2700 // session ledger when the run has handed it back, else the
2701. 2701 // handle's last known record.
2702. 2702 let intent_posture = binding_session(&state, &key)
2703. 2703 .as_deref()
2704. 2704 .and_then(|session_id| state.get(session_id))
2705. 2705 .and_then(|handle| {
2706. 2706 let from_ledger = handle.session.lock().ok().and_then(|guard| {
2707. 2707 guard.as_ref().and_then(|session| {
2708. 2708 session.chain_to_root().iter().rev().find_map(|entry| {
2709. 2709 match &entry.payload {
2710. 2710 vak_session::EntryPayload::Intent(record) => {
2711. 2711 Some(record.engagement.posture.delivery)
2712. 2712 }
2713. 2713 _ => None,
2714. 2714 }
2715. 2715 })
2716. 2716 })
2717. 2717 });
2718. 2718 from_ledger.or_else(|| {
2719. 2719 handle.intent.lock().ok().and_then(|guard| {
2720. 2720 guard.as_ref().map(|record| record.engagement.posture.delivery)
2721. 2721 })
2722. 2722 })
2723. 2723 });
2724. 2724 let outcome_metadata = binding_session(&state, &key)
2725. 2725 .as_deref()
2726. 2726 .and_then(|session_id| state.get(session_id))
2727. 2727 .and_then(|handle| {
2728. 2728 handle.presentation.lock().ok().map(|timeline| {
2729. 2729 timeline
2730. 2730 .items
2731. 2731 .iter()
2732. 2732 .rev()
2733. 2733 .find_map(|item| match &item.content {
2734. 2734 vak_delivery::OutputContent::Document { document } => {
2735. 2735 Some(document.metadata.clone())
2736. 2736 }
2737. 2737 _ => None,
2738. 2738 })
2739. 2739 })
2740. 2740 })
2741. 2741 .flatten();
2742. 2742 match crate::delivery::render_response(
2743. 2743 &core,
2744. 2744 body.surface.trim(),
2745. 2745 body.chat.trim(),
2746. 2746 text.clone(),
2747. 2747 body.capabilities.as_ref(),
2748. 2748 outcome_metadata,
2749. 2749 Some(std::collections::BTreeMap::from([
2750. 2750 ("request_id".into(), request_id.clone()),
2751. 2751 ("agent_id".into(),
2752. 2752 core.agent_identity()
2753. 2753 .map(|agent| agent.id.clone())
2754. 2754 .unwrap_or_else(|| "vak".into())),
2755. 2755 ("audience_id".into(),
2756. 2756 core.conversation_context()
2757. 2757 .map(|context| context.audience_id.clone())
2758. 2758 .unwrap_or_else(|| key.clone())),
2759. 2759 ("conversation_id".into(),
2760. 2760 core.conversation_context()
2761. 2761 .map(|context| context.conversation_id.clone())
2762. 2762 .unwrap_or_else(|| key.clone())),
2763. 2763 ("origin".into(), format!("{}:{}", body.surface.trim(), body.chat.trim())),
2764. 2764 ("bot_id".into(), body.bot_id.clone().unwrap_or_default()),
2765. 2765 ])),
2766. 2766 body.bot_id.as_deref(),
2767. 2767 session_id.as_deref(),
2768. 2768 intent_posture,
2769. 2769 )
2770. 2770 .await
2771. 2771 {
2772. 2772 Ok(delivery) => (
2773. 2773 StatusCode::OK,
2774. 2774 Json(serde_json::json!({
2775. 2775 "state": "completed",
2776. 2776 "request_id": request_id,
2777. 2777 "text": text,
2778. 2778 "session_id": session_id,
2779. 2779 "delivery": delivery,
2780. 2780 "files": files,
2781. 2781 })),
2782. 2782 )
2783. 2783 .into_response(),
2784. 2784 Err(error) => (
2785. 2785 StatusCode::OK,
2786. 2786 Json(serde_json::json!({
2787. 2787 "state": "completed",
2788. 2788 "request_id": request_id,
2789. 2789 "text": text,
2790. 2790 "session_id": session_id,
2791. 2791 "delivery_error": error,
2792. 2792 "files": files,
2793. 2793 })),
2794. 2794 )
2795. 2795 .into_response(),
2796. 2796 }
2797. 2797 }
2798. 2798 Ok(Err(_)) => (
2799. 2799 StatusCode::INTERNAL_SERVER_ERROR,
2800. 2800 Json(serde_json::json!({"error": "turn chain ended without a reply", "request_id": request_id})),
2801. 2801 )
2802. 2802 .into_response(),
2803. 2803 Err(_) => (
2804. 2804 StatusCode::GATEWAY_TIMEOUT,
2805. 2805 Json(serde_json::json!({
2806. 2806 "error": "turn did not finish in time; poll /sessions/{id}/transcript"
2807. 2807 ,"request_id": request_id
2808. 2808 })),
2809. 2809 )
2810. 2810 .into_response(),
2811. 2811 }
2812. 2812}
2813. 2813
2814. 2814async fn gateway_status(State(state): State<AppState>) -> Json<serde_json::Value> {
2815. 2815 let bindings: Vec<serde_json::Value> = state
2816. 2816 .gateway
2817. 2817 .snapshot()
2818. 2818 .into_iter()
2819. 2819 .map(|(target, binding)| {
2820. 2820 let agent_id = state
2821. 2821 .gateway
2822. 2822 .allowlist_get(&target)
2823. 2823 .and_then(|entry| entry.agent_id)
2824. 2824 .unwrap_or_else(|| "vak".into());
2825. 2825 let paused = binding
2826. 2826 .session_id
2827. 2827 .as_deref()
2828. 2828 .and_then(|id| state.get(id))
2829. 2829 .is_some_and(|handle| handle.steering.is_paused());
2830. 2830 serde_json::json!({
2831. 2831 "target": target,
2832. 2832 "agent_id": agent_id,
2833. 2833 "session_id": binding.session_id,
2834. 2834 "provider": binding.provider,
2835. 2835 "model": binding.model,
2836. 2836 "workspace": binding.workspace,
2837. 2837 "route_revision": binding.route_revision,
2838. 2838 "paused": paused,
2839. 2839 })
2840. 2840 })
2841. 2841 .collect();
2842. 2842 Json(serde_json::json!({
2843. 2843 "enabled": state.gateway.enabled,
2844. 2844 "cwd": state.core.cwd(),
2845. 2845 "bindings": bindings,
2846. 2846 "approvals": {
2847. 2847 "mode": state.gateway.approvals_mode(),
2848. 2848 "approver": state.gateway.approver_target(),
2849. 2849 "pending": state.gateway.pending_approval_count(),
2850. 2850 },
2851. 2851 }))
2852. 2852}
2853. 2853
2854. 2854async fn gateway_unbind(State(state): State<AppState>, Path(key): Path<String>) -> StatusCode {
2855. 2855 if state.gateway.unbind(&state.core, &key) {
2856. 2856 StatusCode::OK
2857. 2857 } else {
2858. 2858 StatusCode::NOT_FOUND
2859. 2859 }
2860. 2860}
2861. 2861
2862. 2862// ---- Session resolution -----------------------------------------------------
2863. 2863
2864. 2864fn binding_session(state: &AppState, key: &str) -> Option<String> {
2865. 2865 state
2866. 2866 .gateway
2867. 2867 .bindings
2868. 2868 .lock()
2869. 2869 .unwrap_or_else(std::sync::PoisonError::into_inner)
2870. 2870 .get(key)
2871. 2871 .and_then(|binding| binding.session_id.clone())
2872. 2872}
2873. 2873
2874. 2874fn binding_route(state: &AppState, core: &Core, key: &str) -> (String, String, String) {
2875. 2875 let override_route = state.gateway.effective_route_override(key);
2876. 2876 if let Some((provider, model)) = override_route {
2877. 2877 let revision = format!("channel:{}:{}", provider, model);
2878. 2878 return (provider, model, revision);
2879. 2879 }
2880. 2880 let _ = core.refresh_persisted_route();
2881. 2881 let route = core.effective_route();
2882. 2882 (route.provider, route.model, route.revision)
2883. 2883}
2884. 2884
2885. 2885fn busy_binding_matches_revision(state: &AppState, core: &Core, key: &str, revision: &str) -> bool {
2886. 2886 state
2887. 2887 .gateway
2888. 2888 .bindings
2889. 2889 .lock()
2890. 2890 .unwrap_or_else(std::sync::PoisonError::into_inner)
2891. 2891 .get(key)
2892. 2892 .is_some_and(|binding| {
2893. 2893 binding.route_revision.as_deref() == Some(revision)
2894. 2894 && binding.workspace.as_deref() == Some(core.cwd().as_path())
2895. 2895 })
2896. 2896}
2897. 2897
2898. 2898fn session_matches_route(
2899. 2899 session: &vak_session::SessionLog,
2900. 2900 core: &Core,
2901. 2901 provider: &str,
2902. 2902 model: &str,
2903. 2903) -> bool {
2904. 2904 // A channel-scoped route override (revision prefix "channel:") is an
2905. 2905 // explicit operator-level binding: the session must have been created
2906. 2906 // with the overridden provider/model to be reusable — otherwise the
2907. 2907 // channel would silently get a different identity than intended (rule 23).
2908. 2908 //
2909. 2909 // Without an explicit override the current provider/model is always the
2910. 2910 // effective_route() which every turn already re-evaluates, so there is
2911. 2911 // nothing to rotate on — the existing session is always compatible.
2912. 2912 let has_channel_override = provider.contains('/') || {
2913. 2913 // Detect the "channel:p:m" revision stamp that binding_route emits
2914. 2914 // for overrides. We check by whether the caller got the route from an
2915. 2915 // override or from effective_route(): overrides set revision to
2916. 2916 // "channel:provider:model", effective_route returns a hash/timestamp.
2917. 2917 // The simplest proxy: provider/model differ from the workspace default.
2918. 2918 let route = core.effective_route();
2919. 2919 provider != route.provider || model != route.model
2920. 2920 };
2921. 2921 session.header().is_some_and(|header| {
2922. 2922 let workspace_ok = header.cwd.as_path() == core.cwd().as_path();
2923. 2923 let header_agent_id = header
2924. 2924 .agent
2925. 2925 .as_ref()
2926. 2926 .map(|a| a.id.as_str())
2927. 2927 .unwrap_or("vak");
2928. 2928 let core_agent_id = core
2929. 2929 .agent_identity()
2930. 2930 .map(|a| a.id.as_str())
2931. 2931 .unwrap_or("vak");
2932. 2932 let agent_ok = header_agent_id == core_agent_id;
2933. 2933 let conv_ok = core
2934. 2934 .conversation_context()
2935. 2935 .is_none_or(|expected| header.conversation.as_ref() == Some(expected));
2936. 2936 let capabilities_ok = header.contract.capabilities == core.capability_descriptors();
2937. 2937 if has_channel_override {
2938. 2938 // Channel route overrides: session must match the pinned route.
2939. 2939 // Per-turn routing does not apply across explicit bot/channel splits.
2940. 2940 workspace_ok
2941. 2941 && agent_ok
2942. 2942 && conv_ok
2943. 2943 && capabilities_ok
2944. 2944 && header.contract.provider == provider
2945. 2945 && header.contract.model == model
2946. 2946 } else {
2947. 2947 // No override: per-turn routing handles provider/model, so any
2948. 2948 // session in this workspace+conversation is valid if capabilities match.
2949. 2949 workspace_ok && agent_ok && conv_ok && capabilities_ok
2950. 2950 }
2951. 2951 })
2952. 2952}
2953. 2953
2954. 2954/// Note a prompt-layer change in the security log before rotating.
2955. 2955///
2956. 2956/// Rotation is otherwise indistinguishable from a route change or a deleted
2957. 2957/// ledger, and "my bot started answering differently" is exactly the question
2958. 2958/// an operator brings to the audit trail.
2959. 2959///
2960. 2960/// Takes the already-computed drift rather than a session id: the caller
2961. 2961/// holds the ledger lock through its live handle, so reopening the session
2962. 2962/// here would fail every time and silently record nothing.
2963. 2963fn record_prompt_drift(
2964. 2964 core: &Core,
2965. 2965 key: &str,
2966. 2966 session_id: &str,
2967. 2967 drift: Option<vak_core::prompts::PromptDrift>,
2968. 2968) {
2969. 2969 let Some(drift) = drift else {
2970. 2970 return;
2971. 2971 };
2972. 2972 vak_core::security_events::record(
2973. 2973 &core.sessions_home(),
2974. 2974 vak_core::security_events::EventKind::ConfigChange,
2975. 2975 "prompt layers changed",
2976. 2976 &format!(
2977. 2977 "chat {key} rotated off session {session_id}: {}",
2978. 2978 drift.lines().join("; ")
2979. 2979 ),
2980. 2980 None,
2981. 2981 );
2982. 2982}
2983. 2983
2984. 2984/// Attach-or-create the session bound to `key`. Stale bindings (ledger
2985. 2985/// deleted through the normal endpoint) rebind to a fresh session.
2986. 2986async fn resolve_session(
2987. 2987 state: &AppState,
2988. 2988 core: &Core,
2989. 2989 key: &str,
2990. 2990) -> Result<Arc<SessionHandle>, String> {
2991. 2991 let (provider, model, revision) = binding_route(state, core, key);
2992. 2992 if let Some(sid) = binding_session(state, key) {
2993. 2993 if let Some(handle) = state.get(&sid) {
2994. 2994 let (matches, drift) = {
2995. 2995 let session = handle
2996. 2996 .session
2997. 2997 .lock()
2998. 2998 .unwrap_or_else(std::sync::PoisonError::into_inner);
2999. 2999 match session.as_ref() {
3000. 3000 Some(session) => (
3001. 3001 session_matches_route(session, core, &provider, &model),
3002. 3002 session
3003. 3003 .header()
3004. 3004 .and_then(|header| core.prompt_drift(&header.contract)),
3005. 3005 ),
3006. 3006 None => (
3007. 3007 busy_binding_matches_revision(state, core, key, &revision),
3008. 3008 None,
3009. 3009 ),
3010. 3010 }
3011. 3011 };
3012. 3012 if matches {
3013. 3013 return Ok(handle);
3014. 3014 }
3015. 3015 record_prompt_drift(core, key, &sid, drift);
3016. 3016 state.gateway.rotate(core, key);
3017. 3017 } else {
3018. 3018 match core.open_session(&sid).await {
3019. 3019 Ok(session) => {
3020. 3020 if session_matches_route(&session, core, &provider, &model) {
3021. 3021 let id = session
3022. 3022 .header()
3023. 3023 .map(|h| h.session_id.clone())
3024. 3024 .unwrap_or_else(|| sid.clone());
3025. 3025 return Ok(crate::register_handle(
3026. 3026 state,
3027. 3027 id,
3028. 3028 session,
3029. 3029 core.cwd().clone(),
3030. 3030 core.clone(),
3031. 3031 ));
3032. 3032 }
3033. 3033 record_prompt_drift(
3034. 3034 core,
3035. 3035 key,
3036. 3036 &sid,
3037. 3037 session
3038. 3038 .header()
3039. 3039 .and_then(|header| core.prompt_drift(&header.contract)),
3040. 3040 );
3041. 3041 state.gateway.rotate(core, key);
3042. 3042 }
3043. 3043 Err(_) => {
3044. 3044 state.gateway.rotate(core, key);
3045. 3045 }
3046. 3046 }
3047. 3047 }
3048. 3048 }
3049. 3049 let session = core
3050. 3050 .start_session_with_route(provider, model)
3051. 3051 .await
3052. 3052 .map_err(|e| format!("start session: {e}"))?;
3053. 3053 let id = session
3054. 3054 .header()
3055. 3055 .map(|h| h.session_id.clone())
3056. 3056 .unwrap_or_default();
3057. 3057 let handle =
3058. 3058 crate::register_handle(state, id.clone(), session, core.cwd().clone(), core.clone());
3059. 3059 // Two racing first-messages could each mint a session; last bind wins
3060. 3060 // and the loser stays a hidden header-only draft.
3061. 3061 state.gateway.bind(core, key.to_string(), id, revision);
3062. 3062 Ok(handle)
3063. 3063}
3064. 3064
3065. 3065// ---- Turn execution ---------------------------------------------------------
3066. 3066
3067. 3067/// Run a turn chain: prompt, then any steering left queued by concurrent
3068. 3068/// inbound messages, until the queue is dry. Sends one `RunFinished` per
3069. 3069/// leg so SSE consumers see normal terminal markers. The loop/lock
3070. 3070/// mechanics (run a leg, decide whether to continue) are
3071. 3071/// `crate::run_turn_chain` — the ONE executor also used by the HTTP `/run`
3072. 3072/// and `/steering` endpoints (invariant 30); only the approver construction
3073. 3073/// and the per-leg settle bookkeeping below are gateway-specific.
3074. 3074fn start_turn_chain(
3075. 3075 state: &AppState,
3076. 3076 core: &Core,
3077. 3077 handle: Arc<SessionHandle>,
3078. 3078 prompt: vak_llm::Message,
3079. 3079 reply: Option<oneshot::Sender<ChatReply>>,
3080. 3080) {
3081. 3081 let core = core.clone();
3082. 3082 let gw = state.gateway.clone();
3083. 3083 tokio::spawn(execute_turn_chain(core, gw, handle, prompt, reply));
3084. 3084}
3085. 3085
3086. 3086/// Voice and other non-HTTP surfaces use the same governed executor while
3087. 3087/// already holding the frozen session core and gateway state.
3088. 3088pub(crate) fn start_turn_chain_with_gateway(
3089. 3089 gateway: Arc<GatewayState>,
3090. 3090 core: &Core,
3091. 3091 handle: Arc<SessionHandle>,
3092. 3092 prompt: vak_llm::Message,
3093. 3093 reply: Option<oneshot::Sender<ChatReply>>,
3094. 3094) {
3095. 3095 tokio::spawn(execute_turn_chain(
3096. 3096 core.clone(),
3097. 3097 gateway,
3098. 3098 handle,
3099. 3099 prompt,
3100. 3100 reply,
3101. 3101 ));
3102. 3102}
3103. 3103
3104. 3104async fn execute_turn_chain(
3105. 3105 core: Core,
3106. 3106 gw: Arc<GatewayState>,
3107. 3107 handle: Arc<SessionHandle>,
3108. 3108 prompt: vak_llm::Message,
3109. 3109 mut reply: Option<oneshot::Sender<ChatReply>>,
3110. 3110) {
3111. 3111 let taken = handle
3112. 3112 .session
3113. 3113 .lock()
3114. 3114 .unwrap_or_else(std::sync::PoisonError::into_inner)
3115. 3115 .take();
3116. 3116 let Some(taken) = taken else {
3117. 3117 // Lost the race with another writer; hand our full prompt (text +
3118. 3118 // images) to the winner as steering instead of dropping it.
3119. 3119 handle.steering.push_steering_message(prompt);
3120. 3120 return;
3121. 3121 };
3122. 3122
3123. 3123 let approver_gw = gw.clone();
3124. 3124 let approver_core = core.clone();
3125. 3125 let approver_handle = handle.clone();
3126. 3126 let approver_factory = move |session_id: &str| -> Arc<dyn vak_agent::Approver> {
3127. 3127 // Forward mode is the human-in-the-loop gate (AGENTS rule 16). Under
3128. 3128 // FullAccess the engine returns `Allow` for bash/read/write, so no
3129. 3129 // Ask is ever raised and the approver is never invoked — the gate is
3130. 3130 // silently hollow. Warn once; keep going so legitimate setups still
3131. 3131 // run, but make the bypass unmistakable.
3132. 3132 if approver_gw.forward_mode()
3133. 3133 && matches!(
3134. 3134 approver_core.effective_permission_mode(),
3135. 3135 vak_config::PermissionMode::FullAccess
3136. 3136 )
3137. 3137 && FORWARD_FULLACCESS_WARNED
3138. 3138 .compare_exchange(
3139. 3139 false,
3140. 3140 true,
3141. 3141 std::sync::atomic::Ordering::AcqRel,
3142. 3142 std::sync::atomic::Ordering::Acquire,
3143. 3143 )
3144. 3144 .is_ok()
3145. 3145 {
3146. 3146 eprintln!(
3147. 3147 "vak gateway: WARNING approvals=\"forward\" is configured while the \
3148. 3148 effective permission_mode is FullAccess — bash/read/write resolve to \
3149. 3149 Allow, so no Ask gate is raised and the forward approver is never \
3150. 3150 invoked (the human-in-the-loop is silently bypassed for Allow-class \
3151. 3151 tools). Set [permissions] permission_mode = \"workspace-write\" on this \
3152. 3152 workspace to make forward mode effective, or drop the forward \
3153. 3153 approval configuration."