Vakyartha documentation

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.
3074fn 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.
3088pub(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
3104async 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.
3255async 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.
3268fn 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
3281 .as_ref()
3282 .map(|m| m.text_content())
3283 .filter(|t| !t.trim().is_empty())
3284 .unwrap_or_else(|| run_outcome_message(RunOutcome::Stopped).into()),
3285 TurnOutcome::Failed { .. } => run_outcome_message(RunOutcome::Failed).into(),
3286 TurnOutcome::MaxTurnsReached => run_outcome_message(RunOutcome::MaxTurns).into(),
3287 }
3288}
3289
3290fn outcome_is_error(o: &vak_agent::TurnOutcome) -> bool {
3291 matches!(
3292 o,
3293 vak_agent::TurnOutcome::Failed { .. } | vak_agent::TurnOutcome::MaxTurnsReached
3294 )
3295}
3296
3297fn short_summary(text: &str) -> String {
3298 let first = text.lines().next().unwrap_or("").trim();
3299 let mut s: String = first.chars().take(80).collect();
3300 if first.chars().count() > 80 {
3301 s.push('…');
3302 }
3303 if s.is_empty() { "completed".into() } else { s }
3304}
3305
3306// ---- Outbound delivery ------------------------------------------------------
3307
3308fn http_client() -> reqwest::Client {
3309 static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
3310 CLIENT
3311 .get_or_init(|| {
3312 reqwest::Client::builder()
3313 .timeout(Duration::from_secs(15))
3314 .build()
3315 .unwrap_or_default()
3316 })
3317 .clone()
3318}
3319
3320/// Delivery plus its durable pull-side twin (docs/design/29-personal-os.md
3321/// P6): append the same signal before transport so a failed push cannot erase it.
3322/// `<home>/inbox.jsonl` under `inbox_kind` so unattended output survives
3323/// even when no chat channel is reachable. Inbox recording is best-effort
3324/// by contract — it can never fail a delivery that already happened.
3325pub(crate) async fn deliver_and_record(
3326 core: &Core,
3327 target: &str,
3328 text: &str,
3329 inbox_kind: vak_core::inbox::Kind,
3330 title: String,
3331 session_id: Option<&str>,
3332 task_id: Option<&str>,
3333) -> Result<(), String> {
3334 deliver_and_record_with_result(
3335 core, target, text, inbox_kind, title, session_id, task_id, None,
3336 )
3337 .await
3338 .map(|_| ())
3339}
3340
3341#[allow(clippy::too_many_arguments)]
3342pub(crate) async fn deliver_and_record_with_result(
3343 core: &Core,
3344 target: &str,
3345 text: &str,
3346 inbox_kind: vak_core::inbox::Kind,
3347 title: String,
3348 session_id: Option<&str>,
3349 task_id: Option<&str>,
3350 result_id: Option<&str>,
3351) -> Result<&'static str, String> {
3352 let dedupe_key = result_id.map(|result| format!("{target}|{result}|{}", inbox_kind as u8));
3353 let _ = vak_core::inbox::record_with_result_and_key(
3354 &core.shared_data_home(),
3355 inbox_kind,
3356 &title,
3357 text,
3358 session_id,
3359 task_id,
3360 result_id,
3361 dedupe_key.as_deref(),
3362 );
3363 let cleaned_text = crate::projection::clean_scaffolding(text);
3364 let mut answer = AnswerDraft::from_markdown(cleaned_text);
3365 if let Some(value) = task_id {
3366 answer.metadata.insert("vak_task_id".into(), value.into());
3367 }
3368 if let Some(value) = session_id {
3369 answer
3370 .metadata
3371 .insert("vak_session_id".into(), value.into());
3372 }
3373 if let Some(value) = result_id {
3374 answer.metadata.insert("vak_result_id".into(), value.into());
3375 }
3376 crate::delivery::deliver(
3377 core,
3378 target,
3379 DeliveryKind::TaskSummary,
3380 DeliveryContent::Answer(answer),
3381 )
3382 .await
3383 .map(|packet| {
3384 if packet
3385 .diagnostics
3386 .iter()
3387 .any(|diagnostic| diagnostic.starts_with("delivery held:"))
3388 {
3389 "queued"
3390 } else {
3391 "delivered"
3392 }
3393 })
3394}
3395
3396async fn deliver_approval_and_record(
3397 core: &Core,
3398 target: &str,
3399 approval: ApprovalPayload,
3400 inbox_kind: vak_core::inbox::Kind,
3401 title: String,
3402 session_id: Option<&str>,
3403 task_id: Option<&str>,
3404) -> Result<(), String> {
3405 let _ = vak_core::inbox::record(
3406 &core.shared_data_home(),
3407 inbox_kind,
3408 &title,
3409 &approval.detail,
3410 session_id,
3411 task_id,
3412 );
3413 crate::delivery::deliver(
3414 core,
3415 target,
3416 DeliveryKind::Approval,
3417 DeliveryContent::Approval(approval),
3418 )
3419 .await
3420 .map(|_| ())
3421}
3422
3423fn delivery_action(id: &str, label: &str, verb: &str, request_id: &str) -> DeliveryAction {
3424 DeliveryAction {
3425 id: id.into(),
3426 label: label.into(),
3427 verb: verb.into(),
3428 data: [("request_id".into(), request_id.into())]
3429 .into_iter()
3430 .collect(),
3431 }
3432}
3433
3434/// Transient webhook failures retry with bounded exponential backoff.
3435/// 4xx (except 429) are the receiver's permanent answer and return at once;
3436/// network errors, timeouts, 429 and 5xx are retried.
3437const WEBHOOK_ATTEMPTS: u32 = 3;
3438
3439fn webhook_retryable(status: Option<u16>) -> bool {
3440 match status {
3441 None => true,
3442 Some(429) => true,
3443 Some(c) => c >= 500,
3444 }
3445}
3446
3447pub(crate) async fn deliver_webhook_packet(
3448 core: &Core,
3449 name: &str,
3450 packet: &DeliveryPacket,
3451) -> Result<(), String> {
3452 let hook = core.config().gateway.webhooks.get(name).ok_or_else(|| {
3453 let known: Vec<&String> = core.config().gateway.webhooks.keys().collect();
3454 format!("unknown webhook '{name}'; configured: {known:?}")
3455 })?;
3456 // Fail closed: a configured credential that is missing must not turn
3457 // into an unauthenticated post of agent output.
3458 let token = match &hook.token_env {
3459 Some(env_name) => Some(
3460 vak_config::get_var(env_name)
3461 .ok_or_else(|| format!("webhook '{name}' token_env '{env_name}' is not set"))?,
3462 ),
3463 None => None,
3464 };
3465 let payload = serde_json::json!({
3466 "target": format!("webhook:{name}"),
3467 "text": packet.fallback_markdown,
3468 "ts": chrono::Utc::now().to_rfc3339(),
3469 "job_id": packet.job_id,
3470 "delivery": packet,
3471 });
3472
3473 let mut last_error = String::new();
3474 for attempt in 0..WEBHOOK_ATTEMPTS {
3475 if attempt > 0 {
3476 tokio::time::sleep(Duration::from_millis(400u64 << (attempt - 1))).await;
3477 }
3478 let mut req = http_client()
3479 .post(&hook.url)
3480 .header("Idempotency-Key", &packet.job_id)
3481 .json(&payload);
3482 if let Some(token) = &token {
3483 req = req.bearer_auth(token);
3484 }
3485 match req.send().await {
3486 Ok(resp) => {
3487 let status = resp.status();
3488 if status.is_success() {
3489 return Ok(());
3490 }
3491 last_error = format!("webhook '{name}' returned {status}");
3492 if !webhook_retryable(Some(status.as_u16())) {
3493 return Err(last_error);
3494 }
3495 }
3496 Err(e) => {
3497 last_error = format!("webhook '{name}' post failed: {e}");
3498 }
3499 }
3500 }
3501 Err(last_error)
3502}
3503
3504/// Reflection outcome reporting for chat surfaces: the success line keeps
3505/// its historical format; config-driven and raced-out skips are routine and
3506/// stay silent so a busy gateway does not spam its own log.
3507fn log_gateway_reflection(outcome: vak_core::reflection::ReflectionOutcome) {
3508 use vak_core::reflection::ReflectionOutcome as R;
3509 match outcome {
3510 R::Reflected {
3511 notes_added,
3512 skills_proposed,
3513 } => {
3514 if notes_added > 0 || skills_proposed {
3515 eprintln!(
3516 "[gateway] reflection: {notes_added} note(s) persisted, skill queued: {skills_proposed}"
3517 );
3518 }
3519 }
3520 R::Skipped { reason } => match reason {
3521 "already-in-flight" | "reflection-disabled" | "memory-writes-disabled" => {}
3522 other => eprintln!("[gateway] reflection skipped: {other}"),
3523 },
3524 }
3525}
3526
3527#[cfg(test)]
3528#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
3529mod tests {
3530 use super::*;
3531 use crate::inbox::{INBOX_DIR, sanitize_filename};
3532
3533 /// The path a note names, quoted after "at path".
3534 fn saved_path(note: &str) -> Option<&str> {
3535 note.split_once("at path \"")?
3536 .1
3537 .split_once('"')
3538 .map(|(path, _)| path)
3539 }
3540
3541 #[test]
3542 fn a_channel_reply_for_a_failed_run_is_human_text_never_the_raw_error() {
3543 let outcome = vak_agent::TurnOutcome::Failed {
3544 error: vak_llm::LlmError::Network("connection reset by peer at 10.0.0.1:443".into()),
3545 };
3546 let text = outcome_text(&outcome);
3547 assert!(
3548 !text.contains("10.0.0.1") && !text.contains("connection reset"),
3549 "raw provider error leaked into the channel reply: {text:?}"
3550 );
3551 assert_eq!(
3552 text,
3553 crate::client_events::run_outcome_message(crate::client_events::RunOutcome::Failed)
3554 );
3555 }
3556
3557 #[test]
3558 fn a_channel_reply_for_max_turns_names_the_step_limit_not_a_raw_code() {
3559 let text = outcome_text(&vak_agent::TurnOutcome::MaxTurnsReached);
3560 assert_eq!(
3561 text,
3562 crate::client_events::run_outcome_message(crate::client_events::RunOutcome::MaxTurns)
3563 );
3564 }
3565
3566 #[test]
3567 fn voice_prompt_is_a_normal_user_message() {
3568 let prompt = compose_voice_prompt(" turn the lights on ");
3569 assert_eq!(prompt.role, vak_llm::Role::User);
3570 assert_eq!(prompt.content.len(), 1);
3571 match &prompt.content[0] {
3572 vak_llm::ContentBlock::Text { text } => {
3573 assert_eq!(text, " turn the lights on ");
3574 }
3575 other => panic!("voice prompt used unexpected content block: {other:?}"),
3576 }
3577 }
3578
3579 #[test]
3580 fn a_voice_note_reaches_the_model_only_as_its_text() {
3581 let prompt = compose_prompt(
3582 "[voice note not transcribed: no speech was recognized]",
3583 &[InboundAttachment {
3584 kind: "audio".into(),
3585 mime: "audio/ogg".into(),
3586 data: "not-model-input".into(),
3587 filename: Some("voice.ogg".into()),
3588 error: None,
3589 }],
3590 std::path::Path::new("."),
3591 );
3592 assert_eq!(prompt.content.len(), 1);
3593 assert!(matches!(
3594 &prompt.content[0],
3595 vak_llm::ContentBlock::Text { text } if text.contains("no speech was recognized")
3596 ));
3597 }
3598
3599 #[test]
3600 fn inbound_audio_budget_marks_oversized_payloads() {
3601 struct Channel;
3602 impl InboundChannel for Channel {
3603 fn surface(&self) -> &'static str {
3604 "test"
3605 }
3606 }
3607 let encoded = "A".repeat(16 * 1024 * 1024 * 4 / 3 + 1);
3608 let request = InboundRequest::new(&Channel, "chat", "sender", "voice")
3609 .unwrap()
3610 .with_attachments(vec![serde_json::json!({"kind":"audio", "data": encoded})]);
3611 assert_eq!(request.attachments[0]["data"], "");
3612 assert_eq!(
3613 request.attachments[0]["error"],
3614 "audio attachment exceeds 16 MiB"
3615 );
3616 }
3617
3618 /// The console resolved a chat's mode WITHOUT the bot tier, so a bot
3619 /// pinned narrower than its chat ran narrow and displayed wide — and a
3620 /// chat with no pin under a bot that had one displayed the workspace's
3621 /// mode instead of the bot's. Showing a channel as wider than it runs is
3622 /// the one direction of error that matters here.
3623 #[test]
3624 fn a_bot_pin_narrows_the_chat_and_the_console_says_so() {
3625 use vak_config::PermissionMode::*;
3626 let ws = tempfile::tempdir().unwrap();
3627 vak_config::paths::isolate_home_for_tests();
3628 std::fs::create_dir_all(ws.path().join(".vak")).unwrap();
3629 std::fs::write(
3630 ws.path().join(".vak/config.toml"),
3631 "permission_mode = \"full-access\"\n",
3632 )
3633 .unwrap();
3634 vak_core::trust::record(ws.path()).unwrap();
3635
3636 // Chat asks for more than its bot allows: the bot wins.
3637 let r = resolve_channel_permission(ws.path(), Some(FullAccess), Some(ReadOnly));
3638 assert_eq!(r.effective, ReadOnly);
3639 assert_eq!(r.bot_mode, Some(ReadOnly));
3640 assert!(r.was_capped(), "a reduced grant must be visible");
3641
3642 // Chat has no pin: the bot's applies exactly. Nothing was reduced,
3643 // so this is not a capping event — but the console must still show
3644 // `workspace-write`, where it used to show the workspace's
3645 // `full-access` because the bot tier was never consulted.
3646 let r = resolve_channel_permission(ws.path(), None, Some(WorkspaceWrite));
3647 assert_eq!(r.effective, WorkspaceWrite);
3648 assert_eq!(r.workspace_mode, FullAccess);
3649 assert!(!r.was_capped());
3650
3651 // A chat narrower than its bot is not "capped" — it got what it asked.
3652 let r = resolve_channel_permission(ws.path(), Some(ReadOnly), Some(FullAccess));
3653 assert_eq!(r.effective, ReadOnly);
3654 assert!(!r.was_capped());
3655 }
3656
3657 /// The workspace ceiling still wins over both, in either order.
3658 #[test]
3659 fn the_workspace_ceiling_is_never_escaped_by_a_bot_or_a_chat() {
3660 use vak_config::PermissionMode::*;
3661 let ws = tempfile::tempdir().unwrap();
3662 vak_config::paths::isolate_home_for_tests();
3663 std::fs::create_dir_all(ws.path().join(".vak")).unwrap();
3664 std::fs::write(
3665 ws.path().join(".vak/config.toml"),
3666 "permission_mode = \"read-only\"\n",
3667 )
3668 .unwrap();
3669 vak_core::trust::record(ws.path()).unwrap();
3670
3671 for (chat, bot) in [
3672 (Some(FullAccess), Some(FullAccess)),
3673 (Some(FullAccess), None),
3674 (None, Some(FullAccess)),
3675 (None, None),
3676 ] {
3677 let r = resolve_channel_permission(ws.path(), chat, bot);
3678 assert_eq!(r.effective, ReadOnly, "chat={chat:?} bot={bot:?}");
3679 }
3680 }
3681
3682 /// `forward` with no usable target is not representable: both the
3683 /// constructor and the setter collapse it to `deny`, the same rule the
3684 /// config loader applies.
3685 #[test]
3686 fn an_unbacked_forward_policy_resolves_to_deny() {
3687 let timeout = Duration::from_secs(300);
3688 for approver in [None, Some(""), Some(" "), Some("no-colon")] {
3689 let policy = ApprovalPolicy::resolve("forward", approver, timeout);
3690 assert_eq!(policy.approvals, "deny", "approver={approver:?}");
3691 assert!(policy.approver.is_none());
3692 }
3693 let policy = ApprovalPolicy::resolve("forward", Some("telegram:42"), timeout);
3694 assert_eq!(policy.approvals, "forward");
3695 assert_eq!(policy.approver.as_deref(), Some("telegram:42"));
3696 }
3697
3698 /// Regression for the classic serde `Option<Option<T>>` trap: a plain
3699 /// double-`Option` field can't tell "the key was never sent" apart
3700 /// from "the key was sent as `null`" — both collapse to the outer
3701 /// `None`. `deserialize_present` is the fix; this locks in all three
3702 /// states a PATCH body actually needs.
3703 #[test]
3704 fn deserialize_present_distinguishes_absent_null_and_value() {
3705 #[derive(serde::Deserialize)]
3706 struct Body {
3707 #[serde(default, deserialize_with = "deserialize_present")]
3708 field: Option<Option<String>>,
3709 }
3710
3711 let absent: Body = serde_json::from_str("{}").unwrap();
3712 assert_eq!(absent.field, None, "key never sent must mean 'leave alone'");
3713
3714 let explicit_null: Body = serde_json::from_str(r#"{"field": null}"#).unwrap();
3715 assert_eq!(
3716 explicit_null.field,
3717 Some(None),
3718 "explicit null must mean 'clear it', not be indistinguishable from absent"
3719 );
3720
3721 let set: Body = serde_json::from_str(r#"{"field": "x"}"#).unwrap();
3722 assert_eq!(set.field, Some(Some("x".to_string())));
3723 }
3724
3725 #[test]
3726 fn bare_verdicts_have_no_gate_id() {
3727 assert_eq!(parse_verdict("yes"), Some((true, None)));
3728 assert_eq!(parse_verdict(" NO "), Some((false, None)));
3729 assert_eq!(parse_verdict("approve"), Some((true, None)));
3730 }
3731
3732 #[test]
3733 fn addressed_verdict_extracts_single_short_token() {
3734 assert_eq!(
3735 parse_verdict("yes ab12cd34"),
3736 Some((true, Some("ab12cd34".into())))
3737 );
3738 assert_eq!(
3739 parse_verdict("no deadbeef"),
3740 Some((false, Some("deadbeef".into())))
3741 );
3742 }
3743
3744 #[test]
3745 fn prose_after_verdict_is_never_an_id() {
3746 assert_eq!(parse_verdict("yes please do it now"), Some((true, None)));
3747 assert_eq!(parse_verdict("no way"), Some((false, None)));
3748 }
3749
3750 #[test]
3751 fn chatter_is_not_a_verdict() {
3752 assert_eq!(parse_verdict("sure thing"), None);
3753 assert_eq!(parse_verdict(""), None);
3754 assert_eq!(parse_verdict("approved!"), None);
3755 }
3756
3757 fn document(name: &str, bytes: &[u8]) -> InboundAttachment {
3758 use base64::Engine as _;
3759 InboundAttachment {
3760 mime: "application/octet-stream".into(),
3761 data: base64::engine::general_purpose::STANDARD.encode(bytes),
3762 filename: Some(name.into()),
3763 kind: "document".into(),
3764 error: None,
3765 }
3766 }
3767
3768 fn note(msg: &vak_llm::Message) -> &str {
3769 let vak_llm::ContentBlock::Text { text } = &msg.content[1] else {
3770 unreachable!("expected a text block for a document attachment");
3771 };
3772 text
3773 }
3774
3775 #[test]
3776 fn small_text_document_is_inlined_as_a_fenced_text_block() {
3777 let workspace = tempfile::tempdir().unwrap();
3778 let msg = compose_prompt(
3779 "check this",
3780 &[document("notes.py", b"print('hi')")],
3781 workspace.path(),
3782 );
3783 let text = note(&msg);
3784 assert!(text.contains("Attached file `notes.py`"));
3785 assert!(text.contains("print('hi')"));
3786 assert!(
3787 !workspace.path().join(INBOX_DIR).exists(),
3788 "inlined text is not saved"
3789 );
3790 }
3791
3792 #[test]
3793 fn large_text_is_saved_to_the_inbox_not_inlined() {
3794 let workspace = tempfile::tempdir().unwrap();
3795 let huge = "x".repeat(DOCUMENT_INLINE_MAX_BYTES + 1);
3796 let msg = compose_prompt(
3797 "check this",
3798 &[document("notes.py", huge.as_bytes())],
3799 workspace.path(),
3800 );
3801 let text = note(&msg);
3802 assert!(
3803 !text.contains("xxxx"),
3804 "the raw content must not be inlined"
3805 );
3806 let saved = saved_path(text).unwrap();
3807 assert!(saved.ends_with("-notes.py"), "{text}");
3808 assert_eq!(
3809 std::fs::read_to_string(workspace.path().join(saved)).unwrap(),
3810 huge
3811 );
3812 }
3813
3814 #[test]
3815 fn an_office_file_is_saved_and_named_never_inlined() {
3816 let workspace = tempfile::tempdir().unwrap();
3817 let bytes = vak_ooxml::fixtures::docx();
3818 let msg = compose_prompt(
3819 "summarise",
3820 &[document("Q3 report.docx", &bytes)],
3821 workspace.path(),
3822 );
3823 let text = note(&msg);
3824 assert!(text.contains("Read it with doc_read"), "{text}");
3825 assert!(text.contains("-Q3 report.docx"), "{text}");
3826 assert!(
3827 !text.contains("PK"),
3828 "no package bytes in the prompt: {text}"
3829 );
3830 let inbox = workspace.path().join(INBOX_DIR);
3831 let entries: Vec<_> = std::fs::read_dir(&inbox).unwrap().collect();
3832 assert_eq!(entries.len(), 1);
3833 let path = entries[0].as_ref().unwrap().path();
3834 assert_eq!(std::fs::read(path).unwrap(), bytes);
3835
3836 // The same bytes again are the same file, not a second copy.
3837 let again = compose_prompt(
3838 "again",
3839 &[document("Q3 report.docx", &bytes)],
3840 workspace.path(),
3841 );
3842 assert_eq!(
3843 note(&again).replace("again", ""),
3844 text.replace("summarise", "")
3845 );
3846 assert_eq!(std::fs::read_dir(&inbox).unwrap().count(), 1);
3847 }
3848
3849 #[test]
3850 fn other_binary_files_are_saved_and_described_honestly() {
3851 let workspace = tempfile::tempdir().unwrap();
3852 let msg = compose_prompt(
3853 "look",
3854 &[document("scan.pdf", b"%PDF-1.7\x00\xff\xfe binary")],
3855 workspace.path(),
3856 );
3857 let text = note(&msg);
3858 assert!(text.contains("not a text or Open XML file"), "{text}");
3859 assert!(!text.contains("%PDF"), "{text}");
3860 }
3861
3862 #[test]
3863 fn hostile_filenames_stay_in_the_inbox() {
3864 let workspace = tempfile::tempdir().unwrap();
3865 for name in [
3866 "../../etc/passwd",
3867 "..\\..\\boot.ini",
3868 "...",
3869 "a/b/.hidden",
3870 "sub\x00dir",
3871 ] {
3872 let msg = compose_prompt("x", &[document(name, b"\x00binary")], workspace.path());
3873 let text = note(&msg);
3874 let saved = saved_path(text).unwrap_or_else(|| panic!("{name}: {text}"));
3875 assert!(
3876 !saved[6..].contains('/') && !saved.contains(".."),
3877 "{name} -> {saved}"
3878 );
3879 assert!(workspace.path().join(saved).is_file(), "{name} -> {saved}");
3880 }
3881 assert_eq!(sanitize_filename("../../etc/passwd"), "passwd");
3882 assert_eq!(sanitize_filename("..."), "file");
3883 assert_eq!(sanitize_filename(".hidden"), "hidden");
3884 }
3885
3886 #[cfg(unix)]
3887 #[test]
3888 fn a_planted_inbox_symlink_cannot_redirect_the_write() {
3889 let workspace = tempfile::tempdir().unwrap();
3890 let outside = tempfile::tempdir().unwrap();
3891 std::os::unix::fs::symlink(outside.path(), workspace.path().join(INBOX_DIR)).unwrap();
3892 let msg = compose_prompt("x", &[document("a.bin", b"\x00")], workspace.path());
3893 assert!(note(&msg).contains("could not be saved"), "{}", note(&msg));
3894 assert_eq!(std::fs::read_dir(outside.path()).unwrap().count(), 0);
3895 }
3896
3897 #[test]
3898 fn a_document_over_the_channel_cap_is_not_received() {
3899 let workspace = tempfile::tempdir().unwrap();
3900 let big = vec![0u8; INBOUND_DOCUMENT_MAX_BYTES + 1];
3901 let msg = compose_prompt("x", &[document("big.bin", &big)], workspace.path());
3902 assert!(note(&msg).contains("channel limit; not received"));
3903 assert!(!workspace.path().join(INBOX_DIR).exists());
3904 }
3905
3906 #[test]
3907 fn image_attachment_still_becomes_vision_content() {
3908 let msg = compose_prompt(
3909 "look",
3910 &[InboundAttachment {
3911 mime: "image/png".into(),
3912 data: "aGVsbG8=".into(),
3913 filename: None,
3914 kind: "image".into(),
3915 error: None,
3916 }],
3917 std::path::Path::new("."),
3918 );
3919 assert!(matches!(
3920 msg.content[1],
3921 vak_llm::ContentBlock::Image { .. }
3922 ));
3923 }
3924
3925 #[test]
3926 fn webhook_retry_matrix() {
3927 assert!(webhook_retryable(None), "network error retries");
3928 assert!(webhook_retryable(Some(429)));
3929 assert!(webhook_retryable(Some(503)));
3930 assert!(!webhook_retryable(Some(401)), "auth is permanent");
3931 assert!(!webhook_retryable(Some(404)));
3932 assert!(!webhook_retryable(Some(200)));
3933 }
3934
3935 #[test]
3936 fn legacy_session_map_loads_as_versioned_binding_records() {
3937 let dir = tempfile::tempdir().unwrap();
3938 let core = Core::new(dir.path().to_path_buf()).unwrap();
3939 core.set_sessions_home(dir.path().join("home"));
3940 let path = bindings_path(&core.shared_data_home());
3941 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
3942 std::fs::write(&path, r#"{"telegram:42":"old-session"}"#).unwrap();
3943 let gateway = GatewayState::load(&core, true);
3944 let snapshot = gateway.snapshot();
3945 assert_eq!(snapshot.len(), 1);
3946 assert_eq!(snapshot[0].0, "telegram:42");
3947 assert_eq!(snapshot[0].1.session_id.as_deref(), Some("old-session"));
3948 assert!(snapshot[0].1.provider.is_none());
3949 }
3950
3951 #[tokio::test]
3952 async fn route_change_rotates_binding_without_rewriting_old_session() {
3953 let dir = tempfile::tempdir().unwrap();
3954 let core = Core::new(dir.path().to_path_buf()).unwrap();
3955 core.set_sessions_home(dir.path().join("home"));
3956 let state = AppState::new(core.clone());
3957 let old = core
3958 .start_session_with_route("provider-a".into(), "model-a".into())
3959 .await
3960 .unwrap();
3961 let old_id = old.header().unwrap().session_id.clone();
3962 crate::register_handle(
3963 &state,
3964 old_id.clone(),
3965 old,
3966 core.cwd().clone(),
3967 core.clone(),
3968 );
3969 state.gateway.bind(
3970 &core,
3971 "telegram:42".into(),
3972 old_id.clone(),
3973 "old-revision".into(),
3974 );
3975 state.gateway.set_route_override(
3976 &core,
3977 "telegram:42".into(),
3978 Some(("provider-b".into(), "model-b".into())),
3979 );
3980
3981 let fresh = resolve_session(&state, &core, "telegram:42").await.unwrap();
3982 assert_ne!(fresh.id, old_id);
3983 {
3984 let lock = fresh
3985 .session
3986 .lock()
3987 .unwrap_or_else(std::sync::PoisonError::into_inner);
3988 let contract = &lock.as_ref().unwrap().header().unwrap().contract;
3989 assert_eq!(contract.provider, "provider-b");
3990 assert_eq!(contract.model, "model-b");
3991 }
3992 let old_path =
3993 vak_session::SessionPath::new_session_file(&core.sessions_home(), core.cwd(), &old_id);
3994 assert!(old_path.is_file(), "old append-only ledger remains intact");
3995 }
3996
3997 #[tokio::test]
3998 async fn agent_change_rotates_binding_without_rewriting_old_session() {
3999 let dir = tempfile::tempdir().unwrap();
4000 let core = Core::new(dir.path().to_path_buf()).unwrap();

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.
3074fn 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.
3088pub(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
3104async 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.
3255async 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.
3268fn 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
3281 .as_ref()
3282 .map(|m| m.text_content())
3283 .filter(|t| !t.trim().is_empty())
3284 .unwrap_or_else(|| run_outcome_message(RunOutcome::Stopped).into()),
3285 TurnOutcome::Failed { .. } => run_outcome_message(RunOutcome::Failed).into(),
3286 TurnOutcome::MaxTurnsReached => run_outcome_message(RunOutcome::MaxTurns).into(),
3287 }
3288}
3289
3290fn outcome_is_error(o: &vak_agent::TurnOutcome) -> bool {
3291 matches!(
3292 o,
3293 vak_agent::TurnOutcome::Failed { .. } | vak_agent::TurnOutcome::MaxTurnsReached
3294 )
3295}
3296
3297fn short_summary(text: &str) -> String {
3298 let first = text.lines().next().unwrap_or("").trim();
3299 let mut s: String = first.chars().take(80).collect();
3300 if first.chars().count() > 80 {
3301 s.push('…');
3302 }
3303 if s.is_empty() { "completed".into() } else { s }
3304}
3305
3306// ---- Outbound delivery ------------------------------------------------------
3307
3308fn http_client() -> reqwest::Client {
3309 static CLIENT: std::sync::OnceLock<reqwest::Client> = std::sync::OnceLock::new();
3310 CLIENT
3311 .get_or_init(|| {
3312 reqwest::Client::builder()
3313 .timeout(Duration::from_secs(15))
3314 .build()
3315 .unwrap_or_default()
3316 })
3317 .clone()
3318}
3319
3320/// Delivery plus its durable pull-side twin (docs/design/29-personal-os.md
3321/// P6): append the same signal before transport so a failed push cannot erase it.
3322/// `<home>/inbox.jsonl` under `inbox_kind` so unattended output survives
3323/// even when no chat channel is reachable. Inbox recording is best-effort
3324/// by contract — it can never fail a delivery that already happened.
3325pub(crate) async fn deliver_and_record(
3326 core: &Core,
3327 target: &str,
3328 text: &str,
3329 inbox_kind: vak_core::inbox::Kind,
3330 title: String,
3331 session_id: Option<&str>,
3332 task_id: Option<&str>,
3333) -> Result<(), String> {
3334 deliver_and_record_with_result(
3335 core, target, text, inbox_kind, title, session_id, task_id, None,
3336 )
3337 .await
3338 .map(|_| ())
3339}
3340
3341#[allow(clippy::too_many_arguments)]
3342pub(crate) async fn deliver_and_record_with_result(
3343 core: &Core,
3344 target: &str,
3345 text: &str,
3346 inbox_kind: vak_core::inbox::Kind,
3347 title: String,
3348 session_id: Option<&str>,
3349 task_id: Option<&str>,
3350 result_id: Option<&str>,
3351) -> Result<&'static str, String> {
3352 let dedupe_key = result_id.map(|result| format!("{target}|{result}|{}", inbox_kind as u8));
3353 let _ = vak_core::inbox::record_with_result_and_key(
3354 &core.shared_data_home(),
3355 inbox_kind,
3356 &title,
3357 text,
3358 session_id,
3359 task_id,
3360 result_id,
3361 dedupe_key.as_deref(),
3362 );
3363 let cleaned_text = crate::projection::clean_scaffolding(text);
3364 let mut answer = AnswerDraft::from_markdown(cleaned_text);
3365 if let Some(value) = task_id {
3366 answer.metadata.insert("vak_task_id".into(), value.into());
3367 }
3368 if let Some(value) = session_id {
3369 answer
3370 .metadata
3371 .insert("vak_session_id".into(), value.into());
3372 }
3373 if let Some(value) = result_id {
3374 answer.metadata.insert("vak_result_id".into(), value.into());
3375 }
3376 crate::delivery::deliver(
3377 core,
3378 target,
3379 DeliveryKind::TaskSummary,
3380 DeliveryContent::Answer(answer),
3381 )
3382 .await
3383 .map(|packet| {
3384 if packet
3385 .diagnostics
3386 .iter()
3387 .any(|diagnostic| diagnostic.starts_with("delivery held:"))
3388 {
3389 "queued"
3390 } else {
3391 "delivered"
3392 }
3393 })
3394}
3395
3396async fn deliver_approval_and_record(
3397 core: &Core,
3398 target: &str,
3399 approval: ApprovalPayload,
3400 inbox_kind: vak_core::inbox::Kind,
3401 title: String,
3402 session_id: Option<&str>,
3403 task_id: Option<&str>,
3404) -> Result<(), String> {
3405 let _ = vak_core::inbox::record(
3406 &core.shared_data_home(),
3407 inbox_kind,
3408 &title,
3409 &approval.detail,
3410 session_id,
3411 task_id,
3412 );
3413 crate::delivery::deliver(
3414 core,
3415 target,
3416 DeliveryKind::Approval,
3417 DeliveryContent::Approval(approval),
3418 )
3419 .await
3420 .map(|_| ())
3421}
3422
3423fn delivery_action(id: &str, label: &str, verb: &str, request_id: &str) -> DeliveryAction {
3424 DeliveryAction {
3425 id: id.into(),
3426 label: label.into(),
3427 verb: verb.into(),
3428 data: [("request_id".into(), request_id.into())]
3429 .into_iter()
3430 .collect(),
3431 }
3432}
3433
3434/// Transient webhook failures retry with bounded exponential backoff.
3435/// 4xx (except 429) are the receiver's permanent answer and return at once;
3436/// network errors, timeouts, 429 and 5xx are retried.
3437const WEBHOOK_ATTEMPTS: u32 = 3;
3438
3439fn webhook_retryable(status: Option<u16>) -> bool {
3440 match status {
3441 None => true,
3442 Some(429) => true,
3443 Some(c) => c >= 500,
3444 }
3445}
3446
3447pub(crate) async fn deliver_webhook_packet(
3448 core: &Core,
3449 name: &str,
3450 packet: &DeliveryPacket,
3451) -> Result<(), String> {
3452 let hook = core.config().gateway.webhooks.get(name).ok_or_else(|| {
3453 let known: Vec<&String> = core.config().gateway.webhooks.keys().collect();
3454 format!("unknown webhook '{name}'; configured: {known:?}")
3455 })?;
3456 // Fail closed: a configured credential that is missing must not turn
3457 // into an unauthenticated post of agent output.
3458 let token = match &hook.token_env {
3459 Some(env_name) => Some(
3460 vak_config::get_var(env_name)
3461 .ok_or_else(|| format!("webhook '{name}' token_env '{env_name}' is not set"))?,
3462 ),
3463 None => None,
3464 };
3465 let payload = serde_json::json!({
3466 "target": format!("webhook:{name}"),
3467 "text": packet.fallback_markdown,
3468 "ts": chrono::Utc::now().to_rfc3339(),
3469 "job_id": packet.job_id,
3470 "delivery": packet,
3471 });
3472
3473 let mut last_error = String::new();
3474 for attempt in 0..WEBHOOK_ATTEMPTS {
3475 if attempt > 0 {
3476 tokio::time::sleep(Duration::from_millis(400u64 << (attempt - 1))).await;
3477 }
3478 let mut req = http_client()
3479 .post(&hook.url)
3480 .header("Idempotency-Key", &packet.job_id)
3481 .json(&payload);
3482 if let Some(token) = &token {
3483 req = req.bearer_auth(token);
3484 }
3485 match req.send().await {
3486 Ok(resp) => {
3487 let status = resp.status();
3488 if status.is_success() {
3489 return Ok(());
3490 }
3491 last_error = format!("webhook '{name}' returned {status}");
3492 if !webhook_retryable(Some(status.as_u16())) {
3493 return Err(last_error);
3494 }
3495 }
3496 Err(e) => {
3497 last_error = format!("webhook '{name}' post failed: {e}");
3498 }
3499 }
3500 }
3501 Err(last_error)
3502}
3503
3504/// Reflection outcome reporting for chat surfaces: the success line keeps
3505/// its historical format; config-driven and raced-out skips are routine and
3506/// stay silent so a busy gateway does not spam its own log.
3507fn log_gateway_reflection(outcome: vak_core::reflection::ReflectionOutcome) {
3508 use vak_core::reflection::ReflectionOutcome as R;
3509 match outcome {
3510 R::Reflected {
3511 notes_added,
3512 skills_proposed,
3513 } => {
3514 if notes_added > 0 || skills_proposed {
3515 eprintln!(
3516 "[gateway] reflection: {notes_added} note(s) persisted, skill queued: {skills_proposed}"
3517 );
3518 }
3519 }
3520 R::Skipped { reason } => match reason {
3521 "already-in-flight" | "reflection-disabled" | "memory-writes-disabled" => {}
3522 other => eprintln!("[gateway] reflection skipped: {other}"),
3523 },
3524 }
3525}
3526
3527#[cfg(test)]
3528#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
3529mod tests {
3530 use super::*;
3531 use crate::inbox::{INBOX_DIR, sanitize_filename};
3532
3533 /// The path a note names, quoted after "at path".
3534 fn saved_path(note: &str) -> Option<&str> {
3535 note.split_once("at path \"")?
3536 .1
3537 .split_once('"')
3538 .map(|(path, _)| path)
3539 }
3540
3541 #[test]
3542 fn a_channel_reply_for_a_failed_run_is_human_text_never_the_raw_error() {
3543 let outcome = vak_agent::TurnOutcome::Failed {
3544 error: vak_llm::LlmError::Network("connection reset by peer at 10.0.0.1:443".into()),
3545 };
3546 let text = outcome_text(&outcome);
3547 assert!(
3548 !text.contains("10.0.0.1") && !text.contains("connection reset"),
3549 "raw provider error leaked into the channel reply: {text:?}"
3550 );
3551 assert_eq!(
3552 text,
3553 crate::client_events::run_outcome_message(crate::client_events::RunOutcome::Failed)
3554 );
3555 }
3556
3557 #[test]
3558 fn a_channel_reply_for_max_turns_names_the_step_limit_not_a_raw_code() {
3559 let text = outcome_text(&vak_agent::TurnOutcome::MaxTurnsReached);
3560 assert_eq!(
3561 text,
3562 crate::client_events::run_outcome_message(crate::client_events::RunOutcome::MaxTurns)
3563 );
3564 }
3565
3566 #[test]
3567 fn voice_prompt_is_a_normal_user_message() {
3568 let prompt = compose_voice_prompt(" turn the lights on ");
3569 assert_eq!(prompt.role, vak_llm::Role::User);
3570 assert_eq!(prompt.content.len(), 1);
3571 match &prompt.content[0] {
3572 vak_llm::ContentBlock::Text { text } => {
3573 assert_eq!(text, " turn the lights on ");
3574 }
3575 other => panic!("voice prompt used unexpected content block: {other:?}"),
3576 }
3577 }
3578
3579 #[test]
3580 fn a_voice_note_reaches_the_model_only_as_its_text() {
3581 let prompt = compose_prompt(
3582 "[voice note not transcribed: no speech was recognized]",
3583 &[InboundAttachment {
3584 kind: "audio".into(),
3585 mime: "audio/ogg".into(),
3586 data: "not-model-input".into(),
3587 filename: Some("voice.ogg".into()),
3588 error: None,
3589 }],
3590 std::path::Path::new("."),
3591 );
3592 assert_eq!(prompt.content.len(), 1);
3593 assert!(matches!(
3594 &prompt.content[0],
3595 vak_llm::ContentBlock::Text { text } if text.contains("no speech was recognized")
3596 ));
3597 }
3598
3599 #[test]
3600 fn inbound_audio_budget_marks_oversized_payloads() {
3601 struct Channel;
3602 impl InboundChannel for Channel {
3603 fn surface(&self) -> &'static str {
3604 "test"
3605 }
3606 }
3607 let encoded = "A".repeat(16 * 1024 * 1024 * 4 / 3 + 1);
3608 let request = InboundRequest::new(&Channel, "chat", "sender", "voice")
3609 .unwrap()
3610 .with_attachments(vec![serde_json::json!({"kind":"audio", "data": encoded})]);
3611 assert_eq!(request.attachments[0]["data"], "");
3612 assert_eq!(
3613 request.attachments[0]["error"],
3614 "audio attachment exceeds 16 MiB"
3615 );
3616 }
3617
3618 /// The console resolved a chat's mode WITHOUT the bot tier, so a bot
3619 /// pinned narrower than its chat ran narrow and displayed wide — and a
3620 /// chat with no pin under a bot that had one displayed the workspace's
3621 /// mode instead of the bot's. Showing a channel as wider than it runs is
3622 /// the one direction of error that matters here.
3623 #[test]
3624 fn a_bot_pin_narrows_the_chat_and_the_console_says_so() {
3625 use vak_config::PermissionMode::*;
3626 let ws = tempfile::tempdir().unwrap();
3627 vak_config::paths::isolate_home_for_tests();
3628 std::fs::create_dir_all(ws.path().join(".vak")).unwrap();
3629 std::fs::write(
3630 ws.path().join(".vak/config.toml"),
3631 "permission_mode = \"full-access\"\n",
3632 )
3633 .unwrap();
3634 vak_core::trust::record(ws.path()).unwrap();
3635
3636 // Chat asks for more than its bot allows: the bot wins.
3637 let r = resolve_channel_permission(ws.path(), Some(FullAccess), Some(ReadOnly));
3638 assert_eq!(r.effective, ReadOnly);
3639 assert_eq!(r.bot_mode, Some(ReadOnly));
3640 assert!(r.was_capped(), "a reduced grant must be visible");
3641
3642 // Chat has no pin: the bot's applies exactly. Nothing was reduced,
3643 // so this is not a capping event — but the console must still show
3644 // `workspace-write`, where it used to show the workspace's
3645 // `full-access` because the bot tier was never consulted.
3646 let r = resolve_channel_permission(ws.path(), None, Some(WorkspaceWrite));
3647 assert_eq!(r.effective, WorkspaceWrite);
3648 assert_eq!(r.workspace_mode, FullAccess);
3649 assert!(!r.was_capped());
3650
3651 // A chat narrower than its bot is not "capped" — it got what it asked.
3652 let r = resolve_channel_permission(ws.path(), Some(ReadOnly), Some(FullAccess));
3653 assert_eq!(r.effective, ReadOnly);
3654 assert!(!r.was_capped());
3655 }
3656
3657 /// The workspace ceiling still wins over both, in either order.
3658 #[test]
3659 fn the_workspace_ceiling_is_never_escaped_by_a_bot_or_a_chat() {
3660 use vak_config::PermissionMode::*;
3661 let ws = tempfile::tempdir().unwrap();
3662 vak_config::paths::isolate_home_for_tests();
3663 std::fs::create_dir_all(ws.path().join(".vak")).unwrap();
3664 std::fs::write(
3665 ws.path().join(".vak/config.toml"),
3666 "permission_mode = \"read-only\"\n",
3667 )
3668 .unwrap();
3669 vak_core::trust::record(ws.path()).unwrap();
3670
3671 for (chat, bot) in [
3672 (Some(FullAccess), Some(FullAccess)),
3673 (Some(FullAccess), None),
3674 (None, Some(FullAccess)),
3675 (None, None),
3676 ] {
3677 let r = resolve_channel_permission(ws.path(), chat, bot);
3678 assert_eq!(r.effective, ReadOnly, "chat={chat:?} bot={bot:?}");
3679 }
3680 }
3681
3682 /// `forward` with no usable target is not representable: both the
3683 /// constructor and the setter collapse it to `deny`, the same rule the
3684 /// config loader applies.
3685 #[test]
3686 fn an_unbacked_forward_policy_resolves_to_deny() {
3687 let timeout = Duration::from_secs(300);
3688 for approver in [None, Some(""), Some(" "), Some("no-colon")] {
3689 let policy = ApprovalPolicy::resolve("forward", approver, timeout);
3690 assert_eq!(policy.approvals, "deny", "approver={approver:?}");
3691 assert!(policy.approver.is_none());
3692 }
3693 let policy = ApprovalPolicy::resolve("forward", Some("telegram:42"), timeout);
3694 assert_eq!(policy.approvals, "forward");
3695 assert_eq!(policy.approver.as_deref(), Some("telegram:42"));
3696 }
3697
3698 /// Regression for the classic serde `Option<Option<T>>` trap: a plain
3699 /// double-`Option` field can't tell "the key was never sent" apart
3700 /// from "the key was sent as `null`" — both collapse to the outer
3701 /// `None`. `deserialize_present` is the fix; this locks in all three
3702 /// states a PATCH body actually needs.
3703 #[test]
3704 fn deserialize_present_distinguishes_absent_null_and_value() {
3705 #[derive(serde::Deserialize)]
3706 struct Body {
3707 #[serde(default, deserialize_with = "deserialize_present")]
3708 field: Option<Option<String>>,
3709 }
3710
3711 let absent: Body = serde_json::from_str("{}").unwrap();
3712 assert_eq!(absent.field, None, "key never sent must mean 'leave alone'");
3713
3714 let explicit_null: Body = serde_json::from_str(r#"{"field": null}"#).unwrap();
3715 assert_eq!(
3716 explicit_null.field,
3717 Some(None),
3718 "explicit null must mean 'clear it', not be indistinguishable from absent"
3719 );
3720
3721 let set: Body = serde_json::from_str(r#"{"field": "x"}"#).unwrap();
3722 assert_eq!(set.field, Some(Some("x".to_string())));
3723 }
3724
3725 #[test]
3726 fn bare_verdicts_have_no_gate_id() {
3727 assert_eq!(parse_verdict("yes"), Some((true, None)));
3728 assert_eq!(parse_verdict(" NO "), Some((false, None)));
3729 assert_eq!(parse_verdict("approve"), Some((true, None)));
3730 }
3731
3732 #[test]
3733 fn addressed_verdict_extracts_single_short_token() {
3734 assert_eq!(
3735 parse_verdict("yes ab12cd34"),
3736 Some((true, Some("ab12cd34".into())))
3737 );
3738 assert_eq!(
3739 parse_verdict("no deadbeef"),
3740 Some((false, Some("deadbeef".into())))
3741 );
3742 }
3743
3744 #[test]
3745 fn prose_after_verdict_is_never_an_id() {
3746 assert_eq!(parse_verdict("yes please do it now"), Some((true, None)));
3747 assert_eq!(parse_verdict("no way"), Some((false, None)));
3748 }
3749
3750 #[test]
3751 fn chatter_is_not_a_verdict() {
3752 assert_eq!(parse_verdict("sure thing"), None);
3753 assert_eq!(parse_verdict(""), None);
3754 assert_eq!(parse_verdict("approved!"), None);
3755 }
3756
3757 fn document(name: &str, bytes: &[u8]) -> InboundAttachment {
3758 use base64::Engine as _;
3759 InboundAttachment {
3760 mime: "application/octet-stream".into(),
3761 data: base64::engine::general_purpose::STANDARD.encode(bytes),
3762 filename: Some(name.into()),
3763 kind: "document".into(),
3764 error: None,
3765 }
3766 }
3767
3768 fn note(msg: &vak_llm::Message) -> &str {
3769 let vak_llm::ContentBlock::Text { text } = &msg.content[1] else {
3770 unreachable!("expected a text block for a document attachment");
3771 };
3772 text
3773 }
3774
3775 #[test]
3776 fn small_text_document_is_inlined_as_a_fenced_text_block() {
3777 let workspace = tempfile::tempdir().unwrap();
3778 let msg = compose_prompt(
3779 "check this",
3780 &[document("notes.py", b"print('hi')")],
3781 workspace.path(),
3782 );
3783 let text = note(&msg);
3784 assert!(text.contains("Attached file `notes.py`"));
3785 assert!(text.contains("print('hi')"));
3786 assert!(
3787 !workspace.path().join(INBOX_DIR).exists(),
3788 "inlined text is not saved"
3789 );
3790 }
3791
3792 #[test]
3793 fn large_text_is_saved_to_the_inbox_not_inlined() {
3794 let workspace = tempfile::tempdir().unwrap();
3795 let huge = "x".repeat(DOCUMENT_INLINE_MAX_BYTES + 1);
3796 let msg = compose_prompt(
3797 "check this",
3798 &[document("notes.py", huge.as_bytes())],
3799 workspace.path(),
3800 );
3801 let text = note(&msg);
3802 assert!(
3803 !text.contains("xxxx"),
3804 "the raw content must not be inlined"
3805 );
3806 let saved = saved_path(text).unwrap();
3807 assert!(saved.ends_with("-notes.py"), "{text}");
3808 assert_eq!(
3809 std::fs::read_to_string(workspace.path().join(saved)).unwrap(),
3810 huge
3811 );
3812 }
3813
3814 #[test]
3815 fn an_office_file_is_saved_and_named_never_inlined() {
3816 let workspace = tempfile::tempdir().unwrap();
3817 let bytes = vak_ooxml::fixtures::docx();
3818 let msg = compose_prompt(
3819 "summarise",
3820 &[document("Q3 report.docx", &bytes)],
3821 workspace.path(),
3822 );
3823 let text = note(&msg);
3824 assert!(text.contains("Read it with doc_read"), "{text}");
3825 assert!(text.contains("-Q3 report.docx"), "{text}");
3826 assert!(
3827 !text.contains("PK"),
3828 "no package bytes in the prompt: {text}"
3829 );
3830 let inbox = workspace.path().join(INBOX_DIR);
3831 let entries: Vec<_> = std::fs::read_dir(&inbox).unwrap().collect();
3832 assert_eq!(entries.len(), 1);
3833 let path = entries[0].as_ref().unwrap().path();
3834 assert_eq!(std::fs::read(path).unwrap(), bytes);
3835
3836 // The same bytes again are the same file, not a second copy.
3837 let again = compose_prompt(
3838 "again",
3839 &[document("Q3 report.docx", &bytes)],
3840 workspace.path(),
3841 );
3842 assert_eq!(
3843 note(&again).replace("again", ""),
3844 text.replace("summarise", "")
3845 );
3846 assert_eq!(std::fs::read_dir(&inbox).unwrap().count(), 1);
3847 }
3848
3849 #[test]
3850 fn other_binary_files_are_saved_and_described_honestly() {
3851 let workspace = tempfile::tempdir().unwrap();
3852 let msg = compose_prompt(
3853 "look",
3854 &[document("scan.pdf", b"%PDF-1.7\x00\xff\xfe binary")],
3855 workspace.path(),
3856 );
3857 let text = note(&msg);
3858 assert!(text.contains("not a text or Open XML file"), "{text}");
3859 assert!(!text.contains("%PDF"), "{text}");
3860 }
3861
3862 #[test]
3863 fn hostile_filenames_stay_in_the_inbox() {
3864 let workspace = tempfile::tempdir().unwrap();
3865 for name in [
3866 "../../etc/passwd",
3867 "..\\..\\boot.ini",
3868 "...",
3869 "a/b/.hidden",
3870 "sub\x00dir",
3871 ] {
3872 let msg = compose_prompt("x", &[document(name, b"\x00binary")], workspace.path());
3873 let text = note(&msg);
3874 let saved = saved_path(text).unwrap_or_else(|| panic!("{name}: {text}"));
3875 assert!(
3876 !saved[6..].contains('/') && !saved.contains(".."),
3877 "{name} -> {saved}"
3878 );
3879 assert!(workspace.path().join(saved).is_file(), "{name} -> {saved}");
3880 }
3881 assert_eq!(sanitize_filename("../../etc/passwd"), "passwd");
3882 assert_eq!(sanitize_filename("..."), "file");
3883 assert_eq!(sanitize_filename(".hidden"), "hidden");
3884 }
3885
3886 #[cfg(unix)]
3887 #[test]
3888 fn a_planted_inbox_symlink_cannot_redirect_the_write() {
3889 let workspace = tempfile::tempdir().unwrap();
3890 let outside = tempfile::tempdir().unwrap();
3891 std::os::unix::fs::symlink(outside.path(), workspace.path().join(INBOX_DIR)).unwrap();
3892 let msg = compose_prompt("x", &[document("a.bin", b"\x00")], workspace.path());
3893 assert!(note(&msg).contains("could not be saved"), "{}", note(&msg));
3894 assert_eq!(std::fs::read_dir(outside.path()).unwrap().count(), 0);
3895 }
3896
3897 #[test]
3898 fn a_document_over_the_channel_cap_is_not_received() {
3899 let workspace = tempfile::tempdir().unwrap();
3900 let big = vec![0u8; INBOUND_DOCUMENT_MAX_BYTES + 1];
3901 let msg = compose_prompt("x", &[document("big.bin", &big)], workspace.path());
3902 assert!(note(&msg).contains("channel limit; not received"));
3903 assert!(!workspace.path().join(INBOX_DIR).exists());
3904 }
3905
3906 #[test]
3907 fn image_attachment_still_becomes_vision_content() {
3908 let msg = compose_prompt(
3909 "look",
3910 &[InboundAttachment {
3911 mime: "image/png".into(),
3912 data: "aGVsbG8=".into(),
3913 filename: None,
3914 kind: "image".into(),
3915 error: None,
3916 }],
3917 std::path::Path::new("."),
3918 );
3919 assert!(matches!(
3920 msg.content[1],
3921 vak_llm::ContentBlock::Image { .. }
3922 ));
3923 }
3924
3925 #[test]
3926 fn webhook_retry_matrix() {
3927 assert!(webhook_retryable(None), "network error retries");
3928 assert!(webhook_retryable(Some(429)));
3929 assert!(webhook_retryable(Some(503)));
3930 assert!(!webhook_retryable(Some(401)), "auth is permanent");
3931 assert!(!webhook_retryable(Some(404)));
3932 assert!(!webhook_retryable(Some(200)));
3933 }
3934
3935 #[test]
3936 fn legacy_session_map_loads_as_versioned_binding_records() {
3937 let dir = tempfile::tempdir().unwrap();
3938 let core = Core::new(dir.path().to_path_buf()).unwrap();
3939 core.set_sessions_home(dir.path().join("home"));
3940 let path = bindings_path(&core.shared_data_home());
3941 std::fs::create_dir_all(path.parent().unwrap()).unwrap();
3942 std::fs::write(&path, r#"{"telegram:42":"old-session"}"#).unwrap();
3943 let gateway = GatewayState::load(&core, true);
3944 let snapshot = gateway.snapshot();
3945 assert_eq!(snapshot.len(), 1);
3946 assert_eq!(snapshot[0].0, "telegram:42");
3947 assert_eq!(snapshot[0].1.session_id.as_deref(), Some("old-session"));
3948 assert!(snapshot[0].1.provider.is_none());
3949 }
3950
3951 #[tokio::test]
3952 async fn route_change_rotates_binding_without_rewriting_old_session() {
3953 let dir = tempfile::tempdir().unwrap();
3954 let core = Core::new(dir.path().to_path_buf()).unwrap();
3955 core.set_sessions_home(dir.path().join("home"));
3956 let state = AppState::new(core.clone());
3957 let old = core
3958 .start_session_with_route("provider-a".into(), "model-a".into())
3959 .await
3960 .unwrap();
3961 let old_id = old.header().unwrap().session_id.clone();
3962 crate::register_handle(
3963 &state,
3964 old_id.clone(),
3965 old,
3966 core.cwd().clone(),
3967 core.clone(),
3968 );
3969 state.gateway.bind(
3970 &core,
3971 "telegram:42".into(),
3972 old_id.clone(),
3973 "old-revision".into(),
3974 );
3975 state.gateway.set_route_override(
3976 &core,
3977 "telegram:42".into(),
3978 Some(("provider-b".into(), "model-b".into())),
3979 );
3980
3981 let fresh = resolve_session(&state, &core, "telegram:42").await.unwrap();
3982 assert_ne!(fresh.id, old_id);
3983 {
3984 let lock = fresh
3985 .session
3986 .lock()
3987 .unwrap_or_else(std::sync::PoisonError::into_inner);
3988 let contract = &lock.as_ref().unwrap().header().unwrap().contract;
3989 assert_eq!(contract.provider, "provider-b");
3990 assert_eq!(contract.model, "model-b");
3991 }
3992 let old_path =
3993 vak_session::SessionPath::new_session_file(&core.sessions_home(), core.cwd(), &old_id);
3994 assert!(old_path.is_file(), "old append-only ledger remains intact");
3995 }
3996
3997 #[tokio::test]
3998 async fn agent_change_rotates_binding_without_rewriting_old_session() {
3999 let dir = tempfile::tempdir().unwrap();
4000 let core = Core::new(dir.path().to_path_buf()).unwrap();

Indexing the workspace…