- 2281
.into_response(); - 2282
} - 2283
AllowlistDecision::StillPending => { - 2284
// A lighter, non-security-event log line: this is expected - 2285
// repeat traffic from an already-reviewable key, not a - 2286
// fresh incident to append to the audit trail each time. - 2287
eprintln!("[gateway] chat '{key}' still pending operator review"); - 2288
return ( - 2289
StatusCode::FORBIDDEN, - 2290
Json(serde_json::json!({ - 2291
"error": format!( - 2292
"chat '{key}' rejected: still awaiting operator approval" - 2293
), - 2294
"state": "pending", - 2295
})), - 2296
) - 2297
.into_response(); - 2298
} - 2299
} - 2300
} - 2301
- 2302
// Approval replies from the designated approver surface resolve the - 2303
// addressed gate (or the oldest one) instead of becoming conversation - 2304
// input. Any non-verdict text from that chat falls through to normal - 2305
// routing. - 2306
if state.gateway.forward_mode() - 2307
&& state.gateway.approver_target().as_deref() == Some(key.as_str()) - 2308
&& let Some((verdict, gate_id)) = parse_verdict(&text) - 2309
{ - 2310
return match state.gateway.resolve_gate(verdict, gate_id.as_deref()) { - 2311
Ok(resolved) => { - 2312
// A chat "no" is observable here and nowhere else, so the - 2313
// durable record of the denial is written at the same beat. - 2314
if !verdict { - 2315
let short = resolved.id.get(..8).unwrap_or(resolved.id.as_str()); - 2316
let _ = vak_core::inbox::record( - 2317
&state.core.shared_data_home(), - 2318
vak_core::inbox::Kind::ApprovalDenied, - 2319
&format!("approval denied [{short}]"), - 2320
&format!( - 2321
"session {} denied forwarded gate {} ({} pending)", - 2322
resolved.session_id, resolved.id, resolved.remaining - 2323
), - 2324
Some(&resolved.session_id), - 2325
None, - 2326
); - 2327
} - 2328
( - 2329
StatusCode::OK, - 2330
Json(serde_json::json!({ - 2331
"state": "approval_resolved", - 2332
"approved": verdict, - 2333
"gate": resolved.id, - 2334
"session_id": resolved.session_id, - 2335
"remaining": resolved.remaining, - 2336
})), - 2337
) - 2338
.into_response() - 2339
} - 2340
Err(()) => ( - 2341
StatusCode::OK, - 2342
Json(serde_json::json!({ "state": "no_pending_approvals" })), - 2343
) - 2344
.into_response(), - 2345
}; - 2346
} - 2347
- 2348
// docs/design/34 Phase 2: run this key's entry through its own - 2349
// workspace's Core (sandbox, permission mode, session ledger) — not - 2350
// just its provider/model — when the entry names a workspace other - 2351
// than the gateway's own. Falls back to the gateway's default Core - 2352
// when the entry has no workspace override, exactly as before. - 2353
let core = match state.gateway.core_for_entry(&state.core, &key) { - 2354
Ok(core) => core, - 2355
Err(e) => { - 2356
return ( - 2357
StatusCode::INTERNAL_SERVER_ERROR, - 2358
Json(serde_json::json!({"error": format!("workspace core unavailable: {e}")})), - 2359
) - 2360
.into_response(); - 2361
} - 2362
}; - 2363
- 2364
// Admission owns the conversation context. Stamp it before creating or - 2365
// reopening the bound session so the ledger, prompt contract, and every - 2366
// later delivery can identify the authorized audience and originating - 2367
// transport without reverse-engineering mutable gateway state. - 2368
let conversation_context = vak_session::ConversationContext { - 2369
conversation_id: key.clone(), - 2370
audience_id: key.clone(), - 2371
origin: Some(vak_session::ConversationOrigin { - 2372
surface: body.surface.trim().to_string(), - 2373
address: body.chat.trim().to_string(), - 2374
bot_id: body.bot_id.clone(), - 2375
}), - 2376
}; - 2377
let core = core.with_conversation_context(Some(conversation_context)); - 2378
- 2379
let handle = match resolve_session(&state, &core, &key).await { - 2380
Ok(h) => h, - 2381
Err(e) => { - 2382
return ( - 2383
StatusCode::INTERNAL_SERVER_ERROR, - 2384
Json(serde_json::json!({"error": e})), - 2385
) - 2386
.into_response(); - 2387
} - 2388
}; - 2389
- 2390
let request_id = body - 2391
.request_id - 2392
.as_deref() - 2393
.map(str::trim) - 2394
.filter(|value| !value.is_empty()) - 2395
.map(ToOwned::to_owned) - 2396
.unwrap_or_else(|| format!("gateway-{}", uuid::Uuid::now_v7())); - 2397
let already_admitted = handle - 2398
.admissions - 2399
.lock() - 2400
.unwrap_or_else(std::sync::PoisonError::into_inner) - 2401
.contains(&request_id) - 2402
|| handle - 2403
.session - 2404
.lock() - 2405
.ok() - 2406
.and_then(|guard| { - 2407
guard - 2408
.as_ref() - 2409
.map(|log| log.has_request_admission(&request_id)) - 2410
}) - 2411
.unwrap_or(false); - 2412
if already_admitted { - 2413
return ( - 2414
StatusCode::ACCEPTED, - 2415
Json(serde_json::json!({ - 2416
"request_id": request_id, - 2417
"state": "already_admitted", - 2418
"decision": "duplicate", - 2419
"session_id": binding_session(&state, &key), - 2420
})), - 2421
) - 2422
.into_response(); - 2423
} - 2424
handle - 2425
.admissions - 2426
.lock() - 2427
.unwrap_or_else(std::sync::PoisonError::into_inner) - 2428
.insert(request_id.clone()); - 2429
// Voice notes are transcribed only now — after allowlist admission and - 2430
// request de-duplication, so an unknown chat or a retried request never - 2431
// spends a provider call — through this chat's bot → chat voice tiers. - 2432
// Spoken words are steering text, never control: commands are parsed from - 2433
// what was typed. - 2434
let mut voice_notes = Vec::new(); - 2435
for attachment in body.attachments.iter().filter(|a| a.kind == "audio") { - 2436
voice_notes.push( - 2437
crate::voice::transcribe_voice_note( - 2438
&state, - 2439
&core, - 2440
&key, - 2441
&attachment.mime, - 2442
&attachment.data, - 2443
attachment.error.as_deref(), - 2444
) - 2445
.await, - 2446
); - 2447
} - 2448
let expanded_text = std::iter::once(text.clone()) - 2449
.filter(|typed| !typed.is_empty()) - 2450
.chain(voice_notes.iter().map(crate::voice::VoiceNote::prompt_line)) - 2451
.collect::<Vec<_>>() - 2452
.join("\n"); - 2453
let mut admission_data = std::collections::BTreeMap::from([ - 2454
("request_id".into(), request_id.clone()), - 2455
("sender".into(), body.sender.clone().unwrap_or_default()), - 2456
("origin_surface".into(), body.surface.trim().to_string()), - 2457
("origin_address".into(), body.chat.trim().to_string()), - 2458
( - 2459
"agent_id".into(), - 2460
core.agent_identity() - 2461
.map(|agent| agent.id.clone()) - 2462
.unwrap_or_else(|| "vak".into()), - 2463
), - 2464
]); - 2465
if let Some(context) = core.conversation_context() { - 2466
admission_data.insert("audience_id".into(), context.audience_id.clone()); - 2467
admission_data.insert("conversation_id".into(), context.conversation_id.clone()); - 2468
if let Some(origin) = &context.origin { - 2469
admission_data.insert("bot_id".into(), origin.bot_id.clone().unwrap_or_default()); - 2470
} - 2471
} - 2472
crate::record_activity_or_buffer( - 2473
&handle, - 2474
vak_session::ActivityRecord { - 2475
activity_id: format!("admission-{request_id}"), - 2476
turn: None, - 2477
kind: vak_session::ActivityKind::Run, - 2478
status: vak_session::ActivityStatus::Running, - 2479
label: "Gateway request accepted".into(), - 2480
detail: Some(expanded_text.clone()), - 2481
data: admission_data, - 2482
}, - 2483
); - 2484
- 2485
// Busy? Queue as logged steering input; the running loop consumes it - 2486
// between model steps, and any leftovers run as a continuation turn. - 2487
// The full composed message (text + images) is queued so nothing the - 2488
// sender supplied is degraded to bare text. - 2489
let who = body - 2490
.sender - 2491
.as_deref() - 2492
.map(str::trim) - 2493
.filter(|s| !s.is_empty()) - 2494
.unwrap_or("unknown"); - 2495
let preview: String = expanded_text.chars().take(80).collect(); - 2496
state.hub.emit_gateway_inbound(&body.surface, who, &preview); - 2497
// Only an explicit command is control; everything else a person types - 2498
// while the run is busy is steering text (docs/design/47, control - 2499
// plane). "Stop using semicolons" steers; "/stop" or a bare "stop" - 2500
// cancels. - 2501
let command = vak_intent::parse_command(&text); - 2502
let intervention = command - 2503
.as_ref() - 2504
.map(vak_intent::Command::intervention_kind) - 2505
.unwrap_or(vak_intent::InterventionKind::Steer); - 2506
let session_id = binding_session(&state, &key); - 2507
if matches!( - 2508
intervention, - 2509
vak_intent::InterventionKind::Replan - 2510
| vak_intent::InterventionKind::Reprioritize - 2511
| vak_intent::InterventionKind::AddRequirement - 2512
| vak_intent::InterventionKind::RemoveRequirement - 2513
) && let Some(id) = session_id.clone() - 2514
{ - 2515
return crate::plan_change( - 2516
axum::extract::State(state.clone()), - 2517
axum::extract::Path(id), - 2518
axum::Json(crate::PlanChangeBody { - 2519
text: text.clone(), - 2520
source: "human".into(), - 2521
target_revision: None, - 2522
}), - 2523
) - 2524
.await; - 2525
} - 2526
match intervention { - 2527
vak_intent::InterventionKind::Status => { - 2528
crate::record_control_activity(&handle, "Run status requested", "status"); - 2529
let paused = handle.steering.is_paused(); - 2530
let running = handle - 2531
.session - 2532
.lock() - 2533
.unwrap_or_else(std::sync::PoisonError::into_inner) - 2534
.is_none(); - 2535
return ( - 2536
StatusCode::OK, - 2537
Json(serde_json::json!({ - 2538
"state": "status", - 2539
"session_id": session_id, - 2540
"running": running, - 2541
"paused": paused, - 2542
})), - 2543
) - 2544
.into_response(); - 2545
} - 2546
vak_intent::InterventionKind::Pause => { - 2547
handle.steering.pause(); - 2548
crate::record_control_activity(&handle, "Run paused", "pause"); - 2549
return ( - 2550
StatusCode::ACCEPTED, - 2551
Json(serde_json::json!({ - 2552
"state": "paused", - 2553
"session_id": session_id, - 2554
})), - 2555
) - 2556
.into_response(); - 2557
} - 2558
vak_intent::InterventionKind::Resume => { - 2559
handle.steering.resume(); - 2560
crate::record_control_activity(&handle, "Run resumed", "resume"); - 2561
return ( - 2562
StatusCode::ACCEPTED, - 2563
Json(serde_json::json!({ - 2564
"state": "resumed", - 2565
"session_id": session_id, - 2566
})), - 2567
) - 2568
.into_response(); - 2569
} - 2570
vak_intent::InterventionKind::Cancel => { - 2571
handle - 2572
.cancel - 2573
.lock() - 2574
.unwrap_or_else(std::sync::PoisonError::into_inner) - 2575
.cancel(); - 2576
crate::record_control_activity(&handle, "Run cancelled", "cancel"); - 2577
return ( - 2578
StatusCode::ACCEPTED, - 2579
Json(serde_json::json!({ - 2580
"state": "cancelled", - 2581
"session_id": session_id, - 2582
})), - 2583
) - 2584
.into_response(); - 2585
} - 2586
_ => {} - 2587
} - 2588
let busy = handle - 2589
.session - 2590
.lock() - 2591
.unwrap_or_else(std::sync::PoisonError::into_inner) - 2592
.is_none(); - 2593
if busy { - 2594
let attributed = match body.sender.as_deref().map(str::trim) { - 2595
Some(who) if !who.is_empty() => format!("[from {who}] {expanded_text}"), - 2596
_ => expanded_text.clone(), - 2597
}; - 2598
handle.steering.push_steering_message(compose_prompt( - 2599
&attributed, - 2600
&body.attachments, - 2601
core.cwd(), - 2602
)); - 2603
return ( - 2604
StatusCode::ACCEPTED, - 2605
Json(serde_json::json!({ - 2606
"state": "steering_queued", - 2607
"session_id": session_id.unwrap_or_default(), - 2608
})), - 2609
) - 2610
.into_response(); - 2611
} - 2612
- 2613
if core.provider().is_err() { - 2614
return ( - 2615
StatusCode::SERVICE_UNAVAILABLE, - 2616
Json(serde_json::json!({"error": "no provider credential configured"})), - 2617
) - 2618
.into_response(); - 2619
} - 2620
- 2621
let (reply_tx, reply_rx) = oneshot::channel::<ChatReply>(); - 2622
let want_reply = body.wait; - 2623
// 0c-02: attribute the sender identity to the prompt text. - 2624
let attributed = match body.sender.as_deref().map(str::trim) { - 2625
Some(who) if !who.is_empty() => format!("[from {who}] {expanded_text}"), - 2626
_ => expanded_text, - 2627
}; - 2628
let prompt = compose_prompt(&attributed, &body.attachments, core.cwd()); - 2629
// Each heard voice note is a durable transcript activity on the - 2630
// append-only ledger, the same evidence the web voice socket records. - 2631
for note in &voice_notes { - 2632
if let crate::voice::VoiceNote::Heard(transcript) = note - 2633
&& let Ok(mut session) = handle.session.lock() - 2634
&& let Some(session) = session.as_mut() - 2635
{ - 2636
let _ = session.append_voice_transcript( - 2637
format!("voice:{request_id}"), - 2638
transcript.clone(), - 2639
true, - 2640
); - 2641
} - 2642
} - 2643
// Bind this turn's `tasks` tool default (`Core::with_default_deliver_to`) - 2644
// to the exact chat/bot destination it is running in. Chat surfaces use - 2645
// the three-part target whenever a bot identity is known; this prevents a - 2646
// scheduled result from falling back to another bot's credentials. - 2647
// So "remind me every morning at 8" typed (or spoken, via Gemini Live - 2648
// transcription feeding the same turn) into this chat reports back into - 2649
// this same chat unless the model is told to route it elsewhere. - 2650
// Same clone-and-stamp carries the surface, so the system prompt tells - 2651
// the model its reply is read as a chat message rather than printed in a - 2652
// terminal (docs/design/07-prompt.md). The pooled core underneath is - 2653
// stamped `Server`; this narrows it to the actual transport. - 2654
let core_for_turn = core - 2655
.clone() - 2656
.with_default_deliver_to(Some(format!( - 2657
"{}:{}{}", - 2658
body.surface.trim(), - 2659
body.chat.trim(), - 2660
body.bot_id - 2661
.as_deref() - 2662
.filter(|id| !id.trim().is_empty()) - 2663
.map(|id| format!(":{id}")) - 2664
.unwrap_or_default() - 2665
))) - 2666
.with_surface(vak_core::Surface::Chat { - 2667
channel: body.surface.trim().to_string(), - 2668
}) - 2669
// Must agree with the approver `execute_turn_chain` installs below: - 2670
// `GatewayApprover` in forward mode, `AutoDeny` otherwise. Stamped - 2671
// here, before the session's prompt is composed, so the prompt can - 2672
// decline to advertise a capability this chat could never use. - 2673
.with_approver_answerable(state.gateway.forward_mode()) - 2674
.with_prompt_overlays(state.gateway.resolve_prompt_overlays(&key)); - 2675
start_turn_chain( - 2676
&state, - 2677
&core_for_turn, - 2678
handle, - 2679
prompt, - 2680
want_reply.then_some(reply_tx), - 2681
); - 2682
- 2683
if !want_reply { - 2684
return ( - 2685
StatusCode::ACCEPTED, - 2686
Json(serde_json::json!({ "state": "started", "request_id": request_id })), - 2687
) - 2688
.into_response(); - 2689
} - 2690
match tokio::time::timeout(WAIT_TIMEOUT, reply_rx).await { - 2691
Ok(Ok(reply)) => { - 2692
let accepts_files = body - 2693
.capabilities - 2694
.as_ref() - 2695
.and_then(|capabilities| capabilities.accepts_files) - 2696
.unwrap_or(false); - 2697
let (text, files) = - 2698
return_drafts(&core, reply.text, &reply.drafts, accepts_files).await; - 2699
// The turn's delivery posture, from its intent entry: the - 2700
// session ledger when the run has handed it back, else the - 2701
// handle's last known record. - 2702
let intent_posture = binding_session(&state, &key) - 2703
.as_deref() - 2704
.and_then(|session_id| state.get(session_id)) - 2705
.and_then(|handle| { - 2706
let from_ledger = handle.session.lock().ok().and_then(|guard| { - 2707
guard.as_ref().and_then(|session| { - 2708
session.chain_to_root().iter().rev().find_map(|entry| { - 2709
match &entry.payload { - 2710
vak_session::EntryPayload::Intent(record) => { - 2711
Some(record.engagement.posture.delivery) - 2712
} - 2713
_ => None, - 2714
} - 2715
}) - 2716
}) - 2717
}); - 2718
from_ledger.or_else(|| { - 2719
handle.intent.lock().ok().and_then(|guard| { - 2720
guard.as_ref().map(|record| record.engagement.posture.delivery) - 2721
}) - 2722
}) - 2723
}); - 2724
let outcome_metadata = binding_session(&state, &key) - 2725
.as_deref() - 2726
.and_then(|session_id| state.get(session_id)) - 2727
.and_then(|handle| { - 2728
handle.presentation.lock().ok().map(|timeline| { - 2729
timeline - 2730
.items - 2731
.iter() - 2732
.rev() - 2733
.find_map(|item| match &item.content { - 2734
vak_delivery::OutputContent::Document { document } => { - 2735
Some(document.metadata.clone()) - 2736
} - 2737
_ => None, - 2738
}) - 2739
}) - 2740
}) - 2741
.flatten(); - 2742
match crate::delivery::render_response( - 2743
&core, - 2744
body.surface.trim(), - 2745
body.chat.trim(), - 2746
text.clone(), - 2747
body.capabilities.as_ref(), - 2748
outcome_metadata, - 2749
Some(std::collections::BTreeMap::from([ - 2750
("request_id".into(), request_id.clone()), - 2751
("agent_id".into(), - 2752
core.agent_identity() - 2753
.map(|agent| agent.id.clone()) - 2754
.unwrap_or_else(|| "vak".into())), - 2755
("audience_id".into(), - 2756
core.conversation_context() - 2757
.map(|context| context.audience_id.clone()) - 2758
.unwrap_or_else(|| key.clone())), - 2759
("conversation_id".into(), - 2760
core.conversation_context() - 2761
.map(|context| context.conversation_id.clone()) - 2762
.unwrap_or_else(|| key.clone())), - 2763
("origin".into(), format!("{}:{}", body.surface.trim(), body.chat.trim())), - 2764
("bot_id".into(), body.bot_id.clone().unwrap_or_default()), - 2765
])), - 2766
body.bot_id.as_deref(), - 2767
session_id.as_deref(), - 2768
intent_posture, - 2769
) - 2770
.await - 2771
{ - 2772
Ok(delivery) => ( - 2773
StatusCode::OK, - 2774
Json(serde_json::json!({ - 2775
"state": "completed", - 2776
"request_id": request_id, - 2777
"text": text, - 2778
"session_id": session_id, - 2779
"delivery": delivery, - 2780
"files": files, - 2781
})), - 2782
) - 2783
.into_response(), - 2784
Err(error) => ( - 2785
StatusCode::OK, - 2786
Json(serde_json::json!({ - 2787
"state": "completed", - 2788
"request_id": request_id, - 2789
"text": text, - 2790
"session_id": session_id, - 2791
"delivery_error": error, - 2792
"files": files, - 2793
})), - 2794
) - 2795
.into_response(), - 2796
} - 2797
} - 2798
Ok(Err(_)) => ( - 2799
StatusCode::INTERNAL_SERVER_ERROR, - 2800
Json(serde_json::json!({"error": "turn chain ended without a reply", "request_id": request_id})), - 2801
) - 2802
.into_response(), - 2803
Err(_) => ( - 2804
StatusCode::GATEWAY_TIMEOUT, - 2805
Json(serde_json::json!({ - 2806
"error": "turn did not finish in time; poll /sessions/{id}/transcript" - 2807
,"request_id": request_id - 2808
})), - 2809
) - 2810
.into_response(), - 2811
} - 2812
} - 2813
- 2814
async fn gateway_status(State(state): State<AppState>) -> Json<serde_json::Value> { - 2815
let bindings: Vec<serde_json::Value> = state - 2816
.gateway - 2817
.snapshot() - 2818
.into_iter() - 2819
.map(|(target, binding)| { - 2820
let agent_id = state - 2821
.gateway - 2822
.allowlist_get(&target) - 2823
.and_then(|entry| entry.agent_id) - 2824
.unwrap_or_else(|| "vak".into()); - 2825
let paused = binding - 2826
.session_id - 2827
.as_deref() - 2828
.and_then(|id| state.get(id)) - 2829
.is_some_and(|handle| handle.steering.is_paused()); - 2830
serde_json::json!({ - 2831
"target": target, - 2832
"agent_id": agent_id, - 2833
"session_id": binding.session_id, - 2834
"provider": binding.provider, - 2835
"model": binding.model, - 2836
"workspace": binding.workspace, - 2837
"route_revision": binding.route_revision, - 2838
"paused": paused, - 2839
}) - 2840
}) - 2841
.collect(); - 2842
Json(serde_json::json!({ - 2843
"enabled": state.gateway.enabled, - 2844
"cwd": state.core.cwd(), - 2845
"bindings": bindings, - 2846
"approvals": { - 2847
"mode": state.gateway.approvals_mode(), - 2848
"approver": state.gateway.approver_target(), - 2849
"pending": state.gateway.pending_approval_count(), - 2850
}, - 2851
})) - 2852
} - 2853
- 2854
async fn gateway_unbind(State(state): State<AppState>, Path(key): Path<String>) -> StatusCode { - 2855
if state.gateway.unbind(&state.core, &key) { - 2856
StatusCode::OK - 2857
} else { - 2858
StatusCode::NOT_FOUND - 2859
} - 2860
} - 2861
- 2862
// ---- Session resolution ----------------------------------------------------- - 2863
- 2864
fn binding_session(state: &AppState, key: &str) -> Option<String> { - 2865
state - 2866
.gateway - 2867
.bindings - 2868
.lock() - 2869
.unwrap_or_else(std::sync::PoisonError::into_inner) - 2870
.get(key) - 2871
.and_then(|binding| binding.session_id.clone()) - 2872
} - 2873
- 2874
fn binding_route(state: &AppState, core: &Core, key: &str) -> (String, String, String) { - 2875
let override_route = state.gateway.effective_route_override(key); - 2876
if let Some((provider, model)) = override_route { - 2877
let revision = format!("channel:{}:{}", provider, model); - 2878
return (provider, model, revision); - 2879
} - 2880
let _ = core.refresh_persisted_route(); - 2881
let route = core.effective_route(); - 2882
(route.provider, route.model, route.revision) - 2883
} - 2884
- 2885
fn busy_binding_matches_revision(state: &AppState, core: &Core, key: &str, revision: &str) -> bool { - 2886
state - 2887
.gateway - 2888
.bindings - 2889
.lock() - 2890
.unwrap_or_else(std::sync::PoisonError::into_inner) - 2891
.get(key) - 2892
.is_some_and(|binding| { - 2893
binding.route_revision.as_deref() == Some(revision) - 2894
&& binding.workspace.as_deref() == Some(core.cwd().as_path()) - 2895
}) - 2896
} - 2897
- 2898
fn session_matches_route( - 2899
session: &vak_session::SessionLog, - 2900
core: &Core, - 2901
provider: &str, - 2902
model: &str, - 2903
) -> bool { - 2904
// A channel-scoped route override (revision prefix "channel:") is an - 2905
// explicit operator-level binding: the session must have been created - 2906
// with the overridden provider/model to be reusable — otherwise the - 2907
// channel would silently get a different identity than intended (rule 23). - 2908
// - 2909
// Without an explicit override the current provider/model is always the - 2910
// effective_route() which every turn already re-evaluates, so there is - 2911
// nothing to rotate on — the existing session is always compatible. - 2912
let has_channel_override = provider.contains('/') || { - 2913
// Detect the "channel:p:m" revision stamp that binding_route emits - 2914
// for overrides. We check by whether the caller got the route from an - 2915
// override or from effective_route(): overrides set revision to - 2916
// "channel:provider:model", effective_route returns a hash/timestamp. - 2917
// The simplest proxy: provider/model differ from the workspace default. - 2918
let route = core.effective_route(); - 2919
provider != route.provider || model != route.model - 2920
}; - 2921
session.header().is_some_and(|header| { - 2922
let workspace_ok = header.cwd.as_path() == core.cwd().as_path(); - 2923
let header_agent_id = header - 2924
.agent - 2925
.as_ref() - 2926
.map(|a| a.id.as_str()) - 2927
.unwrap_or("vak"); - 2928
let core_agent_id = core - 2929
.agent_identity() - 2930
.map(|a| a.id.as_str()) - 2931
.unwrap_or("vak"); - 2932
let agent_ok = header_agent_id == core_agent_id; - 2933
let conv_ok = core - 2934
.conversation_context() - 2935
.is_none_or(|expected| header.conversation.as_ref() == Some(expected)); - 2936
let capabilities_ok = header.contract.capabilities == core.capability_descriptors(); - 2937
if has_channel_override { - 2938
// Channel route overrides: session must match the pinned route. - 2939
// Per-turn routing does not apply across explicit bot/channel splits. - 2940
workspace_ok - 2941
&& agent_ok - 2942
&& conv_ok - 2943
&& capabilities_ok - 2944
&& header.contract.provider == provider - 2945
&& header.contract.model == model - 2946
} else { - 2947
// No override: per-turn routing handles provider/model, so any - 2948
// session in this workspace+conversation is valid if capabilities match. - 2949
workspace_ok && agent_ok && conv_ok && capabilities_ok - 2950
} - 2951
}) - 2952
} - 2953
- 2954
/// Note a prompt-layer change in the security log before rotating. - 2955
/// - 2956
/// Rotation is otherwise indistinguishable from a route change or a deleted - 2957
/// ledger, and "my bot started answering differently" is exactly the question - 2958
/// an operator brings to the audit trail. - 2959
/// - 2960
/// Takes the already-computed drift rather than a session id: the caller - 2961
/// holds the ledger lock through its live handle, so reopening the session - 2962
/// here would fail every time and silently record nothing. - 2963
fn record_prompt_drift( - 2964
core: &Core, - 2965
key: &str, - 2966
session_id: &str, - 2967
drift: Option<vak_core::prompts::PromptDrift>, - 2968
) { - 2969
let Some(drift) = drift else { - 2970
return; - 2971
}; - 2972
vak_core::security_events::record( - 2973
&core.sessions_home(), - 2974
vak_core::security_events::EventKind::ConfigChange, - 2975
"prompt layers changed", - 2976
&format!( - 2977
"chat {key} rotated off session {session_id}: {}", - 2978
drift.lines().join("; ") - 2979
), - 2980
None, - 2981
); - 2982
} - 2983
- 2984
/// Attach-or-create the session bound to `key`. Stale bindings (ledger - 2985
/// deleted through the normal endpoint) rebind to a fresh session. - 2986
async fn resolve_session( - 2987
state: &AppState, - 2988
core: &Core, - 2989
key: &str, - 2990
) -> Result<Arc<SessionHandle>, String> { - 2991
let (provider, model, revision) = binding_route(state, core, key); - 2992
if let Some(sid) = binding_session(state, key) { - 2993
if let Some(handle) = state.get(&sid) { - 2994
let (matches, drift) = { - 2995
let session = handle - 2996
.session - 2997
.lock() - 2998
.unwrap_or_else(std::sync::PoisonError::into_inner); - 2999
match session.as_ref() { - 3000
Some(session) => ( - 3001
session_matches_route(session, core, &provider, &model), - 3002
session - 3003
.header() - 3004
.and_then(|header| core.prompt_drift(&header.contract)), - 3005
), - 3006
None => ( - 3007
busy_binding_matches_revision(state, core, key, &revision), - 3008
None, - 3009
), - 3010
} - 3011
}; - 3012
if matches { - 3013
return Ok(handle); - 3014
} - 3015
record_prompt_drift(core, key, &sid, drift); - 3016
state.gateway.rotate(core, key); - 3017
} else { - 3018
match core.open_session(&sid).await { - 3019
Ok(session) => { - 3020
if session_matches_route(&session, core, &provider, &model) { - 3021
let id = session - 3022
.header() - 3023
.map(|h| h.session_id.clone()) - 3024
.unwrap_or_else(|| sid.clone()); - 3025
return Ok(crate::register_handle( - 3026
state, - 3027
id, - 3028
session, - 3029
core.cwd().clone(), - 3030
core.clone(), - 3031
)); - 3032
} - 3033
record_prompt_drift( - 3034
core, - 3035
key, - 3036
&sid, - 3037
session - 3038
.header() - 3039
.and_then(|header| core.prompt_drift(&header.contract)), - 3040
); - 3041
state.gateway.rotate(core, key); - 3042
} - 3043
Err(_) => { - 3044
state.gateway.rotate(core, key); - 3045
} - 3046
} - 3047
} - 3048
} - 3049
let session = core - 3050
.start_session_with_route(provider, model) - 3051
.await - 3052
.map_err(|e| format!("start session: {e}"))?; - 3053
let id = session - 3054
.header() - 3055
.map(|h| h.session_id.clone()) - 3056
.unwrap_or_default(); - 3057
let handle = - 3058
crate::register_handle(state, id.clone(), session, core.cwd().clone(), core.clone()); - 3059
// Two racing first-messages could each mint a session; last bind wins - 3060
// and the loser stays a hidden header-only draft. - 3061
state.gateway.bind(core, key.to_string(), id, revision); - 3062
Ok(handle) - 3063
} - 3064
- 3065
// ---- Turn execution --------------------------------------------------------- - 3066
- 3067
/// Run a turn chain: prompt, then any steering left queued by concurrent - 3068
/// inbound messages, until the queue is dry. Sends one `RunFinished` per - 3069
/// leg so SSE consumers see normal terminal markers. The loop/lock - 3070
/// mechanics (run a leg, decide whether to continue) are - 3071
/// `crate::run_turn_chain` — the ONE executor also used by the HTTP `/run` - 3072
/// and `/steering` endpoints (invariant 30); only the approver construction - 3073
/// and the per-leg settle bookkeeping below are gateway-specific. - 3074
fn start_turn_chain( - 3075
state: &AppState, - 3076
core: &Core, - 3077
handle: Arc<SessionHandle>, - 3078
prompt: vak_llm::Message, - 3079
reply: Option<oneshot::Sender<ChatReply>>, - 3080
) { - 3081
let core = core.clone(); - 3082
let gw = state.gateway.clone(); - 3083
tokio::spawn(execute_turn_chain(core, gw, handle, prompt, reply)); - 3084
} - 3085
- 3086
/// Voice and other non-HTTP surfaces use the same governed executor while - 3087
/// already holding the frozen session core and gateway state. - 3088
pub(crate) fn start_turn_chain_with_gateway( - 3089
gateway: Arc<GatewayState>, - 3090
core: &Core, - 3091
handle: Arc<SessionHandle>, - 3092
prompt: vak_llm::Message, - 3093
reply: Option<oneshot::Sender<ChatReply>>, - 3094
) { - 3095
tokio::spawn(execute_turn_chain( - 3096
core.clone(), - 3097
gateway, - 3098
handle, - 3099
prompt, - 3100
reply, - 3101
)); - 3102
} - 3103
- 3104
async fn execute_turn_chain( - 3105
core: Core, - 3106
gw: Arc<GatewayState>, - 3107
handle: Arc<SessionHandle>, - 3108
prompt: vak_llm::Message, - 3109
mut reply: Option<oneshot::Sender<ChatReply>>, - 3110
) { - 3111
let taken = handle - 3112
.session - 3113
.lock() - 3114
.unwrap_or_else(std::sync::PoisonError::into_inner) - 3115
.take(); - 3116
let Some(taken) = taken else { - 3117
// Lost the race with another writer; hand our full prompt (text + - 3118
// images) to the winner as steering instead of dropping it. - 3119
handle.steering.push_steering_message(prompt); - 3120
return; - 3121
}; - 3122
- 3123
let approver_gw = gw.clone(); - 3124
let approver_core = core.clone(); - 3125
let approver_handle = handle.clone(); - 3126
let approver_factory = move |session_id: &str| -> Arc<dyn vak_agent::Approver> { - 3127
// Forward mode is the human-in-the-loop gate (AGENTS rule 16). Under - 3128
// FullAccess the engine returns `Allow` for bash/read/write, so no - 3129
// Ask is ever raised and the approver is never invoked — the gate is - 3130
// silently hollow. Warn once; keep going so legitimate setups still - 3131
// run, but make the bypass unmistakable. - 3132
if approver_gw.forward_mode() - 3133
&& matches!( - 3134
approver_core.effective_permission_mode(), - 3135
vak_config::PermissionMode::FullAccess - 3136
) - 3137
&& FORWARD_FULLACCESS_WARNED - 3138
.compare_exchange( - 3139
false, - 3140
true, - 3141
std::sync::atomic::Ordering::AcqRel, - 3142
std::sync::atomic::Ordering::Acquire, - 3143
) - 3144
.is_ok() - 3145
{ - 3146
eprintln!( - 3147
"vak gateway: WARNING approvals=\"forward\" is configured while the \ - 3148
effective permission_mode is FullAccess — bash/read/write resolve to \ - 3149
Allow, so no Ask gate is raised and the forward approver is never \ - 3150
invoked (the human-in-the-loop is silently bypassed for Allow-class \ - 3151
tools). Set [permissions] permission_mode = \"workspace-write\" on this \ - 3152
workspace to make forward mode effective, or drop the forward \ - 3153
approval configuration." - 3154
); - 3155
} - 3156
// Unattended policy: deny by default, forward to the approver - 3157
// surface when configured (G2). - 3158
if approver_gw.forward_mode() { - 3159
Arc::new(GatewayApprover { - 3160
events_tx: approver_handle.events_tx.clone(), - 3161
state: approver_gw.clone(), - 3162
core: approver_core.clone(), - 3163
session_id: session_id.to_string(), - 3164
}) - 3165
} else { - 3166
Arc::new(AutoDeny) - 3167
} - 3168
}; - 3169
- 3170
let settle_core = core.clone(); - 3171
let settle = move |session_id: &str, - 3172
outcome: Result< - 3173
(vak_agent::TurnOutcome, vak_session::SessionLog), - 3174
vak_core::CoreError, - 3175
>| { - 3176
let core = settle_core.clone(); - 3177
let reply_tx = reply.take(); - 3178
let session_id = session_id.to_string(); - 3179
async move { - 3180
match outcome { - 3181
Ok((o, log)) => { - 3182
let err = outcome_is_error(&o); - 3183
let text = crate::projection::text_with_run_cards(&log, outcome_text(&o)); - 3184
if let Some(tx) = reply_tx { - 3185
let _ = tx.send(ChatReply { - 3186
text: text.clone(), - 3187
drafts: turn_drafts(&log, core.cwd()), - 3188
}); - 3189
} - 3190
// Background reflection seam (docs/design/29 P1): the - 3191
// shared best-effort pass over the just-settled leg. It - 3192
// runs strictly after the reply above was handed over so - 3193
// delivery never waits on it, and while this chain still - 3194
// owns the ledger — a second in-process handle cannot - 3195
// take the file lock. Bounded; failures collapse into - 3196
// the outcome envelope. - 3197
if !err && core.config().memory.reflection { - 3198
let pass = tokio::time::timeout( - 3199
REFLECTION_CALL_TIMEOUT, - 3200
core.reflect_after_turn(&log, ""), - 3201
) - 3202
.await; - 3203
match pass { - 3204
Ok(outcome) => log_gateway_reflection(outcome), - 3205
Err(_) => eprintln!("[gateway] reflection skipped: timeout"), - 3206
} - 3207
} - 3208
(Some(log), short_summary(&text), err) - 3209
} - 3210
Err(e) => { - 3211
// The full error is recorded internally for operators; - 3212
// the channel reply stays a human sentence (see - 3213
// `outcome_text`) — never the raw `CoreError` display. - 3214
vak_core::security_events::record( - 3215
&core.sessions_home(), - 3216
vak_core::security_events::EventKind::ExecutionError, - 3217
"inbound turn failed", - 3218
&format!("session_id={session_id} error={e}"), - 3219
None, - 3220
); - 3221
let recovered = recover_ledger(&core, &session_id).await; - 3222
let text = crate::client_events::run_outcome_message( - 3223
crate::client_events::RunOutcome::Failed, - 3224
) - 3225
.to_string(); - 3226
if let Some(tx) = reply_tx { - 3227
let _ = tx.send(ChatReply { - 3228
text: text.clone(), - 3229
drafts: Vec::new(), - 3230
}); - 3231
} - 3232
(recovered, short_summary(&text), true) - 3233
} - 3234
} - 3235
} - 3236
}; - 3237
- 3238
crate::run_turn_chain( - 3239
core, - 3240
handle, - 3241
taken, - 3242
crate::TurnStart::message(prompt), - 3243
approver_factory, - 3244
settle, - 3245
) - 3246
.await; - 3247
} - 3248
- 3249
/// A `CoreError` loses the ledger handle (the agent consumed it); reopen the - 3250
/// JSONL so the session stays usable in this long-lived process. `None` - 3251
/// when `session_id` is empty or the reopen itself fails — the caller (the - 3252
/// settle closure above) treats that as "nothing to continue with" and lets - 3253
/// `run_turn_chain` end the chain, exactly like `recover_ledger` returning - 3254
/// `false` used to leave `handle.session` untouched. - 3255
async fn recover_ledger(core: &Core, session_id: &str) -> Option<vak_session::SessionLog> { - 3256
if session_id.is_empty() { - 3257
return None; - 3258
} - 3259
core.open_session(session_id).await.ok() - 3260
} - 3261
- 3262
/// The text sent back to a Telegram/Slack/Discord user as the bot's reply - 3263
/// for this turn. A completed turn's real answer passes through untouched; - 3264
/// anything that did not produce a normal answer falls back to the same - 3265
/// small set of human sentences `ClientEvent::RunFinished` uses for every - 3266
/// other client (`client_events::run_outcome_message`), never the raw - 3267
/// `TurnOutcome::Failed` error — a channel reply is not a debug log. - 3268
fn outcome_text(o: &vak_agent::TurnOutcome) -> String { - 3269
use crate::client_events::{RunOutcome, run_outcome_message}; - 3270
use vak_agent::TurnOutcome; - 3271
match o { - 3272
TurnOutcome::Completed { response } => { - 3273
let t = response.text_content(); - 3274
if t.trim().is_empty() { - 3275
"(no text)".into() - 3276
} else { - 3277
t - 3278
} - 3279
} - 3280
TurnOutcome::Aborted { partial } => partial
Indexing the workspace…
Vakyartha documentation is discovering safe artifacts, anchors, and source references.