Vakyartha documentation

388 serde_json::from_str(activity.data.get("key")?).ok()?;
389 if stored_key != key_json {
390 return None;
391 }
392 serde_json::from_str(activity.data.get("profile")?).ok()
393 })
394 }
395
396 /// Appends a turn's closing card (docs/design/68-context-engine.md §10).
397 /// Written once, at turn close, and never rewritten.
398 pub fn append_turn_card(&mut self, record: TurnCardRecord) -> Result<Entry, SessionError> {
399 let parent = self.tail_id.clone();
400 self.append(Entry::new(parent, EntryPayload::TurnCard(record)))
401 }
402
403 /// `(turn_id, card)` for every `TurnCard` entry along the active chain,
404 /// in the order they were written.
405 pub fn turn_cards(&self) -> Vec<(String, TurnCard)> {
406 self.chain_to_root()
407 .into_iter()
408 .filter_map(|entry| match &entry.payload {
409 EntryPayload::TurnCard(record) => {
410 Some((record.turn_id.clone(), record.card.clone()))
411 }
412 _ => None,
413 })
414 .collect()
415 }
416
417 /// Resolves a tool_use_id to its full `Evidence` — the tool name, its
418 /// call arguments, its whole result content (the evidence body when the
419 /// request carried a window of it), and whether it failed — by
420 /// scanning the active chain for the matching `ToolUse`/`ToolResult`
421 /// pair. Works for evidence in any turn, open or closed, which is what
422 /// lets `recall({ id })` reopen a result from a turn now reduced to a
423 /// card (docs/design/68-context-engine.md §3).
424 pub fn evidence(&self, tool_use_id: &str) -> Option<Evidence> {
425 let chain = self.chain_to_root();
426 let mut tool: Option<String> = None;
427 let mut input = serde_json::Value::Null;
428 for entry in &chain {
429 let EntryPayload::Message(record) = &entry.payload else {
430 continue;
431 };
432 for block in &record.message.content {
433 if let vak_llm::ContentBlock::ToolUse {
434 id,
435 name,
436 input: call_input,
437 } = block
438 && id == tool_use_id
439 {
440 tool = Some(name.clone());
441 input = call_input.clone();
442 }
443 }
444 }
445 let tool = tool?;
446 let body = chain.iter().find_map(|entry| match &entry.payload {
447 EntryPayload::EvidenceBody(body) if body.tool_use_id == tool_use_id => {
448 Some(body.content.clone())
449 }
450 _ => None,
451 });
452 for entry in &chain {
453 let EntryPayload::Message(record) = &entry.payload else {
454 continue;
455 };
456 for block in &record.message.content {
457 if let vak_llm::ContentBlock::ToolResult {
458 tool_use_id: id,
459 content,
460 is_error,
461 } = block
462 && id == tool_use_id
463 {
464 return Some(Evidence {
465 tool,
466 input,
467 content: body.unwrap_or_else(|| content.clone()),
468 is_error: *is_error,
469 });
470 }
471 }
472 }
473 None
474 }
475
476 /// Presentation entries along the active path, root→leaf, with their
477 /// entry ids — the single source both the model-visible history and the
478 /// display channel read.
479 pub fn presentations(&self) -> Vec<(String, &PresentationRecord)> {
480 self.chain_to_root()
481 .into_iter()
482 .filter_map(|e| match &e.payload {
483 EntryPayload::Presentation(record) => Some((e.id.clone(), record)),
484 _ => None,
485 })
486 .collect()
487 }
488
489 /// Whether a presentation with this exact canonical payload already
490 /// exists for this turn — the duplicate rule that drops a repeated
491 /// fence from the model-visible projection (docs/design/68-context-engine.md
492 /// §10: "a second card in the same turn with the same `payload_digest`
493 /// is not written").
494 pub fn has_presentation(&self, turn_id: &str, payload_digest: &str) -> bool {
495 self.presentations()
496 .iter()
497 .any(|(_, record)| record.turn_id == turn_id && record.payload_digest == payload_digest)
498 }
499
500 /// The id of the most recent non-control user message entry in the
501 /// active chain — the directive the current turn answers. `None` before
502 /// any real user message exists.
503 pub fn latest_directive_entry_id(&self) -> Option<String> {
504 self.chain_to_root()
505 .into_iter()
506 .rev()
507 .find_map(|entry| match &entry.payload {
508 EntryPayload::Message(record)
509 if record.message.role == vak_llm::Role::User
510 && record.control_kind().is_none()
511 && record
512 .message
513 .content
514 .iter()
515 .any(|b| matches!(b, vak_llm::ContentBlock::Text { .. }))
516 && !record
517 .message
518 .content
519 .iter()
520 .any(|b| matches!(b, vak_llm::ContentBlock::ToolResult { .. })) =>
521 {
522 Some(entry.id.clone())
523 }
524 _ => None,
525 })
526 }
527
528 /// Ids of every non-card tool result already committed to the ledger
529 /// strictly after `turn_id`, in chain order — the evidence a card built
530 /// after them was derived from (`PresentationRecord::derived_from`).
531 /// `is_card_tool` is supplied by the caller so this crate never needs to
532 /// know what a "card" tool is (that knowledge lives in
533 /// `vak-core::presentation_tools`).
534 pub fn non_card_evidence_since(
535 &self,
536 turn_id: &str,
537 is_card_tool: impl Fn(&str) -> bool,
538 ) -> Vec<String> {
539 let chain = self.chain_to_root();
540 let start = chain
541 .iter()
542 .position(|entry| entry.id == turn_id)
543 .map(|idx| idx + 1)
544 .unwrap_or(0);
545 let mut tool_names: HashMap<String, String> = HashMap::new();
546 let mut out = Vec::new();
547 for entry in &chain[start..] {
548 let EntryPayload::Message(record) = &entry.payload else {
549 continue;
550 };
551 for block in &record.message.content {
552 match block {
553 vak_llm::ContentBlock::ToolUse { id, name, .. } => {
554 tool_names.insert(id.clone(), name.clone());
555 }
556 vak_llm::ContentBlock::ToolResult { tool_use_id, .. } => {
557 let is_card = tool_names
558 .get(tool_use_id)
559 .map(|name| is_card_tool(name))
560 .unwrap_or(false);
561 if !is_card && !out.iter().any(|seen| seen == tool_use_id) {
562 out.push(tool_use_id.clone());
563 }
564 }
565 _ => {}
566 }
567 }
568 }
569 out
570 }
571
572 /// Append a finalized or provisional voice transcript. The transcript
573 /// is an audit projection; callers must append a normal Message entry
574 /// separately when the utterance is committed as model input.
575 pub fn append_voice_transcript(
576 &mut self,
577 activity_id: impl Into<String>,
578 text: impl Into<String>,
579 finalized: bool,
580 ) -> Result<Entry, SessionError> {
581 let mut data = std::collections::BTreeMap::new();
582 data.insert("text".into(), text.into());
583 data.insert("finalized".into(), finalized.to_string());
584 self.append_activity(crate::types::ActivityRecord {
585 activity_id: activity_id.into(),
586 turn: None,
587 kind: crate::types::ActivityKind::VoiceTranscript,
588 status: if finalized {
589 crate::types::ActivityStatus::Succeeded
590 } else {
591 crate::types::ActivityStatus::Running
592 },
593 label: "Voice transcript".into(),
594 detail: None,
595 data,
596 })
597 }
598
599 /// Append an auditable presentation selection without making it part of
600 /// model context. The original result and fallback remain authoritative.
601 pub fn append_presentation_selection(
602 &mut self,
603 activity_id: impl Into<String>,
604 semantic_type: impl Into<String>,
605 spec_id: impl Into<String>,
606 revision: u64,
607 mode: impl Into<String>,
608 fallback_used: bool,
609 ) -> Result<Entry, SessionError> {
610 let mut data = std::collections::BTreeMap::new();
611 data.insert("semantic_type".into(), semantic_type.into());
612 data.insert("spec_id".into(), spec_id.into());
613 data.insert("revision".into(), revision.to_string());
614 data.insert("mode".into(), mode.into());
615 data.insert("fallback_used".into(), fallback_used.to_string());
616 self.append_activity(crate::types::ActivityRecord {
617 activity_id: activity_id.into(),
618 turn: None,
619 kind: crate::types::ActivityKind::PresentationSelection,
620 status: crate::types::ActivityStatus::Succeeded,
621 label: "Presentation selected".into(),
622 detail: None,
623 data,
624 })
625 }
626
627 /// Append a projection choice or correction. If the text is sent to the
628 /// model, callers must also append a normal Message entry so derivation
629 /// remains complete.
630 pub fn append_presentation_feedback(
631 &mut self,
632 activity_id: impl Into<String>,
633 choice: impl Into<String>,
634 feedback: Option<String>,
635 ) -> Result<Entry, SessionError> {
636 let mut data = std::collections::BTreeMap::new();
637 data.insert("choice".into(), choice.into());
638 if let Some(feedback) = feedback {
639 data.insert("feedback".into(), feedback);
640 }
641 self.append_activity(crate::types::ActivityRecord {
642 activity_id: activity_id.into(),
643 turn: None,
644 kind: crate::types::ActivityKind::PresentationFeedback,
645 status: crate::types::ActivityStatus::Succeeded,
646 label: "Presentation feedback".into(),
647 detail: None,
648 data,
649 })
650 }
651
652 /// Append what the playback client reports it emitted. This is distinct
653 /// from provider output because buffering and interruption can prevent
654 /// the user from hearing the complete synthesis.
655 pub fn append_voice_playback(
656 &mut self,
657 activity_id: impl Into<String>,
658 emitted_ms: u64,
659 interrupted: bool,
660 ) -> Result<Entry, SessionError> {
661 let mut data = std::collections::BTreeMap::new();
662 data.insert("emitted_ms".into(), emitted_ms.to_string());
663 data.insert("interrupted".into(), interrupted.to_string());
664 self.append_activity(crate::types::ActivityRecord {
665 activity_id: activity_id.into(),
666 turn: None,
667 kind: crate::types::ActivityKind::VoicePlayback,
668 status: if interrupted {
669 crate::types::ActivityStatus::Partial
670 } else {
671 crate::types::ActivityStatus::Succeeded
672 },
673 label: "Voice playback".into(),
674 detail: None,
675 data,
676 })
677 }
678
679 /// Record this turn's resolved intent.
680 ///
681 /// Appended before the turn dispatches, so the note it carries is in the
682 /// projection the model actually sees — the entry *is* the record of what
683 /// was said, not a description of it (invariant 1).
684 pub fn append_intent(
685 &mut self,
686 record: crate::types::IntentRecord,
687 ) -> Result<Entry, SessionError> {
688 let parent = self.tail_id.clone();
689 self.append(Entry::new(parent, EntryPayload::Intent(Box::new(record))))
690 }
691
692 pub fn append_child_run_status(
693 &mut self,
694 status: crate::types::ChildRunStatus,
695 ) -> Result<Entry, SessionError> {
696 self.append_child_run_result(status, None)
697 }
698
699 pub fn append_child_run_result(
700 &mut self,
701 status: crate::types::ChildRunStatus,
702 outcome: Option<vak_intent::OutcomeSpec>,
703 ) -> Result<Entry, SessionError> {
704 let parent = self.tail_id.clone();
705 self.append(Entry::new(
706 parent,
707 EntryPayload::ChildRun { status, outcome },
708 ))
709 }
710
711 pub fn child_run_status(&self) -> Option<crate::types::ChildRunStatus> {
712 self.chain_to_root()
713 .iter()
714 .rev()
715 .find_map(|entry| match &entry.payload {
716 EntryPayload::ChildRun { status, .. } => Some(status.clone()),
717 _ => None,
718 })
719 }
720
721 pub fn child_run_outcome(&self) -> Option<vak_intent::OutcomeSpec> {
722 self.chain_to_root()
723 .iter()
724 .rev()
725 .find_map(|entry| match &entry.payload {
726 EntryPayload::ChildRun { outcome, .. } => outcome.clone(),
727 _ => None,
728 })
729 }
730
731 /// Records the interface a turn was bound to: in full when it differs
732 /// from the last one this conversation bound, otherwise as a
733 /// `TurnCapabilitiesRef` to that entry.
734 pub fn append_turn_capabilities(
735 &mut self,
736 bound: crate::types::TurnCapabilitiesBound,
737 ) -> Result<Entry, SessionError> {
738 let digest = bound.digest();
739 let previous =
740 self.chain_to_root()
741 .into_iter()
742 .rev()
743 .find_map(|entry| match &entry.payload {
744 EntryPayload::TurnCapabilitiesBound(earlier) => {
745 Some((entry.id.clone(), earlier.digest()))
746 }
747 _ => None,
748 });
749 let parent = self.tail_id.clone();
750 let payload = match previous {
751 Some((entry, earlier)) if earlier == digest => {
752 EntryPayload::TurnCapabilitiesRef(crate::types::TurnCapabilitiesRef {
753 entry,
754 digest,
755 epoch: bound.epoch,
756 })
757 }
758 _ => EntryPayload::TurnCapabilitiesBound(bound),
759 };
760 self.append(Entry::new(parent, payload))
761 }
762
763 pub fn append_work(&mut self, event: WorkEvent) -> Result<Entry, SessionError> {
764 let parent = self.tail_id.clone();
765 let candidate = Entry::new(parent, EntryPayload::Work(event));
766 let mut chain = self.chain_to_root();
767 chain.push(&candidate);
768 crate::work::project_work(&chain).map_err(|error| SessionError::Corrupt {
769 line: 0,
770 message: format!("invalid work event: {error}"),
771 })?;
772 let appended = self.append(candidate)?;
773 self.promote_ready_work_items()?;
774 Ok(appended)
775 }
776
777 fn promote_ready_work_items(&mut self) -> Result<(), SessionError> {
778 let Some(projection) = self
779 .work_projection()
780 .map_err(|error| SessionError::Corrupt {
781 line: 0,
782 message: error.to_string(),
783 })?
784 else {
785 return Ok(());
786 };
787 if projection.status != crate::types::WorkContractStatus::Active {
788 return Ok(());
789 }
790 let ready: Vec<String> = projection
791 .contract
792 .items
793 .iter()
794 .filter(|definition| {
795 projection
796 .items
797 .get(&definition.item_id)
798 .is_some_and(|state| state.status == crate::types::WorkItemStatus::Proposed)
799 && definition.dependencies.iter().all(|dependency| {
800 projection.items.get(dependency).is_some_and(|state| {
801 matches!(
802 state.status,
803 crate::types::WorkItemStatus::Succeeded
804 | crate::types::WorkItemStatus::Skipped
805 )
806 })
807 })
808 })
809 .map(|definition| definition.item_id.clone())
810 .collect();
811 for item_id in ready {
812 let current = self
813 .work_projection()
814 .map_err(|error| SessionError::Corrupt {
815 line: 0,
816 message: error.to_string(),
817 })?;
818 let Some(current) = current else { break };
819 self.append_work(crate::types::WorkEvent {
820 contract_id: current.contract.contract_id,
821 revision: current.contract.revision,
822 kind: crate::types::WorkEventKind::ItemStatusChanged {
823 item_id,
824 from: crate::types::WorkItemStatus::Proposed,
825 to: crate::types::WorkItemStatus::Ready,
826 attempt: 0,
827 reason: "dependencies satisfied".into(),
828 },
829 })?;
830 }
831 Ok(())
832 }
833
834 pub fn work_projection(
835 &self,
836 ) -> Result<Option<crate::work::WorkProjection>, crate::work::WorkError> {
837 crate::work::project_work(&self.chain_to_root())
838 }
839
840 pub fn evidence_exists(&self, evidence: &crate::types::EvidenceRef) -> bool {
841 let session_id = self.header().map(|header| header.session_id.as_str());
842 self.chain_to_root().iter().any(|entry| match evidence {
843 crate::types::EvidenceRef::LedgerEntry {
844 session_id: evidence_session,
845 entry_id,
846 }
847 | crate::types::EvidenceRef::Receipt {
848 session_id: evidence_session,
849 entry_id,
850 } => session_id == Some(evidence_session.as_str()) && entry.id == *entry_id,
851 crate::types::EvidenceRef::ToolResult {
852 session_id: evidence_session,
853 tool_use_id,
854 } => {
855 session_id == Some(evidence_session.as_str())
856 && matches!(
857 &entry.payload,
858 crate::types::EntryPayload::Message(record)
859 if record.message.content.iter().any(|block| matches!(
860 block,
861 vak_llm::ContentBlock::ToolResult { tool_use_id: id, .. }
862 if id == tool_use_id
863 ))
864 )
865 }
866 crate::types::EvidenceRef::CheckpointDiff { .. }
867 | crate::types::EvidenceRef::FlowNode { .. }
868 | crate::types::EvidenceRef::ChildSession { .. }
869 | crate::types::EvidenceRef::ExternalOperation { .. } => false,
870 })
871 }
872
873 /// Reconcile managed items left running by a process restart. Only child
874 /// sessions explicitly reported as live may remain running; every other
875 /// running item is conservatively interrupted and made retry-eligible.
876 /// Possible side effects are never replayed automatically.
877 pub fn reconcile_running_work(
878 &mut self,
879 live_child_sessions: &std::collections::HashSet<String>,
880 ) -> Result<usize, SessionError> {
881 self.reconcile_running_work_with_child_ledgers(live_child_sessions, None)
882 }
883
884 pub fn reconcile_running_work_with_child_ledgers(
885 &mut self,
886 live_child_sessions: &std::collections::HashSet<String>,
887 sessions_home: Option<&Path>,
888 ) -> Result<usize, SessionError> {
889 let Some(projection) = self
890 .work_projection()
891 .map_err(|error| SessionError::Corrupt {
892 line: 0,
893 message: error.to_string(),
894 })?
895 else {
896 return Ok(0);
897 };
898 let stale: Vec<(String, u32, Option<String>)> = projection
899 .items
900 .values()
901 .filter(|item| {
902 item.status == crate::types::WorkItemStatus::Running
903 && !item
904 .child_session_id
905 .as_ref()
906 .is_some_and(|id| live_child_sessions.contains(id))
907 })
908 .map(|item| {
909 (
910 item.item_id.clone(),
911 item.attempt,
912 item.child_session_id.clone(),
913 )
914 })
915 .collect();
916 let mut reconciled = 0;
917 for (item_id, attempt, child_id) in stale {
918 let Some(current) = self
919 .work_projection()
920 .map_err(|error| SessionError::Corrupt {
921 line: 0,
922 message: error.to_string(),
923 })?
924 else {
925 break;
926 };
927 let child_status = child_id.as_deref().and_then(|id| {
928 let home = sessions_home?;
929 let cwd = self.header().map(|h| h.contract_cwd())?;
930 let path = SessionPath::new_session_file(home, &cwd, id);
931 SessionLog::open(path)
932 .ok()
933 .and_then(|child| child.child_run_status())
934 });
935 if matches!(child_status, Some(crate::types::ChildRunStatus::Completed)) {
936 self.append_work(WorkEvent {
937 contract_id: current.contract.contract_id.clone(),
938 revision: current.contract.revision,
939 kind: crate::types::WorkEventKind::EvidenceAttached {
940 item_id: item_id.clone(),
941 evidence: crate::types::EvidenceRef::ChildSession {
942 session_id: child_id.clone().unwrap_or_default(),
943 },
944 },
945 })?;
946 self.append_work(WorkEvent {
947 contract_id: current.contract.contract_id.clone(),
948 revision: current.contract.revision,
949 kind: crate::types::WorkEventKind::ItemStatusChanged {
950 item_id,
951 from: crate::types::WorkItemStatus::Running,
952 to: crate::types::WorkItemStatus::ReadyForVerification,
953 attempt,
954 reason: "recovered completed child; verify its durable evidence".into(),
955 },
956 })?;
957 reconciled += 1;
958 continue;
959 }
960 self.append_work(WorkEvent {
961 contract_id: current.contract.contract_id.clone(),
962 revision: current.contract.revision,
963 kind: crate::types::WorkEventKind::ItemStatusChanged {
964 item_id,
965 from: crate::types::WorkItemStatus::Running,
966 to: crate::types::WorkItemStatus::Interrupted,
967 attempt,
968 reason: match child_status {
969 Some(crate::types::ChildRunStatus::Failed) => {
970 "child failed before restart; review before retry"
971 }
972 Some(crate::types::ChildRunStatus::Aborted) => {
973 "child was aborted before restart; review before retry"
974 }
975 Some(crate::types::ChildRunStatus::MaxTurns) => {
976 "child hit its turn limit; review before retry"
977 }
978 _ => "recovered after process restart; review before retry",
979 }
980 .into(),
981 },
982 })?;
983 reconciled += 1;
984 }
985 Ok(reconciled)
986 }
987
988 pub fn activities(
989 &self,
990 ) -> Vec<(
991 String,
992 chrono::DateTime<chrono::Utc>,
993 crate::types::ActivityRecord,
994 )> {
995 self.chain_to_root()
996 .into_iter()
997 .filter_map(|entry| match &entry.payload {
998 EntryPayload::Activity(activity) => {
999 Some((entry.id.clone(), entry.ts, activity.clone()))
1000 }
1001 _ => None,
1002 })
1003 .collect()
1004 }
1005
1006 /// Successful tool-result receipts with the ledger timestamp at which
1007 /// the result was recorded. This is a replay-safe source for evidence
1008 /// freshness; callers choose the domain-specific validity window.
1009 pub fn successful_tool_receipts(&self) -> Vec<(String, chrono::DateTime<chrono::Utc>)> {
1010 let mut known_calls = std::collections::HashSet::new();
1011 let mut receipts = Vec::new();
1012 for entry in self.chain_to_root() {
1013 if let EntryPayload::Message(record) = &entry.payload {
1014 for block in &record.message.content {
1015 match block {
1016 vak_llm::ContentBlock::ToolUse { id, .. } => {
1017 known_calls.insert(id.clone());
1018 }
1019 vak_llm::ContentBlock::ToolResult {
1020 tool_use_id,
1021 is_error: false,
1022 ..
1023 } if known_calls.contains(tool_use_id) => {
1024 receipts.push((tool_use_id.clone(), entry.ts));
1025 }
1026 _ => {}
1027 }
1028 }
1029 }
1030 }
1031 receipts
1032 }
1033
1034 /// Successful receipts on the active, latest user turn only. Earlier
1035 /// turns are deliberately excluded so a stale unrelated command cannot
1036 /// establish evidence for the current result.
1037 pub fn successful_tool_receipts_for_latest_turn(
1038 &self,
1039 ) -> Vec<(String, chrono::DateTime<chrono::Utc>)> {
1040 let mut known_calls = std::collections::HashSet::new();
1041 let mut receipts = Vec::new();
1042 for entry in self.chain_to_root() {
1043 if let EntryPayload::Message(record) = &entry.payload {
1044 if record.message.role == vak_llm::Role::User
1045 && record.control_kind().is_none()
1046 && record
1047 .message
1048 .content
1049 .iter()
1050 .any(|block| matches!(block, vak_llm::ContentBlock::Text { .. }))
1051 {
1052 known_calls.clear();
1053 receipts.clear();
1054 }
1055 for block in &record.message.content {
1056 match block {
1057 vak_llm::ContentBlock::ToolUse { id, .. } => {
1058 known_calls.insert(id.clone());
1059 }
1060 vak_llm::ContentBlock::ToolResult {
1061 tool_use_id,
1062 is_error: false,
1063 ..
1064 } if known_calls.contains(tool_use_id) => {
1065 receipts.push((tool_use_id.clone(), entry.ts));
1066 }
1067 _ => {}
1068 }
1069 }
1070 }
1071 }
1072 receipts
1073 }
1074
1075 /// Distinct bash commands that ran GREEN on the active chain, in
1076 /// first-run order (docs/design/10-flows.md adoption substrate). A command is
1077 /// settled when its tool_result is not an error.
1078 pub fn settled_bash_commands(&self) -> Vec<String> {
1079 use std::collections::HashMap;
1080 // id -> is_error for tool results
1081 let mut results: HashMap<String, bool> = HashMap::new();
1082 for e in self.chain_to_root() {
1083 if let EntryPayload::Message(r) = &e.payload {
1084 for b in &r.message.content {
1085 if let vak_llm::ContentBlock::ToolResult {
1086 tool_use_id: id,
1087 is_error,
1088 ..
1089 } = b
1090 {
1091 results.insert(id.clone(), *is_error);
1092 }
1093 }
1094 }
1095 }
1096 let mut out: Vec<String> = Vec::new();
1097 for e in self.chain_to_root() {
1098 if let EntryPayload::Message(r) = &e.payload {
1099 for b in &r.message.content {
1100 if let vak_llm::ContentBlock::ToolUse { name, input, id } = b
1101 && name == "bash"
1102 && !results.get(id).copied().unwrap_or(true)
1103 && let Some(cmd) = input.get("command").and_then(|v| v.as_str())
1104 && !out.iter().any(|o| o == cmd)
1105 {
1106 out.push(cmd.to_string());
1107 }
1108 }
1109 }
1110 }
1111 out
1112 }
1113
1114 /// Receipt entries along the active path, root→leaf — forensic view
1115 /// for surfaces that render dispatch history.
1116 pub fn receipts(&self) -> Vec<&vak_llm::WorkReceipt> {
1117 self.chain_to_root()
1118 .into_iter()
1119 .filter_map(|e| match &e.payload {
1120 EntryPayload::Receipt(r) => Some(r),
1121 _ => None,
1122 })
1123 .collect()
1124 }
1125
1126 pub fn branch_at(&mut self, entry_id: &str) -> Result<(), SessionError> {
1127 if !self.by_id.contains_key(entry_id) {
1128 return Err(SessionError::Corrupt {
1129 line: 0,
1130 message: format!("cannot branch at unknown entry {entry_id}"),
1131 });
1132 }
1133 self.tail_id = Some(entry_id.to_string());
1134 Ok(())
1135 }
1136
1137 /// Appends one compaction packet over the inclusive turn range
1138 /// `first_turn_id..=last_turn_id` (docs/design/68-context-engine.md §4).
1139 /// Both ids must be directive entries on the chain. The packet is a
1140 /// cache keyed by that range: it never moves a boundary and never hides
1141 /// the turns it covers from a plan that wants them at `Full` or `Card`.
1142 pub fn append_packet(
1143 &mut self,
1144 first_turn_id: &str,
1145 last_turn_id: &str,
1146 model: &str,
1147 summary: String,
1148 tokens_before: u64,
1149 ) -> Result<Entry, SessionError> {
1150 for id in [first_turn_id, last_turn_id] {
1151 if !self.by_id.contains_key(id) {
1152 return Err(SessionError::Corrupt {
1153 line: 0,
1154 message: format!("unknown turn id {id} in packet range"),
1155 });
1156 }
1157 }
1158 let parent = self.tail_id.clone();
1159 self.append(Entry::new(
1160 parent,
1161 EntryPayload::Compaction(crate::types::CompactionEntry {
1162 summary,
1163 first_turn_id: first_turn_id.to_string(),
1164 last_turn_id: last_turn_id.to_string(),
1165 model: model.to_string(),
1166 tokens_before,
1167 reset_all: false,
1168 }),
1169 ))
1170 }
1171
1172 /// Reset-with-handoff (docs/design/42-managed-work-contracts.md): the projection becomes ONLY
1173 /// this summary. Append-only; the full history stays on disk. This is
1174 /// the one compaction entry that is a real boundary — the rescue for a
1175 /// profile with no usable horizon, where nothing is plannable.
1176 pub fn append_handoff_reset(
1177 &mut self,
1178 summary: String,
1179 tokens_before: u64,
1180 ) -> Result<Entry, SessionError> {
1181 let parent = self.tail_id.clone();
1182 self.append(Entry::new(
1183 parent,
1184 EntryPayload::Compaction(crate::types::CompactionEntry {
1185 summary,
1186 first_turn_id: String::new(),
1187 last_turn_id: String::new(),
1188 model: String::new(),
1189 tokens_before,
1190 reset_all: true,
1191 }),
1192 ))
1193 }
1194
1195 pub fn header(&self) -> Option<&SessionHeader> {
1196 self.entries.iter().find_map(|e| match &e.payload {
1197 EntryPayload::Header(h) => Some(h),
1198 _ => None,
1199 })
1200 }
1201
1202 pub fn len(&self) -> usize {
1203 self.entries.len()
1204 }
1205
1206 pub fn is_empty(&self) -> bool {
1207 self.entries.is_empty()
1208 }
1209
1210 pub fn path(&self) -> &Path {
1211 &self.path
1212 }
1213
1214 pub fn tail_id(&self) -> Option<&String> {
1215 self.tail_id.as_ref()
1216 }
1217
1218 pub fn chain_to_root(&self) -> Vec<&Entry> {
1219 let mut chain = Vec::new();
1220 let mut cursor = self.tail_id.clone();
1221 while let Some(id) = cursor {
1222 let Some(&idx) = self.by_id.get(&id) else {
1223 break;
1224 };
1225 let entry = &self.entries[idx];
1226 chain.push(entry);
1227 cursor = entry.parent_id.clone();
1228 }
1229 chain.reverse();
1230 chain
1231 }
1232
1233 /// Message entries along the active path, root→leaf, with their entry
1234 /// ids — raw ledger view (compaction entries NOT applied).
1235 pub fn message_chain(&self) -> Vec<(String, Message)> {
1236 self.chain_to_root()
1237 .into_iter()
1238 .filter_map(|e| match &e.payload {
1239 EntryPayload::Message(r) => Some((e.id.clone(), r.message.clone())),
1240 _ => None,
1241 })
1242 .collect()
1243 }
1244
1245 fn derive_keyed(&self) -> Vec<(String, Message)> {
1246 self.derive_keyed_tagged()
1247 .into_iter()
1248 .map(|(id, m, _, _)| (id, m))
1249 .collect()
1250 }
1251
1252 /// The plan-free projection: every closed turn at `Full` fidelity. Used
1253 /// by `derive_messages` (goal audits, search, human-facing views), which
1254 /// has no `CapacityProfile` to plan against.
1255 fn derive_keyed_tagged(&self) -> Vec<(String, Message, bool, bool)> {
1256 self.derive_with_plan_tagged(None)
1257 }
1258
1259 /// Like `derive_keyed_tagged`, but a `WorkingSetPlan` (from
1260 /// `vak_context::planner::plan`) selects each closed turn's fidelity
1261 /// instead of defaulting every one to `Full` (docs/design/68-context-
1262 /// engine.md §4/§10) — one implementation, the plan just picks what each
1263 /// turn contributes:
1264 ///
1265 /// - `Full` turns project their `full_record`: real `tool_use` blocks,
1266 /// results as schema-driven digests carrying their evidence id, never
1267 /// a character-count trim.
1268 /// - `Card` turns contribute one line each to a single `<turns>` block
1269 /// (`TurnCard::line`), inserted once, right after any compaction
1270 /// summary.
1271 /// - `Packet` turns, and any closed turn the plan omits entirely,
1272 /// contribute nothing here: they are represented only by an existing
1273 /// `Compaction` entry, which the caller (`Agent`'s incremental
1274 /// compaction, §4) guarantees already covers them before this is
1275 /// called with that plan.
1276 /// - The still-open turn (if any) always projects verbatim, regardless
1277 /// of `plan` — it is never planned.
1278 ///
1279 /// The intent note, work contract, and conversation thread are rendered
1280 /// into the request tail instead (§6/§10), read separately via
1281 /// [`SessionLog::tail_sections`]; this function never contributes them.
1282 /// The reset boundary: the chain position before which everything is
1283 /// invisible to the model, and the handoff summary that stands in for
1284 /// it. Only a `reset_all` compaction entry (reset-with-handoff,
1285 /// docs/design/42) moves this; packets never do. `(0, ..., None)` when
1286 /// no reset has happened.
1287 fn reset_boundary(&self) -> (usize, HashMap<String, usize>, Option<String>) {
1288 let chain = self.chain_to_root();
1289 let position: HashMap<String, usize> = chain
1290 .iter()
1291 .enumerate()
1292 .map(|(i, entry)| (entry.id.clone(), i))
1293 .collect();
1294 let last_reset =
1295 chain
1296 .iter()
1297 .enumerate()
1298 .rev()
1299 .find_map(|(pos, entry)| match &entry.payload {
1300 EntryPayload::Compaction(c) if c.reset_all => Some((pos, c.summary.clone())),
1301 _ => None,
1302 });
1303 match last_reset {
1304 Some((pos, summary)) => (pos, position, Some(summary)),
1305 None => (0, position, None),
1306 }
1307 }
1308
1309 /// Every stored packet (non-reset compaction entry) in ledger order.
1310 fn packets(&self) -> Vec<Packet> {
1311 TurnIndex::from_log(self).packets
1312 }
1313
1314 /// The stored packet whose range is exactly `first_turn_id..=last_turn_id`,
1315 /// newest such entry first (docs/design/68 §4: a packet is reused only
1316 /// for the exact range the plan asks for). `None` when no packet
1317 /// covers that range, whatever other packets exist.
1318 pub fn packet_for(&self, first_turn_id: &str, last_turn_id: &str) -> Option<Packet> {
1319 self.packets().into_iter().rev().find(|packet| {
1320 packet.first_turn_id == first_turn_id && packet.last_turn_id == last_turn_id
1321 })
1322 }
1323
1324 /// Whether the packet range a `WorkingSetPlan` asked for
1325 /// (`packet_range = (first, last)`) still needs a summariser call: true
1326 /// when no stored packet covers exactly that range.
1327 pub fn packet_needs_compaction(&self, first_turn_id: &str, last_turn_id: &str) -> bool {
1328 self.packet_for(first_turn_id, last_turn_id).is_none()
1329 }
1330
1331 /// The summariser input for a packet over `first_turn_id..=last_turn_id`:
1332 /// the longest stored packet that starts at the same turn and ends at or
1333 /// before `last_turn_id` (so work already folded in is reused, never
1334 /// re-read from raw history), followed by one `TurnCard` line per turn
1335 /// after it up to `last_turn_id` — cards, never raw history (§4). Also
1336 /// returns a char-count estimate for the caller's `tokens_before`.
1337 pub fn packet_transcript(&self, first_turn_id: &str, last_turn_id: &str) -> (String, u64) {
1338 let (_, position, _) = self.reset_boundary();
1339 let mut out = String::new();
1340 let (Some(&first_pos), Some(&last_pos)) =
1341 (position.get(first_turn_id), position.get(last_turn_id))
1342 else {
1343 return (out, 0);
1344 };
1345 // Seed: the stored packet with this `first` and the greatest `last`
1346 // not past our target.
1347 let seed = self
1348 .packets()
1349 .into_iter()
1350 .filter(|packet| packet.first_turn_id == first_turn_id)
1351 .filter_map(|packet| {
1352 let end = position.get(packet.last_turn_id.as_str()).copied()?;
1353 (end <= last_pos).then_some((end, packet))
1354 })
1355 .max_by_key(|(end, _)| *end);
1356 let mut from_pos = first_pos;
1357 if let Some((end, packet)) = seed {
1358 out.push_str(&packet.summary);
1359 out.push_str("\n\n");
1360 from_pos = end + 1;
1361 }
1362 let index = TurnIndex::from_log(self);
1363 for (turn_number, turn) in index.turns.iter().enumerate() {
1364 let turn_pos = position.get(turn.id.as_str()).copied().unwrap_or(0);
1365 if turn_pos < from_pos || turn_pos > last_pos {
1366 continue;
1367 }
1368 match &turn.card {
1369 Some(card) => {
1370 out.push_str(&card.line(turn_number + 1));
1371 out.push('\n');
1372 }
1373 None => {
1374 for message in turn.full_record() {
1375 out.push_str(&message.text_content());
1376 out.push('\n');
1377 }
1378 }
1379 }
1380 }
1381 let chars = out.chars().count() as u64;
1382 (out, chars)
1383 }
1384
1385 /// The model-visible projection for one request. With a plan, each
1386 /// closed turn contributes exactly what the plan chose for it; without
1387 /// one, every closed turn rides at `Full` (the plan-free view used by

388 serde_json::from_str(activity.data.get("key")?).ok()?;
389 if stored_key != key_json {
390 return None;
391 }
392 serde_json::from_str(activity.data.get("profile")?).ok()
393 })
394 }
395
396 /// Appends a turn's closing card (docs/design/68-context-engine.md §10).
397 /// Written once, at turn close, and never rewritten.
398 pub fn append_turn_card(&mut self, record: TurnCardRecord) -> Result<Entry, SessionError> {
399 let parent = self.tail_id.clone();
400 self.append(Entry::new(parent, EntryPayload::TurnCard(record)))
401 }
402
403 /// `(turn_id, card)` for every `TurnCard` entry along the active chain,
404 /// in the order they were written.
405 pub fn turn_cards(&self) -> Vec<(String, TurnCard)> {
406 self.chain_to_root()
407 .into_iter()
408 .filter_map(|entry| match &entry.payload {
409 EntryPayload::TurnCard(record) => {
410 Some((record.turn_id.clone(), record.card.clone()))
411 }
412 _ => None,
413 })
414 .collect()
415 }
416
417 /// Resolves a tool_use_id to its full `Evidence` — the tool name, its
418 /// call arguments, its whole result content (the evidence body when the
419 /// request carried a window of it), and whether it failed — by
420 /// scanning the active chain for the matching `ToolUse`/`ToolResult`
421 /// pair. Works for evidence in any turn, open or closed, which is what
422 /// lets `recall({ id })` reopen a result from a turn now reduced to a
423 /// card (docs/design/68-context-engine.md §3).
424 pub fn evidence(&self, tool_use_id: &str) -> Option<Evidence> {
425 let chain = self.chain_to_root();
426 let mut tool: Option<String> = None;
427 let mut input = serde_json::Value::Null;
428 for entry in &chain {
429 let EntryPayload::Message(record) = &entry.payload else {
430 continue;
431 };
432 for block in &record.message.content {
433 if let vak_llm::ContentBlock::ToolUse {
434 id,
435 name,
436 input: call_input,
437 } = block
438 && id == tool_use_id
439 {
440 tool = Some(name.clone());
441 input = call_input.clone();
442 }
443 }
444 }
445 let tool = tool?;
446 let body = chain.iter().find_map(|entry| match &entry.payload {
447 EntryPayload::EvidenceBody(body) if body.tool_use_id == tool_use_id => {
448 Some(body.content.clone())
449 }
450 _ => None,
451 });
452 for entry in &chain {
453 let EntryPayload::Message(record) = &entry.payload else {
454 continue;
455 };
456 for block in &record.message.content {
457 if let vak_llm::ContentBlock::ToolResult {
458 tool_use_id: id,
459 content,
460 is_error,
461 } = block
462 && id == tool_use_id
463 {
464 return Some(Evidence {
465 tool,
466 input,
467 content: body.unwrap_or_else(|| content.clone()),
468 is_error: *is_error,
469 });
470 }
471 }
472 }
473 None
474 }
475
476 /// Presentation entries along the active path, root→leaf, with their
477 /// entry ids — the single source both the model-visible history and the
478 /// display channel read.
479 pub fn presentations(&self) -> Vec<(String, &PresentationRecord)> {
480 self.chain_to_root()
481 .into_iter()
482 .filter_map(|e| match &e.payload {
483 EntryPayload::Presentation(record) => Some((e.id.clone(), record)),
484 _ => None,
485 })
486 .collect()
487 }
488
489 /// Whether a presentation with this exact canonical payload already
490 /// exists for this turn — the duplicate rule that drops a repeated
491 /// fence from the model-visible projection (docs/design/68-context-engine.md
492 /// §10: "a second card in the same turn with the same `payload_digest`
493 /// is not written").
494 pub fn has_presentation(&self, turn_id: &str, payload_digest: &str) -> bool {
495 self.presentations()
496 .iter()
497 .any(|(_, record)| record.turn_id == turn_id && record.payload_digest == payload_digest)
498 }
499
500 /// The id of the most recent non-control user message entry in the
501 /// active chain — the directive the current turn answers. `None` before
502 /// any real user message exists.
503 pub fn latest_directive_entry_id(&self) -> Option<String> {
504 self.chain_to_root()
505 .into_iter()
506 .rev()
507 .find_map(|entry| match &entry.payload {
508 EntryPayload::Message(record)
509 if record.message.role == vak_llm::Role::User
510 && record.control_kind().is_none()
511 && record
512 .message
513 .content
514 .iter()
515 .any(|b| matches!(b, vak_llm::ContentBlock::Text { .. }))
516 && !record
517 .message
518 .content
519 .iter()
520 .any(|b| matches!(b, vak_llm::ContentBlock::ToolResult { .. })) =>
521 {
522 Some(entry.id.clone())
523 }
524 _ => None,
525 })
526 }
527
528 /// Ids of every non-card tool result already committed to the ledger
529 /// strictly after `turn_id`, in chain order — the evidence a card built
530 /// after them was derived from (`PresentationRecord::derived_from`).
531 /// `is_card_tool` is supplied by the caller so this crate never needs to
532 /// know what a "card" tool is (that knowledge lives in
533 /// `vak-core::presentation_tools`).
534 pub fn non_card_evidence_since(
535 &self,
536 turn_id: &str,
537 is_card_tool: impl Fn(&str) -> bool,
538 ) -> Vec<String> {
539 let chain = self.chain_to_root();
540 let start = chain
541 .iter()
542 .position(|entry| entry.id == turn_id)
543 .map(|idx| idx + 1)
544 .unwrap_or(0);
545 let mut tool_names: HashMap<String, String> = HashMap::new();
546 let mut out = Vec::new();
547 for entry in &chain[start..] {
548 let EntryPayload::Message(record) = &entry.payload else {
549 continue;
550 };
551 for block in &record.message.content {
552 match block {
553 vak_llm::ContentBlock::ToolUse { id, name, .. } => {
554 tool_names.insert(id.clone(), name.clone());
555 }
556 vak_llm::ContentBlock::ToolResult { tool_use_id, .. } => {
557 let is_card = tool_names
558 .get(tool_use_id)
559 .map(|name| is_card_tool(name))
560 .unwrap_or(false);
561 if !is_card && !out.iter().any(|seen| seen == tool_use_id) {
562 out.push(tool_use_id.clone());
563 }
564 }
565 _ => {}
566 }
567 }
568 }
569 out
570 }
571
572 /// Append a finalized or provisional voice transcript. The transcript
573 /// is an audit projection; callers must append a normal Message entry
574 /// separately when the utterance is committed as model input.
575 pub fn append_voice_transcript(
576 &mut self,
577 activity_id: impl Into<String>,
578 text: impl Into<String>,
579 finalized: bool,
580 ) -> Result<Entry, SessionError> {
581 let mut data = std::collections::BTreeMap::new();
582 data.insert("text".into(), text.into());
583 data.insert("finalized".into(), finalized.to_string());
584 self.append_activity(crate::types::ActivityRecord {
585 activity_id: activity_id.into(),
586 turn: None,
587 kind: crate::types::ActivityKind::VoiceTranscript,
588 status: if finalized {
589 crate::types::ActivityStatus::Succeeded
590 } else {
591 crate::types::ActivityStatus::Running
592 },
593 label: "Voice transcript".into(),
594 detail: None,
595 data,
596 })
597 }
598
599 /// Append an auditable presentation selection without making it part of
600 /// model context. The original result and fallback remain authoritative.
601 pub fn append_presentation_selection(
602 &mut self,
603 activity_id: impl Into<String>,
604 semantic_type: impl Into<String>,
605 spec_id: impl Into<String>,
606 revision: u64,
607 mode: impl Into<String>,
608 fallback_used: bool,
609 ) -> Result<Entry, SessionError> {
610 let mut data = std::collections::BTreeMap::new();
611 data.insert("semantic_type".into(), semantic_type.into());
612 data.insert("spec_id".into(), spec_id.into());
613 data.insert("revision".into(), revision.to_string());
614 data.insert("mode".into(), mode.into());
615 data.insert("fallback_used".into(), fallback_used.to_string());
616 self.append_activity(crate::types::ActivityRecord {
617 activity_id: activity_id.into(),
618 turn: None,
619 kind: crate::types::ActivityKind::PresentationSelection,
620 status: crate::types::ActivityStatus::Succeeded,
621 label: "Presentation selected".into(),
622 detail: None,
623 data,
624 })
625 }
626
627 /// Append a projection choice or correction. If the text is sent to the
628 /// model, callers must also append a normal Message entry so derivation
629 /// remains complete.
630 pub fn append_presentation_feedback(
631 &mut self,
632 activity_id: impl Into<String>,
633 choice: impl Into<String>,
634 feedback: Option<String>,
635 ) -> Result<Entry, SessionError> {
636 let mut data = std::collections::BTreeMap::new();
637 data.insert("choice".into(), choice.into());
638 if let Some(feedback) = feedback {
639 data.insert("feedback".into(), feedback);
640 }
641 self.append_activity(crate::types::ActivityRecord {
642 activity_id: activity_id.into(),
643 turn: None,
644 kind: crate::types::ActivityKind::PresentationFeedback,
645 status: crate::types::ActivityStatus::Succeeded,
646 label: "Presentation feedback".into(),
647 detail: None,
648 data,
649 })
650 }
651
652 /// Append what the playback client reports it emitted. This is distinct
653 /// from provider output because buffering and interruption can prevent
654 /// the user from hearing the complete synthesis.
655 pub fn append_voice_playback(
656 &mut self,
657 activity_id: impl Into<String>,
658 emitted_ms: u64,
659 interrupted: bool,
660 ) -> Result<Entry, SessionError> {
661 let mut data = std::collections::BTreeMap::new();
662 data.insert("emitted_ms".into(), emitted_ms.to_string());
663 data.insert("interrupted".into(), interrupted.to_string());
664 self.append_activity(crate::types::ActivityRecord {
665 activity_id: activity_id.into(),
666 turn: None,
667 kind: crate::types::ActivityKind::VoicePlayback,
668 status: if interrupted {
669 crate::types::ActivityStatus::Partial
670 } else {
671 crate::types::ActivityStatus::Succeeded
672 },
673 label: "Voice playback".into(),
674 detail: None,
675 data,
676 })
677 }
678
679 /// Record this turn's resolved intent.
680 ///
681 /// Appended before the turn dispatches, so the note it carries is in the
682 /// projection the model actually sees — the entry *is* the record of what
683 /// was said, not a description of it (invariant 1).
684 pub fn append_intent(
685 &mut self,
686 record: crate::types::IntentRecord,
687 ) -> Result<Entry, SessionError> {
688 let parent = self.tail_id.clone();
689 self.append(Entry::new(parent, EntryPayload::Intent(Box::new(record))))
690 }
691
692 pub fn append_child_run_status(
693 &mut self,
694 status: crate::types::ChildRunStatus,
695 ) -> Result<Entry, SessionError> {
696 self.append_child_run_result(status, None)
697 }
698
699 pub fn append_child_run_result(
700 &mut self,
701 status: crate::types::ChildRunStatus,
702 outcome: Option<vak_intent::OutcomeSpec>,
703 ) -> Result<Entry, SessionError> {
704 let parent = self.tail_id.clone();
705 self.append(Entry::new(
706 parent,
707 EntryPayload::ChildRun { status, outcome },
708 ))
709 }
710
711 pub fn child_run_status(&self) -> Option<crate::types::ChildRunStatus> {
712 self.chain_to_root()
713 .iter()
714 .rev()
715 .find_map(|entry| match &entry.payload {
716 EntryPayload::ChildRun { status, .. } => Some(status.clone()),
717 _ => None,
718 })
719 }
720
721 pub fn child_run_outcome(&self) -> Option<vak_intent::OutcomeSpec> {
722 self.chain_to_root()
723 .iter()
724 .rev()
725 .find_map(|entry| match &entry.payload {
726 EntryPayload::ChildRun { outcome, .. } => outcome.clone(),
727 _ => None,
728 })
729 }
730
731 /// Records the interface a turn was bound to: in full when it differs
732 /// from the last one this conversation bound, otherwise as a
733 /// `TurnCapabilitiesRef` to that entry.
734 pub fn append_turn_capabilities(
735 &mut self,
736 bound: crate::types::TurnCapabilitiesBound,
737 ) -> Result<Entry, SessionError> {
738 let digest = bound.digest();
739 let previous =
740 self.chain_to_root()
741 .into_iter()
742 .rev()
743 .find_map(|entry| match &entry.payload {
744 EntryPayload::TurnCapabilitiesBound(earlier) => {
745 Some((entry.id.clone(), earlier.digest()))
746 }
747 _ => None,
748 });
749 let parent = self.tail_id.clone();
750 let payload = match previous {
751 Some((entry, earlier)) if earlier == digest => {
752 EntryPayload::TurnCapabilitiesRef(crate::types::TurnCapabilitiesRef {
753 entry,
754 digest,
755 epoch: bound.epoch,
756 })
757 }
758 _ => EntryPayload::TurnCapabilitiesBound(bound),
759 };
760 self.append(Entry::new(parent, payload))
761 }
762
763 pub fn append_work(&mut self, event: WorkEvent) -> Result<Entry, SessionError> {
764 let parent = self.tail_id.clone();
765 let candidate = Entry::new(parent, EntryPayload::Work(event));
766 let mut chain = self.chain_to_root();
767 chain.push(&candidate);
768 crate::work::project_work(&chain).map_err(|error| SessionError::Corrupt {
769 line: 0,
770 message: format!("invalid work event: {error}"),
771 })?;
772 let appended = self.append(candidate)?;
773 self.promote_ready_work_items()?;
774 Ok(appended)
775 }
776
777 fn promote_ready_work_items(&mut self) -> Result<(), SessionError> {
778 let Some(projection) = self
779 .work_projection()
780 .map_err(|error| SessionError::Corrupt {
781 line: 0,
782 message: error.to_string(),
783 })?
784 else {
785 return Ok(());
786 };
787 if projection.status != crate::types::WorkContractStatus::Active {
788 return Ok(());
789 }
790 let ready: Vec<String> = projection
791 .contract
792 .items
793 .iter()
794 .filter(|definition| {
795 projection
796 .items
797 .get(&definition.item_id)
798 .is_some_and(|state| state.status == crate::types::WorkItemStatus::Proposed)
799 && definition.dependencies.iter().all(|dependency| {
800 projection.items.get(dependency).is_some_and(|state| {
801 matches!(
802 state.status,
803 crate::types::WorkItemStatus::Succeeded
804 | crate::types::WorkItemStatus::Skipped
805 )
806 })
807 })
808 })
809 .map(|definition| definition.item_id.clone())
810 .collect();
811 for item_id in ready {
812 let current = self
813 .work_projection()
814 .map_err(|error| SessionError::Corrupt {
815 line: 0,
816 message: error.to_string(),
817 })?;
818 let Some(current) = current else { break };
819 self.append_work(crate::types::WorkEvent {
820 contract_id: current.contract.contract_id,
821 revision: current.contract.revision,
822 kind: crate::types::WorkEventKind::ItemStatusChanged {
823 item_id,
824 from: crate::types::WorkItemStatus::Proposed,
825 to: crate::types::WorkItemStatus::Ready,
826 attempt: 0,
827 reason: "dependencies satisfied".into(),
828 },
829 })?;
830 }
831 Ok(())
832 }
833
834 pub fn work_projection(
835 &self,
836 ) -> Result<Option<crate::work::WorkProjection>, crate::work::WorkError> {
837 crate::work::project_work(&self.chain_to_root())
838 }
839
840 pub fn evidence_exists(&self, evidence: &crate::types::EvidenceRef) -> bool {
841 let session_id = self.header().map(|header| header.session_id.as_str());
842 self.chain_to_root().iter().any(|entry| match evidence {
843 crate::types::EvidenceRef::LedgerEntry {
844 session_id: evidence_session,
845 entry_id,
846 }
847 | crate::types::EvidenceRef::Receipt {
848 session_id: evidence_session,
849 entry_id,
850 } => session_id == Some(evidence_session.as_str()) && entry.id == *entry_id,
851 crate::types::EvidenceRef::ToolResult {
852 session_id: evidence_session,
853 tool_use_id,
854 } => {
855 session_id == Some(evidence_session.as_str())
856 && matches!(
857 &entry.payload,
858 crate::types::EntryPayload::Message(record)
859 if record.message.content.iter().any(|block| matches!(
860 block,
861 vak_llm::ContentBlock::ToolResult { tool_use_id: id, .. }
862 if id == tool_use_id
863 ))
864 )
865 }
866 crate::types::EvidenceRef::CheckpointDiff { .. }
867 | crate::types::EvidenceRef::FlowNode { .. }
868 | crate::types::EvidenceRef::ChildSession { .. }
869 | crate::types::EvidenceRef::ExternalOperation { .. } => false,
870 })
871 }
872
873 /// Reconcile managed items left running by a process restart. Only child
874 /// sessions explicitly reported as live may remain running; every other
875 /// running item is conservatively interrupted and made retry-eligible.
876 /// Possible side effects are never replayed automatically.
877 pub fn reconcile_running_work(
878 &mut self,
879 live_child_sessions: &std::collections::HashSet<String>,
880 ) -> Result<usize, SessionError> {
881 self.reconcile_running_work_with_child_ledgers(live_child_sessions, None)
882 }
883
884 pub fn reconcile_running_work_with_child_ledgers(
885 &mut self,
886 live_child_sessions: &std::collections::HashSet<String>,
887 sessions_home: Option<&Path>,
888 ) -> Result<usize, SessionError> {
889 let Some(projection) = self
890 .work_projection()
891 .map_err(|error| SessionError::Corrupt {
892 line: 0,
893 message: error.to_string(),
894 })?
895 else {
896 return Ok(0);
897 };
898 let stale: Vec<(String, u32, Option<String>)> = projection
899 .items
900 .values()
901 .filter(|item| {
902 item.status == crate::types::WorkItemStatus::Running
903 && !item
904 .child_session_id
905 .as_ref()
906 .is_some_and(|id| live_child_sessions.contains(id))
907 })
908 .map(|item| {
909 (
910 item.item_id.clone(),
911 item.attempt,
912 item.child_session_id.clone(),
913 )
914 })
915 .collect();
916 let mut reconciled = 0;
917 for (item_id, attempt, child_id) in stale {
918 let Some(current) = self
919 .work_projection()
920 .map_err(|error| SessionError::Corrupt {
921 line: 0,
922 message: error.to_string(),
923 })?
924 else {
925 break;
926 };
927 let child_status = child_id.as_deref().and_then(|id| {
928 let home = sessions_home?;
929 let cwd = self.header().map(|h| h.contract_cwd())?;
930 let path = SessionPath::new_session_file(home, &cwd, id);
931 SessionLog::open(path)
932 .ok()
933 .and_then(|child| child.child_run_status())
934 });
935 if matches!(child_status, Some(crate::types::ChildRunStatus::Completed)) {
936 self.append_work(WorkEvent {
937 contract_id: current.contract.contract_id.clone(),
938 revision: current.contract.revision,
939 kind: crate::types::WorkEventKind::EvidenceAttached {
940 item_id: item_id.clone(),
941 evidence: crate::types::EvidenceRef::ChildSession {
942 session_id: child_id.clone().unwrap_or_default(),
943 },
944 },
945 })?;
946 self.append_work(WorkEvent {
947 contract_id: current.contract.contract_id.clone(),
948 revision: current.contract.revision,
949 kind: crate::types::WorkEventKind::ItemStatusChanged {
950 item_id,
951 from: crate::types::WorkItemStatus::Running,
952 to: crate::types::WorkItemStatus::ReadyForVerification,
953 attempt,
954 reason: "recovered completed child; verify its durable evidence".into(),
955 },
956 })?;
957 reconciled += 1;
958 continue;
959 }
960 self.append_work(WorkEvent {
961 contract_id: current.contract.contract_id.clone(),
962 revision: current.contract.revision,
963 kind: crate::types::WorkEventKind::ItemStatusChanged {
964 item_id,
965 from: crate::types::WorkItemStatus::Running,
966 to: crate::types::WorkItemStatus::Interrupted,
967 attempt,
968 reason: match child_status {
969 Some(crate::types::ChildRunStatus::Failed) => {
970 "child failed before restart; review before retry"
971 }
972 Some(crate::types::ChildRunStatus::Aborted) => {
973 "child was aborted before restart; review before retry"
974 }
975 Some(crate::types::ChildRunStatus::MaxTurns) => {
976 "child hit its turn limit; review before retry"
977 }
978 _ => "recovered after process restart; review before retry",
979 }
980 .into(),
981 },
982 })?;
983 reconciled += 1;
984 }
985 Ok(reconciled)
986 }
987
988 pub fn activities(
989 &self,
990 ) -> Vec<(
991 String,
992 chrono::DateTime<chrono::Utc>,
993 crate::types::ActivityRecord,
994 )> {
995 self.chain_to_root()
996 .into_iter()
997 .filter_map(|entry| match &entry.payload {
998 EntryPayload::Activity(activity) => {
999 Some((entry.id.clone(), entry.ts, activity.clone()))
1000 }
1001 _ => None,
1002 })
1003 .collect()
1004 }
1005
1006 /// Successful tool-result receipts with the ledger timestamp at which
1007 /// the result was recorded. This is a replay-safe source for evidence
1008 /// freshness; callers choose the domain-specific validity window.
1009 pub fn successful_tool_receipts(&self) -> Vec<(String, chrono::DateTime<chrono::Utc>)> {
1010 let mut known_calls = std::collections::HashSet::new();
1011 let mut receipts = Vec::new();
1012 for entry in self.chain_to_root() {
1013 if let EntryPayload::Message(record) = &entry.payload {
1014 for block in &record.message.content {
1015 match block {
1016 vak_llm::ContentBlock::ToolUse { id, .. } => {
1017 known_calls.insert(id.clone());
1018 }
1019 vak_llm::ContentBlock::ToolResult {
1020 tool_use_id,
1021 is_error: false,
1022 ..
1023 } if known_calls.contains(tool_use_id) => {
1024 receipts.push((tool_use_id.clone(), entry.ts));
1025 }
1026 _ => {}
1027 }
1028 }
1029 }
1030 }
1031 receipts
1032 }
1033
1034 /// Successful receipts on the active, latest user turn only. Earlier
1035 /// turns are deliberately excluded so a stale unrelated command cannot
1036 /// establish evidence for the current result.
1037 pub fn successful_tool_receipts_for_latest_turn(
1038 &self,
1039 ) -> Vec<(String, chrono::DateTime<chrono::Utc>)> {
1040 let mut known_calls = std::collections::HashSet::new();
1041 let mut receipts = Vec::new();
1042 for entry in self.chain_to_root() {
1043 if let EntryPayload::Message(record) = &entry.payload {
1044 if record.message.role == vak_llm::Role::User
1045 && record.control_kind().is_none()
1046 && record
1047 .message
1048 .content
1049 .iter()
1050 .any(|block| matches!(block, vak_llm::ContentBlock::Text { .. }))
1051 {
1052 known_calls.clear();
1053 receipts.clear();
1054 }
1055 for block in &record.message.content {
1056 match block {
1057 vak_llm::ContentBlock::ToolUse { id, .. } => {
1058 known_calls.insert(id.clone());
1059 }
1060 vak_llm::ContentBlock::ToolResult {
1061 tool_use_id,
1062 is_error: false,
1063 ..
1064 } if known_calls.contains(tool_use_id) => {
1065 receipts.push((tool_use_id.clone(), entry.ts));
1066 }
1067 _ => {}
1068 }
1069 }
1070 }
1071 }
1072 receipts
1073 }
1074
1075 /// Distinct bash commands that ran GREEN on the active chain, in
1076 /// first-run order (docs/design/10-flows.md adoption substrate). A command is
1077 /// settled when its tool_result is not an error.
1078 pub fn settled_bash_commands(&self) -> Vec<String> {
1079 use std::collections::HashMap;
1080 // id -> is_error for tool results
1081 let mut results: HashMap<String, bool> = HashMap::new();
1082 for e in self.chain_to_root() {
1083 if let EntryPayload::Message(r) = &e.payload {
1084 for b in &r.message.content {
1085 if let vak_llm::ContentBlock::ToolResult {
1086 tool_use_id: id,
1087 is_error,
1088 ..
1089 } = b
1090 {
1091 results.insert(id.clone(), *is_error);
1092 }
1093 }
1094 }
1095 }
1096 let mut out: Vec<String> = Vec::new();
1097 for e in self.chain_to_root() {
1098 if let EntryPayload::Message(r) = &e.payload {
1099 for b in &r.message.content {
1100 if let vak_llm::ContentBlock::ToolUse { name, input, id } = b
1101 && name == "bash"
1102 && !results.get(id).copied().unwrap_or(true)
1103 && let Some(cmd) = input.get("command").and_then(|v| v.as_str())
1104 && !out.iter().any(|o| o == cmd)
1105 {
1106 out.push(cmd.to_string());
1107 }
1108 }
1109 }
1110 }
1111 out
1112 }
1113
1114 /// Receipt entries along the active path, root→leaf — forensic view
1115 /// for surfaces that render dispatch history.
1116 pub fn receipts(&self) -> Vec<&vak_llm::WorkReceipt> {
1117 self.chain_to_root()
1118 .into_iter()
1119 .filter_map(|e| match &e.payload {
1120 EntryPayload::Receipt(r) => Some(r),
1121 _ => None,
1122 })
1123 .collect()
1124 }
1125
1126 pub fn branch_at(&mut self, entry_id: &str) -> Result<(), SessionError> {
1127 if !self.by_id.contains_key(entry_id) {
1128 return Err(SessionError::Corrupt {
1129 line: 0,
1130 message: format!("cannot branch at unknown entry {entry_id}"),
1131 });
1132 }
1133 self.tail_id = Some(entry_id.to_string());
1134 Ok(())
1135 }
1136
1137 /// Appends one compaction packet over the inclusive turn range
1138 /// `first_turn_id..=last_turn_id` (docs/design/68-context-engine.md §4).
1139 /// Both ids must be directive entries on the chain. The packet is a
1140 /// cache keyed by that range: it never moves a boundary and never hides
1141 /// the turns it covers from a plan that wants them at `Full` or `Card`.
1142 pub fn append_packet(
1143 &mut self,
1144 first_turn_id: &str,
1145 last_turn_id: &str,
1146 model: &str,
1147 summary: String,
1148 tokens_before: u64,
1149 ) -> Result<Entry, SessionError> {
1150 for id in [first_turn_id, last_turn_id] {
1151 if !self.by_id.contains_key(id) {
1152 return Err(SessionError::Corrupt {
1153 line: 0,
1154 message: format!("unknown turn id {id} in packet range"),
1155 });
1156 }
1157 }
1158 let parent = self.tail_id.clone();
1159 self.append(Entry::new(
1160 parent,
1161 EntryPayload::Compaction(crate::types::CompactionEntry {
1162 summary,
1163 first_turn_id: first_turn_id.to_string(),
1164 last_turn_id: last_turn_id.to_string(),
1165 model: model.to_string(),
1166 tokens_before,
1167 reset_all: false,
1168 }),
1169 ))
1170 }
1171
1172 /// Reset-with-handoff (docs/design/42-managed-work-contracts.md): the projection becomes ONLY
1173 /// this summary. Append-only; the full history stays on disk. This is
1174 /// the one compaction entry that is a real boundary — the rescue for a
1175 /// profile with no usable horizon, where nothing is plannable.
1176 pub fn append_handoff_reset(
1177 &mut self,
1178 summary: String,
1179 tokens_before: u64,
1180 ) -> Result<Entry, SessionError> {
1181 let parent = self.tail_id.clone();
1182 self.append(Entry::new(
1183 parent,
1184 EntryPayload::Compaction(crate::types::CompactionEntry {
1185 summary,
1186 first_turn_id: String::new(),
1187 last_turn_id: String::new(),
1188 model: String::new(),
1189 tokens_before,
1190 reset_all: true,
1191 }),
1192 ))
1193 }
1194
1195 pub fn header(&self) -> Option<&SessionHeader> {
1196 self.entries.iter().find_map(|e| match &e.payload {
1197 EntryPayload::Header(h) => Some(h),
1198 _ => None,
1199 })
1200 }
1201
1202 pub fn len(&self) -> usize {
1203 self.entries.len()
1204 }
1205
1206 pub fn is_empty(&self) -> bool {
1207 self.entries.is_empty()
1208 }
1209
1210 pub fn path(&self) -> &Path {
1211 &self.path
1212 }
1213
1214 pub fn tail_id(&self) -> Option<&String> {
1215 self.tail_id.as_ref()
1216 }
1217
1218 pub fn chain_to_root(&self) -> Vec<&Entry> {
1219 let mut chain = Vec::new();
1220 let mut cursor = self.tail_id.clone();
1221 while let Some(id) = cursor {
1222 let Some(&idx) = self.by_id.get(&id) else {
1223 break;
1224 };
1225 let entry = &self.entries[idx];
1226 chain.push(entry);
1227 cursor = entry.parent_id.clone();
1228 }
1229 chain.reverse();
1230 chain
1231 }
1232
1233 /// Message entries along the active path, root→leaf, with their entry
1234 /// ids — raw ledger view (compaction entries NOT applied).
1235 pub fn message_chain(&self) -> Vec<(String, Message)> {
1236 self.chain_to_root()
1237 .into_iter()
1238 .filter_map(|e| match &e.payload {
1239 EntryPayload::Message(r) => Some((e.id.clone(), r.message.clone())),
1240 _ => None,
1241 })
1242 .collect()
1243 }
1244
1245 fn derive_keyed(&self) -> Vec<(String, Message)> {
1246 self.derive_keyed_tagged()
1247 .into_iter()
1248 .map(|(id, m, _, _)| (id, m))
1249 .collect()
1250 }
1251
1252 /// The plan-free projection: every closed turn at `Full` fidelity. Used
1253 /// by `derive_messages` (goal audits, search, human-facing views), which
1254 /// has no `CapacityProfile` to plan against.
1255 fn derive_keyed_tagged(&self) -> Vec<(String, Message, bool, bool)> {
1256 self.derive_with_plan_tagged(None)
1257 }
1258
1259 /// Like `derive_keyed_tagged`, but a `WorkingSetPlan` (from
1260 /// `vak_context::planner::plan`) selects each closed turn's fidelity
1261 /// instead of defaulting every one to `Full` (docs/design/68-context-
1262 /// engine.md §4/§10) — one implementation, the plan just picks what each
1263 /// turn contributes:
1264 ///
1265 /// - `Full` turns project their `full_record`: real `tool_use` blocks,
1266 /// results as schema-driven digests carrying their evidence id, never
1267 /// a character-count trim.
1268 /// - `Card` turns contribute one line each to a single `<turns>` block
1269 /// (`TurnCard::line`), inserted once, right after any compaction
1270 /// summary.
1271 /// - `Packet` turns, and any closed turn the plan omits entirely,
1272 /// contribute nothing here: they are represented only by an existing
1273 /// `Compaction` entry, which the caller (`Agent`'s incremental
1274 /// compaction, §4) guarantees already covers them before this is
1275 /// called with that plan.
1276 /// - The still-open turn (if any) always projects verbatim, regardless
1277 /// of `plan` — it is never planned.
1278 ///
1279 /// The intent note, work contract, and conversation thread are rendered
1280 /// into the request tail instead (§6/§10), read separately via
1281 /// [`SessionLog::tail_sections`]; this function never contributes them.
1282 /// The reset boundary: the chain position before which everything is
1283 /// invisible to the model, and the handoff summary that stands in for
1284 /// it. Only a `reset_all` compaction entry (reset-with-handoff,
1285 /// docs/design/42) moves this; packets never do. `(0, ..., None)` when
1286 /// no reset has happened.
1287 fn reset_boundary(&self) -> (usize, HashMap<String, usize>, Option<String>) {
1288 let chain = self.chain_to_root();
1289 let position: HashMap<String, usize> = chain
1290 .iter()
1291 .enumerate()
1292 .map(|(i, entry)| (entry.id.clone(), i))
1293 .collect();
1294 let last_reset =
1295 chain
1296 .iter()
1297 .enumerate()
1298 .rev()
1299 .find_map(|(pos, entry)| match &entry.payload {
1300 EntryPayload::Compaction(c) if c.reset_all => Some((pos, c.summary.clone())),
1301 _ => None,
1302 });
1303 match last_reset {
1304 Some((pos, summary)) => (pos, position, Some(summary)),
1305 None => (0, position, None),
1306 }
1307 }
1308
1309 /// Every stored packet (non-reset compaction entry) in ledger order.
1310 fn packets(&self) -> Vec<Packet> {
1311 TurnIndex::from_log(self).packets
1312 }
1313
1314 /// The stored packet whose range is exactly `first_turn_id..=last_turn_id`,
1315 /// newest such entry first (docs/design/68 §4: a packet is reused only
1316 /// for the exact range the plan asks for). `None` when no packet
1317 /// covers that range, whatever other packets exist.
1318 pub fn packet_for(&self, first_turn_id: &str, last_turn_id: &str) -> Option<Packet> {
1319 self.packets().into_iter().rev().find(|packet| {
1320 packet.first_turn_id == first_turn_id && packet.last_turn_id == last_turn_id
1321 })
1322 }
1323
1324 /// Whether the packet range a `WorkingSetPlan` asked for
1325 /// (`packet_range = (first, last)`) still needs a summariser call: true
1326 /// when no stored packet covers exactly that range.
1327 pub fn packet_needs_compaction(&self, first_turn_id: &str, last_turn_id: &str) -> bool {
1328 self.packet_for(first_turn_id, last_turn_id).is_none()
1329 }
1330
1331 /// The summariser input for a packet over `first_turn_id..=last_turn_id`:
1332 /// the longest stored packet that starts at the same turn and ends at or
1333 /// before `last_turn_id` (so work already folded in is reused, never
1334 /// re-read from raw history), followed by one `TurnCard` line per turn
1335 /// after it up to `last_turn_id` — cards, never raw history (§4). Also
1336 /// returns a char-count estimate for the caller's `tokens_before`.
1337 pub fn packet_transcript(&self, first_turn_id: &str, last_turn_id: &str) -> (String, u64) {
1338 let (_, position, _) = self.reset_boundary();
1339 let mut out = String::new();
1340 let (Some(&first_pos), Some(&last_pos)) =
1341 (position.get(first_turn_id), position.get(last_turn_id))
1342 else {
1343 return (out, 0);
1344 };
1345 // Seed: the stored packet with this `first` and the greatest `last`
1346 // not past our target.
1347 let seed = self
1348 .packets()
1349 .into_iter()
1350 .filter(|packet| packet.first_turn_id == first_turn_id)
1351 .filter_map(|packet| {
1352 let end = position.get(packet.last_turn_id.as_str()).copied()?;
1353 (end <= last_pos).then_some((end, packet))
1354 })
1355 .max_by_key(|(end, _)| *end);
1356 let mut from_pos = first_pos;
1357 if let Some((end, packet)) = seed {
1358 out.push_str(&packet.summary);
1359 out.push_str("\n\n");
1360 from_pos = end + 1;
1361 }
1362 let index = TurnIndex::from_log(self);
1363 for (turn_number, turn) in index.turns.iter().enumerate() {
1364 let turn_pos = position.get(turn.id.as_str()).copied().unwrap_or(0);
1365 if turn_pos < from_pos || turn_pos > last_pos {
1366 continue;
1367 }
1368 match &turn.card {
1369 Some(card) => {
1370 out.push_str(&card.line(turn_number + 1));
1371 out.push('\n');
1372 }
1373 None => {
1374 for message in turn.full_record() {
1375 out.push_str(&message.text_content());
1376 out.push('\n');
1377 }
1378 }
1379 }
1380 }
1381 let chars = out.chars().count() as u64;
1382 (out, chars)
1383 }
1384
1385 /// The model-visible projection for one request. With a plan, each
1386 /// closed turn contributes exactly what the plan chose for it; without
1387 /// one, every closed turn rides at `Full` (the plan-free view used by

Indexing the workspace…