Vakyartha documentation

1132 approver: config.approver.clone(),
1133 }));
1134 config.tool_definitions = Some(vak_tools::definitions(&config.tools));
1135 }
1136 Agent {
1137 provider,
1138 session,
1139 config,
1140 run_call_counts: std::sync::Mutex::new(HashMap::new()),
1141 presented_cards: std::sync::Mutex::new(std::collections::HashSet::new()),
1142 deliveries: std::sync::Mutex::new(RunDeliveries::default()),
1143 call_yields: CallYields::default(),
1144 active_goal: None,
1145 obligations: Vec::new(),
1146 handoff_used: false,
1147 repair: RepairState::default(),
1148 drift_streak: 0,
1149 }
1150 }
1151
1152 /// One `WorkingSetPlan` for the ledger as it stands
1153 /// (`vak_context::plan_for_session`). Called fresh per step (and again
1154 /// after an incremental compaction) so it always reflects the latest
1155 /// chain; the lock is held only for the pure planning pass.
1156 async fn build_working_set_plan(
1157 &self,
1158 profile: &CapacityProfile,
1159 prefix_tokens: u64,
1160 tail_tokens: u64,
1161 ) -> vak_session::WorkingSetPlan {
1162 let session = self.session.lock().await;
1163 planner::plan_for_session(&session, profile, prefix_tokens, tail_tokens)
1164 }
1165
1166 /// The `CapacityProfile` to plan this turn against: the host-wired one
1167 /// (`Core::capacity_profile_for`) when present, or a metadata-only
1168 /// profile built from `declared_window`/`max_output` — the "construct
1169 /// `CapacityProfile::from_metadata_only` from the discovered window"
1170 /// fallback (docs/design/68-context-engine.md §4), never a magic
1171 /// number baked into the planner itself.
1172 fn effective_capacity_profile(&self) -> CapacityProfile {
1173 self.config.capacity.clone().unwrap_or_else(|| {
1174 CapacityProfile::from_metadata_only(
1175 self.config.declared_window,
1176 self.config.max_output,
1177 "no-capacity-wired".to_string(),
1178 std::time::SystemTime::now(),
1179 )
1180 })
1181 }
1182
1183 /// Arms goal mode for the next run: durable objective + acceptance
1184 /// criteria; completion becomes audited, never self-reported.
1185 pub fn set_goal(&mut self, objective: impl Into<String>, criteria: Vec<String>) {
1186 self.active_goal = Some(goal::GoalState {
1187 objective: objective.into(),
1188 criteria,
1189 audits_left: self.config.max_audit_blocks,
1190 });
1191 }
1192
1193 fn normalize_input(&self, message: Message) -> Result<Message, String> {
1194 match &self.config.input_normalizer {
1195 Some(normalizer) => normalizer(message),
1196 None => Ok(message),
1197 }
1198 }
1199
1200 /// The admitted set before `discovered_tools` is folded in: names,
1201 /// order and initial `defer` flags this agent was configured with
1202 /// (plus the conditional `work` tool). Stable across a run, so a
1203 /// caller that freezes ITS OWN clone of this once per turn
1204 /// (`turn_tools` in `run_message_inner`, §7) and re-applies
1205 /// `load_discovered` against that same clone every step gets a tools
1206 /// array whose base names/order never move -- only a `defer` flag may
1207 /// flip in place (a promotion, never a reorder) and a genuinely new
1208 /// name is appended at the end.
1209 fn base_tool_definitions(&self) -> Vec<vak_llm::ToolDefinition> {
1210 let mut definitions = self
1211 .config
1212 .tool_definitions
1213 .clone()
1214 .unwrap_or_else(|| vak_tools::definitions(&self.config.tools));
1215 // The `work` tool is offered iff the `flow` capability was admitted:
1216 // `flow_dispatcher` is None when admission rejected it, so this checks
1217 // that rather than `work_mode == Managed` alone.
1218 if self.config.flow_dispatcher.is_some()
1219 && !definitions
1220 .iter()
1221 .any(|definition| definition.name == "work")
1222 {
1223 definitions.push(vak_llm::ToolDefinition::new(
1224 "work",
1225 "Inspect and update the durable managed work contract. Model transitions may only move ready to running, running to blocked, or running to ready_for_verification.",
1226 serde_json::json!({
1227 "type": "object",
1228 "properties": {
1229 "operation": {"type": "string", "enum": ["get", "transition", "attach_evidence"]},
1230 "item_id": {"type": "string"},
1231 "to": {"type": "string", "enum": ["running", "blocked", "ready_for_verification"]},
1232 "reason": {"type": "string"},
1233 "evidence": {"type": "object"}
1234 },
1235 "required": ["operation"]
1236 }),
1237 ));
1238 }
1239 definitions
1240 }
1241
1242 fn tool_definitions(&self) -> Vec<vak_llm::ToolDefinition> {
1243 let mut definitions = self.base_tool_definitions();
1244 let discovered = self
1245 .config
1246 .discovered_tools
1247 .lock()
1248 .unwrap_or_else(std::sync::PoisonError::into_inner)
1249 .clone();
1250 load_discovered(&mut definitions, discovered);
1251 definitions
1252 }
1253
1254 pub async fn run(
1255 &mut self,
1256 prompt: &str,
1257 steering: &SteeringQueues,
1258 cancel: CancellationToken,
1259 events: mpsc::Sender<AgentEvent>,
1260 ) -> TurnOutcome {
1261 self.run_message(
1262 MessageRecord {
1263 message: Message::user_text(prompt),
1264 meta: None,
1265 },
1266 steering,
1267 cancel,
1268 events,
1269 )
1270 .await
1271 }
1272
1273 /// Run with a prebuilt prompt — the seam for multimodal (image) input
1274 /// and attached files; the ledger stores exactly what the model sees,
1275 /// with the prompt's metadata (`MessageMeta::attachments`).
1276 pub async fn run_message(
1277 &mut self,
1278 prompt: MessageRecord,
1279 steering: &SteeringQueues,
1280 cancel: CancellationToken,
1281 events: mpsc::Sender<AgentEvent>,
1282 ) -> TurnOutcome {
1283 let outcome = self
1284 .run_message_inner(prompt, steering, cancel.clone(), events.clone())
1285 .await;
1286 // A system-authored completion (drift, card-repeat and stale-data
1287 // outcomes) is the turn's answer as much as a model-authored one:
1288 // it must be on the ledger, or the turn never closes and the next
1289 // request would fold this one into it (invariant 1: model-visible
1290 // means logged).
1291 if let TurnOutcome::Completed { response } = &outcome {
1292 let already_logged = {
1293 let session = self.session.lock().await;
1294 session.message_chain().last().is_some_and(|(_, last)| {
1295 last.role == Role::Assistant && last.content == response.content
1296 })
1297 };
1298 if !already_logged {
1299 self.append_assistant(response).await;
1300 }
1301 }
1302 // Turn-close hook (docs/design/68-context-engine.md §10): builds and
1303 // appends the TurnCard once the turn has actually closed. A turn
1304 // that never got a final assistant text (most `Failed`/aborted
1305 // exits) stays open and this is a no-op — there is nothing to card
1306 // yet, and the next call to `run_message` will pick it up once it
1307 // does close.
1308 self.close_turn(&outcome).await;
1309 outcome
1310 }
1311
1312 async fn run_message_inner(
1313 &mut self,
1314 prompt: MessageRecord,
1315 steering: &SteeringQueues,
1316 cancel: CancellationToken,
1317 events: mpsc::Sender<AgentEvent>,
1318 ) -> TurnOutcome {
1319 let MessageRecord {
1320 message: prompt,
1321 meta: prompt_meta,
1322 } = prompt;
1323 let prompt = match self.normalize_input(prompt) {
1324 Ok(prompt) => prompt,
1325 Err(error) => {
1326 return TurnOutcome::Failed {
1327 error: LlmError::InvalidRequest(error),
1328 };
1329 }
1330 };
1331 let prompt_owned = prompt.text_content();
1332 let mut receipts = stop_policy::ReceiptSummary {
1333 continued_saved_file: self.config.continued_saved_file,
1334 ..Default::default()
1335 };
1336 let mut verification_stale = false;
1337 let mut user_completion_released = false;
1338 self.obligations.clear();
1339 self.handoff_used = false;
1340 self.run_call_counts
1341 .lock()
1342 .unwrap_or_else(std::sync::PoisonError::into_inner)
1343 .clear();
1344 self.presented_cards
1345 .lock()
1346 .unwrap_or_else(std::sync::PoisonError::into_inner)
1347 .clear();
1348 *self
1349 .deliveries
1350 .lock()
1351 .unwrap_or_else(std::sync::PoisonError::into_inner) = RunDeliveries::default();
1352 if self.active_goal.is_some() {
1353 // Goal lifecycle opens the run (audit-only entry).
1354 let mut session = self.session.lock().await;
1355 if let Some(g) = &self.active_goal {
1356 let _ = session.append_goal(vak_session::types::GoalEntry {
1357 goal_id: format!("goal-{}", chrono::Utc::now().timestamp_millis()),
1358 objective: g.objective.clone(),
1359 criteria: g.criteria.clone(),
1360 status: vak_session::types::GoalStatus::Active,
1361 });
1362 }
1363 }
1364 let prompt_entry = self.session.lock().await.append_message(MessageRecord {
1365 message: prompt,
1366 meta: prompt_meta,
1367 });
1368 let prompt_entry = match prompt_entry {
1369 Ok(entry) => entry,
1370 Err(error) => {
1371 return TurnOutcome::Failed {
1372 error: LlmError::Network(format!("session write failed: {error}")),
1373 };
1374 }
1375 };
1376 if self.config.work_mode == WorkMode::Managed && !self.config.work_enabled {
1377 return TurnOutcome::Failed {
1378 error: LlmError::InvalidRequest("managed work is disabled by configuration".into()),
1379 };
1380 }
1381 if self.config.work_enabled && self.config.work_mode == WorkMode::Managed {
1382 let entry_id = prompt_entry.id.clone();
1383 if let Err(error) = self.resolve_managed_input(&prompt_owned).await {
1384 return TurnOutcome::Failed {
1385 error: LlmError::InvalidRequest(error),
1386 };
1387 }
1388 if let Err(error) = self
1389 .start_managed_contract(&prompt_owned, entry_id, &cancel, &events)
1390 .await
1391 {
1392 return TurnOutcome::Failed {
1393 error: LlmError::InvalidRequest(error),
1394 };
1395 }
1396 self.emit_work_state(&events).await;
1397 if self
1398 .session
1399 .lock()
1400 .await
1401 .work_projection()
1402 .ok()
1403 .flatten()
1404 .is_some_and(|work| {
1405 work.status == vak_session::types::WorkContractStatus::AwaitingInput
1406 })
1407 {
1408 let mut response = AssistantMessage::empty(self.config.model.clone());
1409 let pending = self
1410 .session
1411 .lock()
1412 .await
1413 .work_projection()
1414 .ok()
1415 .flatten()
1416 .map(|work| {
1417 work.contract
1418 .assumptions
1419 .iter()
1420 .filter(|assumption| {
1421 assumption.requires_confirmation && assumption.resolution.is_none()
1422 })
1423 .map(|assumption| {
1424 format!("- {}: {}", assumption.assumption_id, assumption.text)
1425 })
1426 .collect::<Vec<_>>()
1427 .join("\n")
1428 })
1429 .unwrap_or_default();
1430 response.content.push(ContentBlock::text(
1431 format!(
1432 "I created the managed work contract, but need these assumptions confirmed before starting:\n{pending}\nReply with: answer: <your answer>.",
1433 ),
1434 ));
1435 let _ = self.session.lock().await.append_message(MessageRecord {
1436 message: response.clone().into_message(),
1437 meta: None,
1438 });
1439 return TurnOutcome::Completed { response };
1440 }
1441 }
1442
1443 // Captured once for the whole turn (docs/design/68-context-engine.md
1444 // §6/§10): every step's request must carry byte-identical tail bytes
1445 // so the provider's prefix cache serves the growing middle instead
1446 // of re-billing it on every step. Session-derived sections (intent,
1447 // work contract, conversation thread) are read from the ledger here
1448 // rather than supplied by the caller, since they are derived state,
1449 // not host configuration.
1450 //
1451 // The conversation thread section depends on the working-set plan
1452 // (it lists only directives the projection leaves out), and the
1453 // plan's budget depends on the tail's size — so the tail is sized
1454 // from a preliminary plan built against the thread-less sections,
1455 // then composed from the plan-aware ones. The per-step plans below
1456 // may differ from this one by at most the thread's own few lines;
1457 // a turn that demotes to `Card` as a result still carries its
1458 // directive on its card line.
1459 let turn_tail = {
1460 let profile = self.effective_capacity_profile();
1461 let tool_defs = self.tool_definitions();
1462 let prefix_tokens =
1463 profile.estimate_tokens(prefix_chars(&self.config.system_prefix, &tool_defs));
1464 let base_tail = {
1465 let session = self.session.lock().await;
1466 compose_tail(&self.config.tail, &session.tail_sections(None))
1467 };
1468 let base_tail_tokens = profile.estimate_tokens(base_tail.chars().count() as u64);
1469 let preliminary_plan = self
1470 .build_working_set_plan(&profile, prefix_tokens, base_tail_tokens)
1471 .await;
1472 let session = self.session.lock().await;
1473 let sections = session.tail_sections(Some(&preliminary_plan));
1474 compose_tail(&self.config.tail, &sections)
1475 };
1476
1477 let mut turn = 0usize;
1478 let mut outcome_turns = 0usize;
1479 self.repair.reset();
1480 self.run_call_counts
1481 .lock()
1482 .unwrap_or_else(std::sync::PoisonError::into_inner)
1483 .clear();
1484 self.presented_cards
1485 .lock()
1486 .unwrap_or_else(std::sync::PoisonError::into_inner)
1487 .clear();
1488 *self
1489 .deliveries
1490 .lock()
1491 .unwrap_or_else(std::sync::PoisonError::into_inner) = RunDeliveries::default();
1492 let mut stop_blocks_left = self
1493 .config
1494 .stop_policy
1495 .as_ref()
1496 .map(|p| p.max_blocks)
1497 .unwrap_or(0);
1498 // Names of retrieval-shaped tools that succeeded on the immediately
1499 // preceding turn (see `AgentConfig::retrieval_check`), cleared and
1500 // recomputed every time a tool-call batch runs. Consulted the very
1501 // next time the model produces a final text-only answer, to catch
1502 // the case where a search/fetch tool call succeeded and the model's
1503 // very next turn ignored the result instead of grounding on it.
1504 let mut pending_grounding_check: Option<Vec<String>> = None;
1505 let mut grounding_repair_attempted = false;
1506 // Whether any call that observes current state succeeded at any
1507 // point in this run, for the freshness check: a directive whose
1508 // reading carries the `live-data` domain (temporal deixis —
1509 // "current", "right now") asks for a value observed this turn, and
1510 // an answer or card that arrives without one is repeating what an
1511 // earlier turn found. A retrieval counts, and so does any other
1512 // observation (`AgentConfig::observation_check`): reading the file
1513 // or running the command is how a local "current" value is found.
1514 let mut observed_this_run = false;
1515 let mut freshness_repair_attempted = false;
1516 let mut empty_step_repair_attempted = false;
1517 let mut card_repeat_streak: u32 = 0;
1518 let mut topic_repair_attempted = false;
1519 // The most recent non-card tool result's own text this run, for the
1520 // fail-closed fallback below: when a topic-mismatched card has to be
1521 // refused twice, the evidence that WAS gathered is worth showing
1522 // rather than nothing (found live: a real `tavily_search` for "AI
1523 // news" succeeded, then the model wrote an unrelated "Noida Weather"
1524 // card twice in a row — the search result was sitting right there
1525 // and never got refused nothing).
1526 let mut last_evidence_snippet: Option<String> = None;
1527 let wants_live_data = self
1528 .session
1529 .lock()
1530 .await
1531 .latest_reading()
1532 .is_some_and(|reading| reading.domains.iter().any(|d| d == "live-data"));
1533 let mut malformed_fence_repair_attempted = false;
1534 // `semantic_type`s successfully emitted via an `emit_*_card` tool
1535 // call in the immediately preceding batch. A tool-emitted card is
1536 // already shown to the user (vak-server's projection pushes it from
1537 // the tool result independently of the assistant's own text), so if
1538 // the model's very next answer *also* writes a `vak` fence
1539 // repeating the same semantic_type, that's a second, duplicate card
1540 // — observed live against the real local model (gemma4:e2b-mlx):
1541 // it called `emit_chart_card` successfully, then still wrote out
1542 // the identical chart as a trailing fence. Same bounded-repair
1543 // pattern as `pending_grounding_check`/`malformed_fence_repair_attempted`.
1544 let mut pending_duplicate_card_check: Option<Vec<String>> = None;
1545 let mut duplicate_card_repair_attempted = false;
1546 // Whether any `emit_*_card` call succeeded so far in this run (unlike
1547 // `pending_duplicate_card_check`, which only covers the last batch),
1548 // and the one-shot flag for the presentation check below.
1549 let mut cards_emitted_this_run = false;
1550 // A tool delivered a file the person reviews through its own card
1551 // (an Office draft): the answer need not present it again.
1552 let mut file_delivered_this_run = false;
1553 let mut presentation_repair_attempted = false;
1554 // Append-only requests within a turn (docs/design/68-context-
1555 // engine.md §7): the working-set plan and the tools array are each
1556 // resolved ONCE per turn and reused by every later step, so a
1557 // message this turn already sent stays byte-identical on the next
1558 // step's request instead of silently changing shape underneath a
1559 // replayed thinking block. `turn_plan` is invalidated (set back to
1560 // `None`) only by a handoff reset; an over-length rejection or an
1561 // incremental compaction still re-plans mid-turn, but explicitly,
1562 // and the result becomes the new baseline for the rest of the
1563 // turn rather than being recomputed from scratch every step.
1564 // `turn_tool_base` freezes `base_tool_definitions()` (names, order,
1565 // and each one's INITIAL `defer` flag) at the first step;
1566 // `load_discovered` is re-applied against a fresh clone of it every
1567 // step, using whatever `discovered_tools` holds by then. A
1568 // `find_tools`/presentation-check promotion can still flip an
1569 // already-admitted tool's `defer` flag in place -- required so a
1570 // deferred tool that `tools_for_leg` was stripping from a
1571 // non-Anthropic leg's wire request actually starts being sent --
1572 // but the base set's names and order never move, and a genuinely
1573 // new name is only ever appended at the end.
1574 let mut turn_plan: Option<vak_session::WorkingSetPlan> = None;
1575 let mut turn_tool_base: Option<Vec<vak_llm::ToolDefinition>> = None;
1576 loop {
1577 if cancel.is_cancelled() {
1578 return TurnOutcome::Aborted { partial: None };
1579 }
1580 if !steering.wait_if_paused(&cancel).await {
1581 return TurnOutcome::Aborted { partial: None };
1582 }
1583 if turn >= self.config.max_turns {
1584 return TurnOutcome::MaxTurnsReached;
1585 }
1586 if self
1587 .config
1588 .outcome
1589 .as_ref()
1590 .and_then(|outcome| outcome.max_turns)
1591 .is_some_and(|max| outcome_turns >= max)
1592 {
1593 return TurnOutcome::MaxTurnsReached;
1594 }
1595
1596 {
1597 let mut session = self.session.lock().await;
1598 while let Some(update) = steering.take_outcome_update() {
1599 self.config.outcome = update.outcome.clone();
1600 if let Err(error) = session.append_intent(update) {
1601 return TurnOutcome::Failed {
1602 error: LlmError::Network(format!(
1603 "outcome revision write failed: {error}"
1604 )),
1605 };
1606 }
1607 }
1608 for message in steering.drain(DrainMode::OneAtATime) {
1609 let message = match self.normalize_input(message) {
1610 Ok(message) => message,
1611 Err(error) => {
1612 return TurnOutcome::Failed {
1613 error: LlmError::InvalidRequest(error),
1614 };
1615 }
1616 };
1617 // Only an explicit command changes the goal's shape; free
1618 // text adds to it (docs/design/47, control plane).
1619 let update = session.next_goal_update(&message.text_content());
1620 if let Err(error) = session.append_goal_update(update) {
1621 return TurnOutcome::Failed {
1622 error: LlmError::Network(format!("goal update write failed: {error}")),
1623 };
1624 }
1625 if StopPolicy::is_done_message(&message.text_content()) {
1626 user_completion_released = true;
1627 }
1628 let _ = session.append_message(MessageRecord {
1629 message,
1630 meta: None,
1631 });
1632 }
1633 }
1634
1635 let _ = events.send(AgentEvent::TurnStart { turn }).await;
1636
1637 // config.model is set by run_turn_inner from effective_model() on
1638 // every turn, so it always reflects the current live provider route.
1639 let model = self.config.model.clone();
1640
1641 // Working-set planning (docs/design/68-context-engine.md §4/§7/
1642 // §10): built once per TURN, not fresh every step, so a past
1643 // turn's fidelity cannot change mid-turn and break the
1644 // append-only request contract. `turn_plan`/`turn_tools`
1645 // (declared before the loop) hold the frozen baseline; only an
1646 // over-length rejection or an incremental compaction below
1647 // explicitly replans (updating the baseline for the rest of
1648 // the turn too), and only a handoff reset clears it back to
1649 // `None`. The lock is taken only for short read/plan/apply
1650 // phases and is NEVER held across the summarizer network call
1651 // below.
1652 let profile = self.effective_capacity_profile();
1653 if turn_tool_base.is_none() {
1654 turn_tool_base = Some(self.base_tool_definitions());
1655 }
1656 let mut tool_defs = turn_tool_base.clone().unwrap_or_default();
1657 let discovered = self
1658 .config
1659 .discovered_tools
1660 .lock()
1661 .unwrap_or_else(std::sync::PoisonError::into_inner)
1662 .clone();
1663 load_discovered(&mut tool_defs, discovered);
1664 let prefix_tokens =
1665 profile.estimate_tokens(prefix_chars(&self.config.system_prefix, &tool_defs));
1666 let tail_tokens = profile.estimate_tokens(turn_tail.chars().count() as u64);
1667
1668 let mut plan = match &turn_plan {
1669 Some(p) => p.clone(),
1670 None => {
1671 let fresh = self
1672 .build_working_set_plan(&profile, prefix_tokens, tail_tokens)
1673 .await;
1674 turn_plan = Some(fresh.clone());
1675 fresh
1676 }
1677 };
1678
1679 // No usable horizon at all: the open turn alone (plus prefix,
1680 // tail, output reserve) already exceeds the horizon. Nothing is
1681 // plannable, so the reset-with-handoff rescue is the surviving
1682 // recovery (§4's "the handoff reset stays as the recovery when
1683 // a profile has no usable horizon").
1684 if plan.budget == 0 {
1685 let est_tokens = prefix_tokens
1686 .saturating_add(tail_tokens)
1687 .saturating_add(profile.output_reserve);
1688 if !self.handoff_used && self.config.handoff_reset {
1689 self.handoff_used = true;
1690 if let Ok(handoff) = self
1691 .write_handoff(est_tokens, &prompt_owned, &cancel, &events)
1692 .await
1693 {
1694 let mut session = self.session.lock().await;
1695 match session.append_handoff_reset(handoff, est_tokens) {
1696 Ok(_) => {
1697 let _ = events
1698 .send(AgentEvent::HandoffReset {
1699 before_tokens: est_tokens,
1700 })
1701 .await;
1702 drop(session);
1703 // The reset replaced everything before it:
1704 // the frozen plan baseline is stale and
1705 // must be rebuilt against the new chain.
1706 turn_plan = None;
1707 continue;
1708 }
1709 Err(e) => {
1710 return TurnOutcome::Failed {
1711 error: LlmError::Context(format!(
1712 "context over budget and handoff write failed: {e}"
1713 )),
1714 };
1715 }
1716 }
1717 }
1718 }
1719 return TurnOutcome::Failed {
1720 error: LlmError::Context("context over budget: no usable horizon".into()),
1721 };
1722 }
1723
1724 // Incremental compaction (docs/design/68-context-engine.md §4):
1725 // the plan collapsed some turns into a packet that no existing
1726 // `Compaction` entry covers yet. Summarize their CARDS (never
1727 // raw history) and append one, then re-plan — the packet
1728 // disappears from the new plan once it is covered.
1729 if let Some((first_turn_id, last_turn_id)) = plan.packet_range.clone() {
1730 let needs_compaction = {
1731 let session = self.session.lock().await;
1732 session.packet_needs_compaction(&first_turn_id, &last_turn_id)
1733 };
1734 if needs_compaction {
1735 let (transcript, transcript_chars) = {
1736 let session = self.session.lock().await;
1737 session.packet_transcript(&first_turn_id, &last_turn_id)
1738 };
1739 let tokens_before = profile.estimate_tokens(transcript_chars);
1740 let _ = events
1741 .send(AgentEvent::ContextCompacting {
1742 estimated_tokens: tokens_before,
1743 })
1744 .await;
1745 let req = assemble::compaction_request(&model, &transcript);
1746 let mut ledger = StepLedger::new(
1747 WorkPurpose::Summarize,
1748 self.provider.name(),
1749 &model,
1750 self.config.dispatch_ceiling,
1751 );
1752 let summary_msg = match self
1753 .complete_with_reliability(&req, &cancel, &events, false, &mut ledger)
1754 .await
1755 {
1756 Ok(m) => {
1757 let sid = {
1758 let mut session = self.session.lock().await;
1759 let sid = session
1760 .header()
1761 .map(|h| h.session_id.clone())
1762 .unwrap_or_default();
1763 let _ = session.append_receipt(ledger.take_receipt());
1764 sid
1765 };
1766 if let Some(gate) = &self.config.spend_gate {
1767 gate.record_settled_with_latency(
1768 self.provider.name(),
1769 &m.model,
1770 &sid,
1771 &m.usage,
1772 ledger.receipt.attempts.iter().map(|a| a.latency_ms).sum(),
1773 );
1774 }
1775 m
1776 }
1777 Err(LlmError::Aborted { .. }) => {
1778 let _ = self
1779 .session
1780 .lock()
1781 .await
1782 .append_receipt(ledger.take_receipt());
1783 return TurnOutcome::Aborted { partial: None };
1784 }
1785 Err(e) => {
1786 let _ = self
1787 .session
1788 .lock()
1789 .await
1790 .append_receipt(ledger.take_receipt());
1791 return TurnOutcome::Failed {
1792 error: LlmError::Network(format!("compaction call failed: {e}")),
1793 };
1794 }
1795 };
1796 let summary = summary_msg.text_content();
1797 if summary.trim().is_empty() {
1798 return TurnOutcome::Failed {
1799 error: LlmError::Network("compaction produced an empty summary".into()),
1800 };
1801 }
1802 {
1803 let mut session = self.session.lock().await;
1804 if let Err(e) = session.append_incremental_compaction(
1805 &first_turn_id,
1806 &last_turn_id,
1807 &model,
1808 summary,
1809 tokens_before,
1810 ) {
1811 return TurnOutcome::Failed {
1812 error: LlmError::Network(format!("compaction write failed: {e}")),
1813 };
1814 }
1815 }
1816 let after_plan = self
1817 .build_working_set_plan(&profile, prefix_tokens, tail_tokens)
1818 .await;
1819 let _ = events
1820 .send(AgentEvent::ContextCompacted {
1821 before_tokens: tokens_before,
1822 after_tokens: after_plan.spent,
1823 summarized_turns: 0,
1824 })
1825 .await;
1826 plan = after_plan;
1827 // Compaction wrote a new ledger entry: the packeted
1828 // range is now covered, so the frozen baseline must
1829 // reflect it for the rest of this turn too.
1830 turn_plan = Some(plan.clone());
1831 }
1832 }
1833
1834 // One model step = connect + stream + collect, wrapped with the
1835 // full reliability machinery (watchdog, retries+backoff,
1836 // circuit breaker, dispatch ceiling). Every dispatch is recorded
1837 // into the work receipt, which lands in the ledger on every
1838 // exit path. User aborts and partial-output aborts are never
1839 // retried; they propagate for caller handling.
1840 let mut ledger = StepLedger::new(
1841 WorkPurpose::Execute,
1842 self.provider.name(),
1843 &model,
1844 self.config.dispatch_ceiling,
1845 );
1846 // Recorded on every exit path below, success or failure: the
1847 // digest describes what was SENT, not what came back
1848 // (docs/design/68-context-engine.md §6/§7).
1849 ledger.receipt.prefix_digest =
1850 assemble::prefix_digest(&self.config.system_prefix, &tool_defs);
1851 let base_request = {
1852 let session = self.session.lock().await;
1853 // `messages` is already the fidelity-selected projection
1854 // (docs/design/68-context-engine.md §4/§10): retrieved-by-
1855 // relevance turns ride at Full inside it, so there is no
1856 // separate proactive-retrieval prepend step any more.
1857 let (mut messages, directive_at) = session.derive_with_plan_and_directive(&plan);
1858 // The tail is one final text block on the turn's directive
1859 // (after any tool_result blocks the directive itself
1860 // carries), never a separate consecutive user message and
1861 // never re-homed onto a later step's tool result or nudge
1862 // (docs/design/68-context-engine.md §6/§7).
1863 attach_tail(&mut messages, &turn_tail, directive_at);
1864 let session_key = session
1865 .header()
1866 .map(|header| header.session_id.clone())
1867 .unwrap_or_default();
1868 let cache = (!session_key.is_empty()).then(|| vak_llm::CacheHints {
1869 session_key,
1870 breakpoints: cache_breakpoints(&messages),
1871 });
1872 ChatRequest {
1873 model,
1874 system: Some(self.config.system_prefix.clone()),
1875 messages,
1876 tools: tool_defs.clone(),
1877 max_tokens: self.config.max_output as u32,
1878 temperature: None,
1879 cache,
1880 previous_response_id: None,
1881 think: None,
1882 effort: None,
1883 }
1884 };
1885 let mut request = base_request.clone();
1886
1887 let mut response = {
1888 // Run-level endurance: a sustained fault window (rate-limit
1889 // burst, slow/hung upstream, truncating proxy) can outlast
1890 // one step's retry budget. The ledger has not been touched,
1891 // so re-attempting the whole turn is exact. Aborts, permanent
1892 // errors, and ceiling exhaustion still fail/abort immediately.
1893 let mut run_attempt: u32 = 0;
1894 let mut backoff_ms = self.config.run_retry_base_backoff_ms.max(1);
1895 // Over-length replan (docs/design/68-context-engine.md §5):
1896 // a provider context-length rejection is a CapacityProfile
1897 // contradiction, not a transient fault — retried once, with
1898 // the horizon lowered and the request replanned smaller. A
1899 // second rejection on the retry is the turn's failure.
1900 let mut context_replan_used = false;
1901 loop {
1902 match self
1903 .complete_with_reliability(&request, &cancel, &events, true, &mut ledger)
1904 .await
1905 {
1906 Ok(r) => break r,
1907 Err(LlmError::Aborted { partial }) => {
1908 let _ = self
1909 .session
1910 .lock()
1911 .await
1912 .append_receipt(ledger.take_receipt());
1913 let partial = partial.map(|boxed| *boxed);
1914 if let Some(p) = &partial {
1915 let _ = self.append_assistant(p).await;
1916 }
1917 return TurnOutcome::Aborted { partial };
1918 }
1919 Err(e) if ledger.budget.remaining() == 0 => {
1920 let _ = self
1921 .session
1922 .lock()
1923 .await
1924 .append_receipt(ledger.take_receipt());
1925 return TurnOutcome::Failed {
1926 error: LlmError::Network(format!(
1927 "dispatch ceiling of {} exhausted for this step; last error: {e}",
1928 self.config.dispatch_ceiling
1929 )),
1930 };
1931 }
1932 Err(LlmError::Context(reason)) if !context_replan_used => {
1933 context_replan_used = true;
1934 let request_tokens =
1935 profile.estimate_tokens(chat_request_chars(&request));
1936 let mut lowered = profile.clone();
1937 lowered.observe_over_length(request_tokens);
1938 self.config.capacity = Some(lowered.clone());
1939 let mut data = self.capacity_activity_data(&lowered);
1940 data.insert("reason".into(), reason.clone());
1941 data.insert("request_tokens".into(), request_tokens.to_string());
1942 self.record_activity(
1943 vak_session::ActivityKind::CapacityFeedback,
1944 vak_session::ActivityStatus::Succeeded,
1945 "Capacity horizon lowered by an over-length rejection".into(),
1946 Some(reason),
1947 data,
1948 )
1949 .await;
1950 let new_prefix_tokens = lowered.estimate_tokens(prefix_chars(
1951 &self.config.system_prefix,
1952 &tool_defs,
1953 ));
1954 let new_tail_tokens =
1955 lowered.estimate_tokens(turn_tail.chars().count() as u64);
1956 plan = self
1957 .build_working_set_plan(
1958 &lowered,
1959 new_prefix_tokens,
1960 new_tail_tokens,
1961 )
1962 .await;
1963 // The lowered plan is the new baseline for the
1964 // rest of this turn, not just this retry.
1965 turn_plan = Some(plan.clone());
1966 request = {
1967 let session = self.session.lock().await;
1968 let (mut messages, directive_at) =
1969 session.derive_with_plan_and_directive(&plan);
1970 attach_tail(&mut messages, &turn_tail, directive_at);
1971 // This request just changed shape earlier
1972 // than the open turn's own tail (a smaller
1973 // plan): replaying a thinking block from a
1974 // step already sent under the OLD shape is
1975 // rejected outright by a provider that
1976 // requires nothing earlier to have changed
1977 // since it was produced (docs/design/68
1978 // §7), so it is dropped here instead.
1979 strip_replayed_thinking(&mut messages);
1980 let session_key = session
1981 .header()
1982 .map(|header| header.session_id.clone())
1983 .unwrap_or_default();
1984 let cache =
1985 (!session_key.is_empty()).then(|| vak_llm::CacheHints {
1986 session_key,
1987 breakpoints: cache_breakpoints(&messages),
1988 });
1989 ChatRequest {
1990 model: self.config.model.clone(),
1991 system: Some(self.config.system_prefix.clone()),
1992 messages,
1993 tools: tool_defs.clone(),
1994 max_tokens: self.config.max_output as u32,
1995 temperature: None,
1996 cache,
1997 previous_response_id: None,
1998 think: None,
1999 effort: None,
2000 }
2001 };
2002 continue;
2003 }
2004 Err(e)
2005 if run_attempt < self.config.run_retry_attempts
2006 && is_transient_step_error(&e) =>
2007 {
2008 run_attempt += 1;
2009 let delay = backoff_ms.min(30_000);
2010 let reason = format!(
2011 "step exhausted ({e}); run-level re-attempt {run_attempt}/{}",
2012 self.config.run_retry_attempts
2013 );
2014 self.record_activity(
2015 vak_session::ActivityKind::Retry,
2016 vak_session::ActivityStatus::Running,
2017 format!("Retry attempt {run_attempt}"),
2018 Some(reason.clone()),
2019 [
2020 ("attempt".into(), run_attempt.to_string()),
2021 ("delay_ms".into(), delay.to_string()),
2022 ]
2023 .into(),
2024 )
2025 .await;
2026 let _ = events
2027 .send(AgentEvent::RetryScheduled {
2028 attempt: run_attempt,
2029 delay_ms: delay,
2030 reason,
2031 })
2032 .await;
2033 if tokio::select! {
2034 _ = cancel.cancelled() => false,
2035 _ = tokio::time::sleep(std::time::Duration::from_millis(delay)) => true,
2036 } {
2037 backoff_ms = backoff_ms.saturating_mul(2);
2038 continue;
2039 }
2040 let _ = self
2041 .session
2042 .lock()
2043 .await
2044 .append_receipt(ledger.take_receipt());
2045 return TurnOutcome::Aborted { partial: None };
2046 }
2047 Err(e) => {
2048 let _ = self
2049 .session
2050 .lock()
2051 .await
2052 .append_receipt(ledger.take_receipt());
2053 return TurnOutcome::Failed { error: e };
2054 }
2055 }
2056 }
2057 };
2058
2059 // Provider dialect quirks are normalized before the assistant
2060 // message reaches the append-only ledger. Execution, replay,
2061 // presentation projection, and the next model step must all see
2062 // the same canonical call rather than a live-only repaired copy.
2063 let mcp_index = self
2064 .config
2065 .mcp_tool_index
2066 .lock()
2067 .unwrap_or_else(std::sync::PoisonError::into_inner)
2068 .clone();
2069 normalize_response_tool_uses(&mut response, &self.config.tools, &mcp_index);
2070 // Some small models cannot emit a structured `tool_use` block
2071 // and spell one out as text instead: rewrite `response` BEFORE
2072 // it reaches the ledger, so what is recorded is already a
2073 // valid tool_use/tool_result pair rather than envelope text
2074 // (docs/design/68-context-engine.md §5/§7).
2075 let calls: Vec<PendingToolCall> = extract_tool_calls(&mut response, &self.config.tools)
2076 .into_iter()
2077 .map(normalize_tool_call)
2078 .collect();
2079
2080 outcome_turns += 1;
2081
2082 let usage = response.usage.clone();
2083 let mut settled_provider_slot: Option<String> = None;
2084 let settled_session_id = {
2085 let mut session = self.session.lock().await;
2086 let sid = session
2087 .header()
2088 .map(|h| h.session_id.clone())
2089 .unwrap_or_default();
2090 let mut receipt = ledger.take_receipt();
2091 let settled_provider = receipt.provider.clone();
2092 if !receipt.prefix_digest.is_empty() {
2093 // A digest that differs from the immediately preceding
2094 // receipt's is a cache-breaking event, surfaced so a
2095 // regression is visible in the ledger rather than only in
2096 // the bill (docs/design/68-context-engine.md §7).
2097 if let Some(previous) = last_prefix_digest(&session)
2098 .filter(|previous| previous != &receipt.prefix_digest)
2099 {
2100 let now = chrono::Utc::now();
2101 let activity = vak_session::ActivityRecord {
2102 activity_id: format!(
2103 "activity-{}",
2104 now.timestamp_nanos_opt()
2105 .unwrap_or_else(|| now.timestamp_micros() * 1_000)
2106 ),
2107 turn: None,
2108 kind: vak_session::ActivityKind::Diagnostic,
2109 status: vak_session::ActivityStatus::Succeeded,
2110 label: "prefix-changed".to_string(),
2111 detail: None,
2112 data: [
2113 ("previous".to_string(), previous),
2114 ("current".to_string(), receipt.prefix_digest.clone()),
2115 ]
2116 .into_iter()
2117 .collect(),
2118 };
2119 let _ = session.append_activity(activity);
2120 }
2121 // Measured once per digest: the provider's reported input
2122 // tokens minus an estimate of the messages alone. A later
2123 // request with the same digest reuses this measurement
2124 // rather than re-deriving it from a cache-served step.
2125 if !prefix_digest_seen(&session, &receipt.prefix_digest) {
2126 let profile = self.effective_capacity_profile();
2127 let messages_tokens =
2128 profile.estimate_tokens(messages_chars(&request.messages));
2129 receipt.prefix_tokens =
2130 Some(usage.prompt_tokens().saturating_sub(messages_tokens));
2131 }

1132 approver: config.approver.clone(),
1133 }));
1134 config.tool_definitions = Some(vak_tools::definitions(&config.tools));
1135 }
1136 Agent {
1137 provider,
1138 session,
1139 config,
1140 run_call_counts: std::sync::Mutex::new(HashMap::new()),
1141 presented_cards: std::sync::Mutex::new(std::collections::HashSet::new()),
1142 deliveries: std::sync::Mutex::new(RunDeliveries::default()),
1143 call_yields: CallYields::default(),
1144 active_goal: None,
1145 obligations: Vec::new(),
1146 handoff_used: false,
1147 repair: RepairState::default(),
1148 drift_streak: 0,
1149 }
1150 }
1151
1152 /// One `WorkingSetPlan` for the ledger as it stands
1153 /// (`vak_context::plan_for_session`). Called fresh per step (and again
1154 /// after an incremental compaction) so it always reflects the latest
1155 /// chain; the lock is held only for the pure planning pass.
1156 async fn build_working_set_plan(
1157 &self,
1158 profile: &CapacityProfile,
1159 prefix_tokens: u64,
1160 tail_tokens: u64,
1161 ) -> vak_session::WorkingSetPlan {
1162 let session = self.session.lock().await;
1163 planner::plan_for_session(&session, profile, prefix_tokens, tail_tokens)
1164 }
1165
1166 /// The `CapacityProfile` to plan this turn against: the host-wired one
1167 /// (`Core::capacity_profile_for`) when present, or a metadata-only
1168 /// profile built from `declared_window`/`max_output` — the "construct
1169 /// `CapacityProfile::from_metadata_only` from the discovered window"
1170 /// fallback (docs/design/68-context-engine.md §4), never a magic
1171 /// number baked into the planner itself.
1172 fn effective_capacity_profile(&self) -> CapacityProfile {
1173 self.config.capacity.clone().unwrap_or_else(|| {
1174 CapacityProfile::from_metadata_only(
1175 self.config.declared_window,
1176 self.config.max_output,
1177 "no-capacity-wired".to_string(),
1178 std::time::SystemTime::now(),
1179 )
1180 })
1181 }
1182
1183 /// Arms goal mode for the next run: durable objective + acceptance
1184 /// criteria; completion becomes audited, never self-reported.
1185 pub fn set_goal(&mut self, objective: impl Into<String>, criteria: Vec<String>) {
1186 self.active_goal = Some(goal::GoalState {
1187 objective: objective.into(),
1188 criteria,
1189 audits_left: self.config.max_audit_blocks,
1190 });
1191 }
1192
1193 fn normalize_input(&self, message: Message) -> Result<Message, String> {
1194 match &self.config.input_normalizer {
1195 Some(normalizer) => normalizer(message),
1196 None => Ok(message),
1197 }
1198 }
1199
1200 /// The admitted set before `discovered_tools` is folded in: names,
1201 /// order and initial `defer` flags this agent was configured with
1202 /// (plus the conditional `work` tool). Stable across a run, so a
1203 /// caller that freezes ITS OWN clone of this once per turn
1204 /// (`turn_tools` in `run_message_inner`, §7) and re-applies
1205 /// `load_discovered` against that same clone every step gets a tools
1206 /// array whose base names/order never move -- only a `defer` flag may
1207 /// flip in place (a promotion, never a reorder) and a genuinely new
1208 /// name is appended at the end.
1209 fn base_tool_definitions(&self) -> Vec<vak_llm::ToolDefinition> {
1210 let mut definitions = self
1211 .config
1212 .tool_definitions
1213 .clone()
1214 .unwrap_or_else(|| vak_tools::definitions(&self.config.tools));
1215 // The `work` tool is offered iff the `flow` capability was admitted:
1216 // `flow_dispatcher` is None when admission rejected it, so this checks
1217 // that rather than `work_mode == Managed` alone.
1218 if self.config.flow_dispatcher.is_some()
1219 && !definitions
1220 .iter()
1221 .any(|definition| definition.name == "work")
1222 {
1223 definitions.push(vak_llm::ToolDefinition::new(
1224 "work",
1225 "Inspect and update the durable managed work contract. Model transitions may only move ready to running, running to blocked, or running to ready_for_verification.",
1226 serde_json::json!({
1227 "type": "object",
1228 "properties": {
1229 "operation": {"type": "string", "enum": ["get", "transition", "attach_evidence"]},
1230 "item_id": {"type": "string"},
1231 "to": {"type": "string", "enum": ["running", "blocked", "ready_for_verification"]},
1232 "reason": {"type": "string"},
1233 "evidence": {"type": "object"}
1234 },
1235 "required": ["operation"]
1236 }),
1237 ));
1238 }
1239 definitions
1240 }
1241
1242 fn tool_definitions(&self) -> Vec<vak_llm::ToolDefinition> {
1243 let mut definitions = self.base_tool_definitions();
1244 let discovered = self
1245 .config
1246 .discovered_tools
1247 .lock()
1248 .unwrap_or_else(std::sync::PoisonError::into_inner)
1249 .clone();
1250 load_discovered(&mut definitions, discovered);
1251 definitions
1252 }
1253
1254 pub async fn run(
1255 &mut self,
1256 prompt: &str,
1257 steering: &SteeringQueues,
1258 cancel: CancellationToken,
1259 events: mpsc::Sender<AgentEvent>,
1260 ) -> TurnOutcome {
1261 self.run_message(
1262 MessageRecord {
1263 message: Message::user_text(prompt),
1264 meta: None,
1265 },
1266 steering,
1267 cancel,
1268 events,
1269 )
1270 .await
1271 }
1272
1273 /// Run with a prebuilt prompt — the seam for multimodal (image) input
1274 /// and attached files; the ledger stores exactly what the model sees,
1275 /// with the prompt's metadata (`MessageMeta::attachments`).
1276 pub async fn run_message(
1277 &mut self,
1278 prompt: MessageRecord,
1279 steering: &SteeringQueues,
1280 cancel: CancellationToken,
1281 events: mpsc::Sender<AgentEvent>,
1282 ) -> TurnOutcome {
1283 let outcome = self
1284 .run_message_inner(prompt, steering, cancel.clone(), events.clone())
1285 .await;
1286 // A system-authored completion (drift, card-repeat and stale-data
1287 // outcomes) is the turn's answer as much as a model-authored one:
1288 // it must be on the ledger, or the turn never closes and the next
1289 // request would fold this one into it (invariant 1: model-visible
1290 // means logged).
1291 if let TurnOutcome::Completed { response } = &outcome {
1292 let already_logged = {
1293 let session = self.session.lock().await;
1294 session.message_chain().last().is_some_and(|(_, last)| {
1295 last.role == Role::Assistant && last.content == response.content
1296 })
1297 };
1298 if !already_logged {
1299 self.append_assistant(response).await;
1300 }
1301 }
1302 // Turn-close hook (docs/design/68-context-engine.md §10): builds and
1303 // appends the TurnCard once the turn has actually closed. A turn
1304 // that never got a final assistant text (most `Failed`/aborted
1305 // exits) stays open and this is a no-op — there is nothing to card
1306 // yet, and the next call to `run_message` will pick it up once it
1307 // does close.
1308 self.close_turn(&outcome).await;
1309 outcome
1310 }
1311
1312 async fn run_message_inner(
1313 &mut self,
1314 prompt: MessageRecord,
1315 steering: &SteeringQueues,
1316 cancel: CancellationToken,
1317 events: mpsc::Sender<AgentEvent>,
1318 ) -> TurnOutcome {
1319 let MessageRecord {
1320 message: prompt,
1321 meta: prompt_meta,
1322 } = prompt;
1323 let prompt = match self.normalize_input(prompt) {
1324 Ok(prompt) => prompt,
1325 Err(error) => {
1326 return TurnOutcome::Failed {
1327 error: LlmError::InvalidRequest(error),
1328 };
1329 }
1330 };
1331 let prompt_owned = prompt.text_content();
1332 let mut receipts = stop_policy::ReceiptSummary {
1333 continued_saved_file: self.config.continued_saved_file,
1334 ..Default::default()
1335 };
1336 let mut verification_stale = false;
1337 let mut user_completion_released = false;
1338 self.obligations.clear();
1339 self.handoff_used = false;
1340 self.run_call_counts
1341 .lock()
1342 .unwrap_or_else(std::sync::PoisonError::into_inner)
1343 .clear();
1344 self.presented_cards
1345 .lock()
1346 .unwrap_or_else(std::sync::PoisonError::into_inner)
1347 .clear();
1348 *self
1349 .deliveries
1350 .lock()
1351 .unwrap_or_else(std::sync::PoisonError::into_inner) = RunDeliveries::default();
1352 if self.active_goal.is_some() {
1353 // Goal lifecycle opens the run (audit-only entry).
1354 let mut session = self.session.lock().await;
1355 if let Some(g) = &self.active_goal {
1356 let _ = session.append_goal(vak_session::types::GoalEntry {
1357 goal_id: format!("goal-{}", chrono::Utc::now().timestamp_millis()),
1358 objective: g.objective.clone(),
1359 criteria: g.criteria.clone(),
1360 status: vak_session::types::GoalStatus::Active,
1361 });
1362 }
1363 }
1364 let prompt_entry = self.session.lock().await.append_message(MessageRecord {
1365 message: prompt,
1366 meta: prompt_meta,
1367 });
1368 let prompt_entry = match prompt_entry {
1369 Ok(entry) => entry,
1370 Err(error) => {
1371 return TurnOutcome::Failed {
1372 error: LlmError::Network(format!("session write failed: {error}")),
1373 };
1374 }
1375 };
1376 if self.config.work_mode == WorkMode::Managed && !self.config.work_enabled {
1377 return TurnOutcome::Failed {
1378 error: LlmError::InvalidRequest("managed work is disabled by configuration".into()),
1379 };
1380 }
1381 if self.config.work_enabled && self.config.work_mode == WorkMode::Managed {
1382 let entry_id = prompt_entry.id.clone();
1383 if let Err(error) = self.resolve_managed_input(&prompt_owned).await {
1384 return TurnOutcome::Failed {
1385 error: LlmError::InvalidRequest(error),
1386 };
1387 }
1388 if let Err(error) = self
1389 .start_managed_contract(&prompt_owned, entry_id, &cancel, &events)
1390 .await
1391 {
1392 return TurnOutcome::Failed {
1393 error: LlmError::InvalidRequest(error),
1394 };
1395 }
1396 self.emit_work_state(&events).await;
1397 if self
1398 .session
1399 .lock()
1400 .await
1401 .work_projection()
1402 .ok()
1403 .flatten()
1404 .is_some_and(|work| {
1405 work.status == vak_session::types::WorkContractStatus::AwaitingInput
1406 })
1407 {
1408 let mut response = AssistantMessage::empty(self.config.model.clone());
1409 let pending = self
1410 .session
1411 .lock()
1412 .await
1413 .work_projection()
1414 .ok()
1415 .flatten()
1416 .map(|work| {
1417 work.contract
1418 .assumptions
1419 .iter()
1420 .filter(|assumption| {
1421 assumption.requires_confirmation && assumption.resolution.is_none()
1422 })
1423 .map(|assumption| {
1424 format!("- {}: {}", assumption.assumption_id, assumption.text)
1425 })
1426 .collect::<Vec<_>>()
1427 .join("\n")
1428 })
1429 .unwrap_or_default();
1430 response.content.push(ContentBlock::text(
1431 format!(
1432 "I created the managed work contract, but need these assumptions confirmed before starting:\n{pending}\nReply with: answer: <your answer>.",
1433 ),
1434 ));
1435 let _ = self.session.lock().await.append_message(MessageRecord {
1436 message: response.clone().into_message(),
1437 meta: None,
1438 });
1439 return TurnOutcome::Completed { response };
1440 }
1441 }
1442
1443 // Captured once for the whole turn (docs/design/68-context-engine.md
1444 // §6/§10): every step's request must carry byte-identical tail bytes
1445 // so the provider's prefix cache serves the growing middle instead
1446 // of re-billing it on every step. Session-derived sections (intent,
1447 // work contract, conversation thread) are read from the ledger here
1448 // rather than supplied by the caller, since they are derived state,
1449 // not host configuration.
1450 //
1451 // The conversation thread section depends on the working-set plan
1452 // (it lists only directives the projection leaves out), and the
1453 // plan's budget depends on the tail's size — so the tail is sized
1454 // from a preliminary plan built against the thread-less sections,
1455 // then composed from the plan-aware ones. The per-step plans below
1456 // may differ from this one by at most the thread's own few lines;
1457 // a turn that demotes to `Card` as a result still carries its
1458 // directive on its card line.
1459 let turn_tail = {
1460 let profile = self.effective_capacity_profile();
1461 let tool_defs = self.tool_definitions();
1462 let prefix_tokens =
1463 profile.estimate_tokens(prefix_chars(&self.config.system_prefix, &tool_defs));
1464 let base_tail = {
1465 let session = self.session.lock().await;
1466 compose_tail(&self.config.tail, &session.tail_sections(None))
1467 };
1468 let base_tail_tokens = profile.estimate_tokens(base_tail.chars().count() as u64);
1469 let preliminary_plan = self
1470 .build_working_set_plan(&profile, prefix_tokens, base_tail_tokens)
1471 .await;
1472 let session = self.session.lock().await;
1473 let sections = session.tail_sections(Some(&preliminary_plan));
1474 compose_tail(&self.config.tail, &sections)
1475 };
1476
1477 let mut turn = 0usize;
1478 let mut outcome_turns = 0usize;
1479 self.repair.reset();
1480 self.run_call_counts
1481 .lock()
1482 .unwrap_or_else(std::sync::PoisonError::into_inner)
1483 .clear();
1484 self.presented_cards
1485 .lock()
1486 .unwrap_or_else(std::sync::PoisonError::into_inner)
1487 .clear();
1488 *self
1489 .deliveries
1490 .lock()
1491 .unwrap_or_else(std::sync::PoisonError::into_inner) = RunDeliveries::default();
1492 let mut stop_blocks_left = self
1493 .config
1494 .stop_policy
1495 .as_ref()
1496 .map(|p| p.max_blocks)
1497 .unwrap_or(0);
1498 // Names of retrieval-shaped tools that succeeded on the immediately
1499 // preceding turn (see `AgentConfig::retrieval_check`), cleared and
1500 // recomputed every time a tool-call batch runs. Consulted the very
1501 // next time the model produces a final text-only answer, to catch
1502 // the case where a search/fetch tool call succeeded and the model's
1503 // very next turn ignored the result instead of grounding on it.
1504 let mut pending_grounding_check: Option<Vec<String>> = None;
1505 let mut grounding_repair_attempted = false;
1506 // Whether any call that observes current state succeeded at any
1507 // point in this run, for the freshness check: a directive whose
1508 // reading carries the `live-data` domain (temporal deixis —
1509 // "current", "right now") asks for a value observed this turn, and
1510 // an answer or card that arrives without one is repeating what an
1511 // earlier turn found. A retrieval counts, and so does any other
1512 // observation (`AgentConfig::observation_check`): reading the file
1513 // or running the command is how a local "current" value is found.
1514 let mut observed_this_run = false;
1515 let mut freshness_repair_attempted = false;
1516 let mut empty_step_repair_attempted = false;
1517 let mut card_repeat_streak: u32 = 0;
1518 let mut topic_repair_attempted = false;
1519 // The most recent non-card tool result's own text this run, for the
1520 // fail-closed fallback below: when a topic-mismatched card has to be
1521 // refused twice, the evidence that WAS gathered is worth showing
1522 // rather than nothing (found live: a real `tavily_search` for "AI
1523 // news" succeeded, then the model wrote an unrelated "Noida Weather"
1524 // card twice in a row — the search result was sitting right there
1525 // and never got refused nothing).
1526 let mut last_evidence_snippet: Option<String> = None;
1527 let wants_live_data = self
1528 .session
1529 .lock()
1530 .await
1531 .latest_reading()
1532 .is_some_and(|reading| reading.domains.iter().any(|d| d == "live-data"));
1533 let mut malformed_fence_repair_attempted = false;
1534 // `semantic_type`s successfully emitted via an `emit_*_card` tool
1535 // call in the immediately preceding batch. A tool-emitted card is
1536 // already shown to the user (vak-server's projection pushes it from
1537 // the tool result independently of the assistant's own text), so if
1538 // the model's very next answer *also* writes a `vak` fence
1539 // repeating the same semantic_type, that's a second, duplicate card
1540 // — observed live against the real local model (gemma4:e2b-mlx):
1541 // it called `emit_chart_card` successfully, then still wrote out
1542 // the identical chart as a trailing fence. Same bounded-repair
1543 // pattern as `pending_grounding_check`/`malformed_fence_repair_attempted`.
1544 let mut pending_duplicate_card_check: Option<Vec<String>> = None;
1545 let mut duplicate_card_repair_attempted = false;
1546 // Whether any `emit_*_card` call succeeded so far in this run (unlike
1547 // `pending_duplicate_card_check`, which only covers the last batch),
1548 // and the one-shot flag for the presentation check below.
1549 let mut cards_emitted_this_run = false;
1550 // A tool delivered a file the person reviews through its own card
1551 // (an Office draft): the answer need not present it again.
1552 let mut file_delivered_this_run = false;
1553 let mut presentation_repair_attempted = false;
1554 // Append-only requests within a turn (docs/design/68-context-
1555 // engine.md §7): the working-set plan and the tools array are each
1556 // resolved ONCE per turn and reused by every later step, so a
1557 // message this turn already sent stays byte-identical on the next
1558 // step's request instead of silently changing shape underneath a
1559 // replayed thinking block. `turn_plan` is invalidated (set back to
1560 // `None`) only by a handoff reset; an over-length rejection or an
1561 // incremental compaction still re-plans mid-turn, but explicitly,
1562 // and the result becomes the new baseline for the rest of the
1563 // turn rather than being recomputed from scratch every step.
1564 // `turn_tool_base` freezes `base_tool_definitions()` (names, order,
1565 // and each one's INITIAL `defer` flag) at the first step;
1566 // `load_discovered` is re-applied against a fresh clone of it every
1567 // step, using whatever `discovered_tools` holds by then. A
1568 // `find_tools`/presentation-check promotion can still flip an
1569 // already-admitted tool's `defer` flag in place -- required so a
1570 // deferred tool that `tools_for_leg` was stripping from a
1571 // non-Anthropic leg's wire request actually starts being sent --
1572 // but the base set's names and order never move, and a genuinely
1573 // new name is only ever appended at the end.
1574 let mut turn_plan: Option<vak_session::WorkingSetPlan> = None;
1575 let mut turn_tool_base: Option<Vec<vak_llm::ToolDefinition>> = None;
1576 loop {
1577 if cancel.is_cancelled() {
1578 return TurnOutcome::Aborted { partial: None };
1579 }
1580 if !steering.wait_if_paused(&cancel).await {
1581 return TurnOutcome::Aborted { partial: None };
1582 }
1583 if turn >= self.config.max_turns {
1584 return TurnOutcome::MaxTurnsReached;
1585 }
1586 if self
1587 .config
1588 .outcome
1589 .as_ref()
1590 .and_then(|outcome| outcome.max_turns)
1591 .is_some_and(|max| outcome_turns >= max)
1592 {
1593 return TurnOutcome::MaxTurnsReached;
1594 }
1595
1596 {
1597 let mut session = self.session.lock().await;
1598 while let Some(update) = steering.take_outcome_update() {
1599 self.config.outcome = update.outcome.clone();
1600 if let Err(error) = session.append_intent(update) {
1601 return TurnOutcome::Failed {
1602 error: LlmError::Network(format!(
1603 "outcome revision write failed: {error}"
1604 )),
1605 };
1606 }
1607 }
1608 for message in steering.drain(DrainMode::OneAtATime) {
1609 let message = match self.normalize_input(message) {
1610 Ok(message) => message,
1611 Err(error) => {
1612 return TurnOutcome::Failed {
1613 error: LlmError::InvalidRequest(error),
1614 };
1615 }
1616 };
1617 // Only an explicit command changes the goal's shape; free
1618 // text adds to it (docs/design/47, control plane).
1619 let update = session.next_goal_update(&message.text_content());
1620 if let Err(error) = session.append_goal_update(update) {
1621 return TurnOutcome::Failed {
1622 error: LlmError::Network(format!("goal update write failed: {error}")),
1623 };
1624 }
1625 if StopPolicy::is_done_message(&message.text_content()) {
1626 user_completion_released = true;
1627 }
1628 let _ = session.append_message(MessageRecord {
1629 message,
1630 meta: None,
1631 });
1632 }
1633 }
1634
1635 let _ = events.send(AgentEvent::TurnStart { turn }).await;
1636
1637 // config.model is set by run_turn_inner from effective_model() on
1638 // every turn, so it always reflects the current live provider route.
1639 let model = self.config.model.clone();
1640
1641 // Working-set planning (docs/design/68-context-engine.md §4/§7/
1642 // §10): built once per TURN, not fresh every step, so a past
1643 // turn's fidelity cannot change mid-turn and break the
1644 // append-only request contract. `turn_plan`/`turn_tools`
1645 // (declared before the loop) hold the frozen baseline; only an
1646 // over-length rejection or an incremental compaction below
1647 // explicitly replans (updating the baseline for the rest of
1648 // the turn too), and only a handoff reset clears it back to
1649 // `None`. The lock is taken only for short read/plan/apply
1650 // phases and is NEVER held across the summarizer network call
1651 // below.
1652 let profile = self.effective_capacity_profile();
1653 if turn_tool_base.is_none() {
1654 turn_tool_base = Some(self.base_tool_definitions());
1655 }
1656 let mut tool_defs = turn_tool_base.clone().unwrap_or_default();
1657 let discovered = self
1658 .config
1659 .discovered_tools
1660 .lock()
1661 .unwrap_or_else(std::sync::PoisonError::into_inner)
1662 .clone();
1663 load_discovered(&mut tool_defs, discovered);
1664 let prefix_tokens =
1665 profile.estimate_tokens(prefix_chars(&self.config.system_prefix, &tool_defs));
1666 let tail_tokens = profile.estimate_tokens(turn_tail.chars().count() as u64);
1667
1668 let mut plan = match &turn_plan {
1669 Some(p) => p.clone(),
1670 None => {
1671 let fresh = self
1672 .build_working_set_plan(&profile, prefix_tokens, tail_tokens)
1673 .await;
1674 turn_plan = Some(fresh.clone());
1675 fresh
1676 }
1677 };
1678
1679 // No usable horizon at all: the open turn alone (plus prefix,
1680 // tail, output reserve) already exceeds the horizon. Nothing is
1681 // plannable, so the reset-with-handoff rescue is the surviving
1682 // recovery (§4's "the handoff reset stays as the recovery when
1683 // a profile has no usable horizon").
1684 if plan.budget == 0 {
1685 let est_tokens = prefix_tokens
1686 .saturating_add(tail_tokens)
1687 .saturating_add(profile.output_reserve);
1688 if !self.handoff_used && self.config.handoff_reset {
1689 self.handoff_used = true;
1690 if let Ok(handoff) = self
1691 .write_handoff(est_tokens, &prompt_owned, &cancel, &events)
1692 .await
1693 {
1694 let mut session = self.session.lock().await;
1695 match session.append_handoff_reset(handoff, est_tokens) {
1696 Ok(_) => {
1697 let _ = events
1698 .send(AgentEvent::HandoffReset {
1699 before_tokens: est_tokens,
1700 })
1701 .await;
1702 drop(session);
1703 // The reset replaced everything before it:
1704 // the frozen plan baseline is stale and
1705 // must be rebuilt against the new chain.
1706 turn_plan = None;
1707 continue;
1708 }
1709 Err(e) => {
1710 return TurnOutcome::Failed {
1711 error: LlmError::Context(format!(
1712 "context over budget and handoff write failed: {e}"
1713 )),
1714 };
1715 }
1716 }
1717 }
1718 }
1719 return TurnOutcome::Failed {
1720 error: LlmError::Context("context over budget: no usable horizon".into()),
1721 };
1722 }
1723
1724 // Incremental compaction (docs/design/68-context-engine.md §4):
1725 // the plan collapsed some turns into a packet that no existing
1726 // `Compaction` entry covers yet. Summarize their CARDS (never
1727 // raw history) and append one, then re-plan — the packet
1728 // disappears from the new plan once it is covered.
1729 if let Some((first_turn_id, last_turn_id)) = plan.packet_range.clone() {
1730 let needs_compaction = {
1731 let session = self.session.lock().await;
1732 session.packet_needs_compaction(&first_turn_id, &last_turn_id)
1733 };
1734 if needs_compaction {
1735 let (transcript, transcript_chars) = {
1736 let session = self.session.lock().await;
1737 session.packet_transcript(&first_turn_id, &last_turn_id)
1738 };
1739 let tokens_before = profile.estimate_tokens(transcript_chars);
1740 let _ = events
1741 .send(AgentEvent::ContextCompacting {
1742 estimated_tokens: tokens_before,
1743 })
1744 .await;
1745 let req = assemble::compaction_request(&model, &transcript);
1746 let mut ledger = StepLedger::new(
1747 WorkPurpose::Summarize,
1748 self.provider.name(),
1749 &model,
1750 self.config.dispatch_ceiling,
1751 );
1752 let summary_msg = match self
1753 .complete_with_reliability(&req, &cancel, &events, false, &mut ledger)
1754 .await
1755 {
1756 Ok(m) => {
1757 let sid = {
1758 let mut session = self.session.lock().await;
1759 let sid = session
1760 .header()
1761 .map(|h| h.session_id.clone())
1762 .unwrap_or_default();
1763 let _ = session.append_receipt(ledger.take_receipt());
1764 sid
1765 };
1766 if let Some(gate) = &self.config.spend_gate {
1767 gate.record_settled_with_latency(
1768 self.provider.name(),
1769 &m.model,
1770 &sid,
1771 &m.usage,
1772 ledger.receipt.attempts.iter().map(|a| a.latency_ms).sum(),
1773 );
1774 }
1775 m
1776 }
1777 Err(LlmError::Aborted { .. }) => {
1778 let _ = self
1779 .session
1780 .lock()
1781 .await
1782 .append_receipt(ledger.take_receipt());
1783 return TurnOutcome::Aborted { partial: None };
1784 }
1785 Err(e) => {
1786 let _ = self
1787 .session
1788 .lock()
1789 .await
1790 .append_receipt(ledger.take_receipt());
1791 return TurnOutcome::Failed {
1792 error: LlmError::Network(format!("compaction call failed: {e}")),
1793 };
1794 }
1795 };
1796 let summary = summary_msg.text_content();
1797 if summary.trim().is_empty() {
1798 return TurnOutcome::Failed {
1799 error: LlmError::Network("compaction produced an empty summary".into()),
1800 };
1801 }
1802 {
1803 let mut session = self.session.lock().await;
1804 if let Err(e) = session.append_incremental_compaction(
1805 &first_turn_id,
1806 &last_turn_id,
1807 &model,
1808 summary,
1809 tokens_before,
1810 ) {
1811 return TurnOutcome::Failed {
1812 error: LlmError::Network(format!("compaction write failed: {e}")),
1813 };
1814 }
1815 }
1816 let after_plan = self
1817 .build_working_set_plan(&profile, prefix_tokens, tail_tokens)
1818 .await;
1819 let _ = events
1820 .send(AgentEvent::ContextCompacted {
1821 before_tokens: tokens_before,
1822 after_tokens: after_plan.spent,
1823 summarized_turns: 0,
1824 })
1825 .await;
1826 plan = after_plan;
1827 // Compaction wrote a new ledger entry: the packeted
1828 // range is now covered, so the frozen baseline must
1829 // reflect it for the rest of this turn too.
1830 turn_plan = Some(plan.clone());
1831 }
1832 }
1833
1834 // One model step = connect + stream + collect, wrapped with the
1835 // full reliability machinery (watchdog, retries+backoff,
1836 // circuit breaker, dispatch ceiling). Every dispatch is recorded
1837 // into the work receipt, which lands in the ledger on every
1838 // exit path. User aborts and partial-output aborts are never
1839 // retried; they propagate for caller handling.
1840 let mut ledger = StepLedger::new(
1841 WorkPurpose::Execute,
1842 self.provider.name(),
1843 &model,
1844 self.config.dispatch_ceiling,
1845 );
1846 // Recorded on every exit path below, success or failure: the
1847 // digest describes what was SENT, not what came back
1848 // (docs/design/68-context-engine.md §6/§7).
1849 ledger.receipt.prefix_digest =
1850 assemble::prefix_digest(&self.config.system_prefix, &tool_defs);
1851 let base_request = {
1852 let session = self.session.lock().await;
1853 // `messages` is already the fidelity-selected projection
1854 // (docs/design/68-context-engine.md §4/§10): retrieved-by-
1855 // relevance turns ride at Full inside it, so there is no
1856 // separate proactive-retrieval prepend step any more.
1857 let (mut messages, directive_at) = session.derive_with_plan_and_directive(&plan);
1858 // The tail is one final text block on the turn's directive
1859 // (after any tool_result blocks the directive itself
1860 // carries), never a separate consecutive user message and
1861 // never re-homed onto a later step's tool result or nudge
1862 // (docs/design/68-context-engine.md §6/§7).
1863 attach_tail(&mut messages, &turn_tail, directive_at);
1864 let session_key = session
1865 .header()
1866 .map(|header| header.session_id.clone())
1867 .unwrap_or_default();
1868 let cache = (!session_key.is_empty()).then(|| vak_llm::CacheHints {
1869 session_key,
1870 breakpoints: cache_breakpoints(&messages),
1871 });
1872 ChatRequest {
1873 model,
1874 system: Some(self.config.system_prefix.clone()),
1875 messages,
1876 tools: tool_defs.clone(),
1877 max_tokens: self.config.max_output as u32,
1878 temperature: None,
1879 cache,
1880 previous_response_id: None,
1881 think: None,
1882 effort: None,
1883 }
1884 };
1885 let mut request = base_request.clone();
1886
1887 let mut response = {
1888 // Run-level endurance: a sustained fault window (rate-limit
1889 // burst, slow/hung upstream, truncating proxy) can outlast
1890 // one step's retry budget. The ledger has not been touched,
1891 // so re-attempting the whole turn is exact. Aborts, permanent
1892 // errors, and ceiling exhaustion still fail/abort immediately.
1893 let mut run_attempt: u32 = 0;
1894 let mut backoff_ms = self.config.run_retry_base_backoff_ms.max(1);
1895 // Over-length replan (docs/design/68-context-engine.md §5):
1896 // a provider context-length rejection is a CapacityProfile
1897 // contradiction, not a transient fault — retried once, with
1898 // the horizon lowered and the request replanned smaller. A
1899 // second rejection on the retry is the turn's failure.
1900 let mut context_replan_used = false;
1901 loop {
1902 match self
1903 .complete_with_reliability(&request, &cancel, &events, true, &mut ledger)
1904 .await
1905 {
1906 Ok(r) => break r,
1907 Err(LlmError::Aborted { partial }) => {
1908 let _ = self
1909 .session
1910 .lock()
1911 .await
1912 .append_receipt(ledger.take_receipt());
1913 let partial = partial.map(|boxed| *boxed);
1914 if let Some(p) = &partial {
1915 let _ = self.append_assistant(p).await;
1916 }
1917 return TurnOutcome::Aborted { partial };
1918 }
1919 Err(e) if ledger.budget.remaining() == 0 => {
1920 let _ = self
1921 .session
1922 .lock()
1923 .await
1924 .append_receipt(ledger.take_receipt());
1925 return TurnOutcome::Failed {
1926 error: LlmError::Network(format!(
1927 "dispatch ceiling of {} exhausted for this step; last error: {e}",
1928 self.config.dispatch_ceiling
1929 )),
1930 };
1931 }
1932 Err(LlmError::Context(reason)) if !context_replan_used => {
1933 context_replan_used = true;
1934 let request_tokens =
1935 profile.estimate_tokens(chat_request_chars(&request));
1936 let mut lowered = profile.clone();
1937 lowered.observe_over_length(request_tokens);
1938 self.config.capacity = Some(lowered.clone());
1939 let mut data = self.capacity_activity_data(&lowered);
1940 data.insert("reason".into(), reason.clone());
1941 data.insert("request_tokens".into(), request_tokens.to_string());
1942 self.record_activity(
1943 vak_session::ActivityKind::CapacityFeedback,
1944 vak_session::ActivityStatus::Succeeded,
1945 "Capacity horizon lowered by an over-length rejection".into(),
1946 Some(reason),
1947 data,
1948 )
1949 .await;
1950 let new_prefix_tokens = lowered.estimate_tokens(prefix_chars(
1951 &self.config.system_prefix,
1952 &tool_defs,
1953 ));
1954 let new_tail_tokens =
1955 lowered.estimate_tokens(turn_tail.chars().count() as u64);
1956 plan = self
1957 .build_working_set_plan(
1958 &lowered,
1959 new_prefix_tokens,
1960 new_tail_tokens,
1961 )
1962 .await;
1963 // The lowered plan is the new baseline for the
1964 // rest of this turn, not just this retry.
1965 turn_plan = Some(plan.clone());
1966 request = {
1967 let session = self.session.lock().await;
1968 let (mut messages, directive_at) =
1969 session.derive_with_plan_and_directive(&plan);
1970 attach_tail(&mut messages, &turn_tail, directive_at);
1971 // This request just changed shape earlier
1972 // than the open turn's own tail (a smaller
1973 // plan): replaying a thinking block from a
1974 // step already sent under the OLD shape is
1975 // rejected outright by a provider that
1976 // requires nothing earlier to have changed
1977 // since it was produced (docs/design/68
1978 // §7), so it is dropped here instead.
1979 strip_replayed_thinking(&mut messages);
1980 let session_key = session
1981 .header()
1982 .map(|header| header.session_id.clone())
1983 .unwrap_or_default();
1984 let cache =
1985 (!session_key.is_empty()).then(|| vak_llm::CacheHints {
1986 session_key,
1987 breakpoints: cache_breakpoints(&messages),
1988 });
1989 ChatRequest {
1990 model: self.config.model.clone(),
1991 system: Some(self.config.system_prefix.clone()),
1992 messages,
1993 tools: tool_defs.clone(),
1994 max_tokens: self.config.max_output as u32,
1995 temperature: None,
1996 cache,
1997 previous_response_id: None,
1998 think: None,
1999 effort: None,
2000 }
2001 };
2002 continue;
2003 }
2004 Err(e)
2005 if run_attempt < self.config.run_retry_attempts
2006 && is_transient_step_error(&e) =>
2007 {
2008 run_attempt += 1;
2009 let delay = backoff_ms.min(30_000);
2010 let reason = format!(
2011 "step exhausted ({e}); run-level re-attempt {run_attempt}/{}",
2012 self.config.run_retry_attempts
2013 );
2014 self.record_activity(
2015 vak_session::ActivityKind::Retry,
2016 vak_session::ActivityStatus::Running,
2017 format!("Retry attempt {run_attempt}"),
2018 Some(reason.clone()),
2019 [
2020 ("attempt".into(), run_attempt.to_string()),
2021 ("delay_ms".into(), delay.to_string()),
2022 ]
2023 .into(),
2024 )
2025 .await;
2026 let _ = events
2027 .send(AgentEvent::RetryScheduled {
2028 attempt: run_attempt,
2029 delay_ms: delay,
2030 reason,
2031 })
2032 .await;
2033 if tokio::select! {
2034 _ = cancel.cancelled() => false,
2035 _ = tokio::time::sleep(std::time::Duration::from_millis(delay)) => true,
2036 } {
2037 backoff_ms = backoff_ms.saturating_mul(2);
2038 continue;
2039 }
2040 let _ = self
2041 .session
2042 .lock()
2043 .await
2044 .append_receipt(ledger.take_receipt());
2045 return TurnOutcome::Aborted { partial: None };
2046 }
2047 Err(e) => {
2048 let _ = self
2049 .session
2050 .lock()
2051 .await
2052 .append_receipt(ledger.take_receipt());
2053 return TurnOutcome::Failed { error: e };
2054 }
2055 }
2056 }
2057 };
2058
2059 // Provider dialect quirks are normalized before the assistant
2060 // message reaches the append-only ledger. Execution, replay,
2061 // presentation projection, and the next model step must all see
2062 // the same canonical call rather than a live-only repaired copy.
2063 let mcp_index = self
2064 .config
2065 .mcp_tool_index
2066 .lock()
2067 .unwrap_or_else(std::sync::PoisonError::into_inner)
2068 .clone();
2069 normalize_response_tool_uses(&mut response, &self.config.tools, &mcp_index);
2070 // Some small models cannot emit a structured `tool_use` block
2071 // and spell one out as text instead: rewrite `response` BEFORE
2072 // it reaches the ledger, so what is recorded is already a
2073 // valid tool_use/tool_result pair rather than envelope text
2074 // (docs/design/68-context-engine.md §5/§7).
2075 let calls: Vec<PendingToolCall> = extract_tool_calls(&mut response, &self.config.tools)
2076 .into_iter()
2077 .map(normalize_tool_call)
2078 .collect();
2079
2080 outcome_turns += 1;
2081
2082 let usage = response.usage.clone();
2083 let mut settled_provider_slot: Option<String> = None;
2084 let settled_session_id = {
2085 let mut session = self.session.lock().await;
2086 let sid = session
2087 .header()
2088 .map(|h| h.session_id.clone())
2089 .unwrap_or_default();
2090 let mut receipt = ledger.take_receipt();
2091 let settled_provider = receipt.provider.clone();
2092 if !receipt.prefix_digest.is_empty() {
2093 // A digest that differs from the immediately preceding
2094 // receipt's is a cache-breaking event, surfaced so a
2095 // regression is visible in the ledger rather than only in
2096 // the bill (docs/design/68-context-engine.md §7).
2097 if let Some(previous) = last_prefix_digest(&session)
2098 .filter(|previous| previous != &receipt.prefix_digest)
2099 {
2100 let now = chrono::Utc::now();
2101 let activity = vak_session::ActivityRecord {
2102 activity_id: format!(
2103 "activity-{}",
2104 now.timestamp_nanos_opt()
2105 .unwrap_or_else(|| now.timestamp_micros() * 1_000)
2106 ),
2107 turn: None,
2108 kind: vak_session::ActivityKind::Diagnostic,
2109 status: vak_session::ActivityStatus::Succeeded,
2110 label: "prefix-changed".to_string(),
2111 detail: None,
2112 data: [
2113 ("previous".to_string(), previous),
2114 ("current".to_string(), receipt.prefix_digest.clone()),
2115 ]
2116 .into_iter()
2117 .collect(),
2118 };
2119 let _ = session.append_activity(activity);
2120 }
2121 // Measured once per digest: the provider's reported input
2122 // tokens minus an estimate of the messages alone. A later
2123 // request with the same digest reuses this measurement
2124 // rather than re-deriving it from a cache-served step.
2125 if !prefix_digest_seen(&session, &receipt.prefix_digest) {
2126 let profile = self.effective_capacity_profile();
2127 let messages_tokens =
2128 profile.estimate_tokens(messages_chars(&request.messages));
2129 receipt.prefix_tokens =
2130 Some(usage.prompt_tokens().saturating_sub(messages_tokens));
2131 }

Lib

Indexing the workspace…

Lib