Vakyartha documentation

455 vak_agent::TurnOutcome::MaxTurnsReached => Advancement::Stalled {
456 reason: "exhausted the turn budget without moving a criterion".into(),
457 },
458 }
459}
460
461/// Close out this turn's episode.
462pub fn end_episode(
463 sessions_home: &Path,
464 handle: &EpisodeHandle,
465 advancement: Advancement,
466 spend_usd: f64,
467) {
468 let ledger = CommitmentLedger::new(sessions_home);
469 if let Err(error) = ledger.append(&Event::new(
470 &handle.commitment_id,
471 EventKind::EpisodeEnded {
472 episode_id: handle.episode_id.clone(),
473 advancement,
474 spend_usd,
475 },
476 )) {
477 eprintln!("[commit] could not close the episode: {error}");
478 }
479}
480
481/// Suspend a commitment on a question nobody here can answer.
482///
483/// This is the `Defer` half of the human-in-the-loop contract: an unattended
484/// surface used to fail closed unconditionally, which is right for a one-shot
485/// turn and destroys month-long work that merely needed to wait.
486pub fn defer_for_human(
487 sessions_home: &Path,
488 commitment_id: &str,
489 question: &str,
490 addressed_to: Option<String>,
491 escalation: vak_intent::Escalation,
492) -> Result<String, vak_commit::LedgerError> {
493 let question_id = uuid::Uuid::now_v7().to_string();
494 CommitmentLedger::new(sessions_home).append(&Event::new(
495 commitment_id,
496 EventKind::Suspended {
497 suspension: vak_commit::Suspension::Human {
498 question_id: question_id.clone(),
499 question: question.to_string(),
500 addressed_to,
501 escalation,
502 },
503 },
504 ))?;
505 Ok(question_id)
506}
507
508/// Record a criterion evaluation the runtime performed.
509pub fn record_evaluation(
510 sessions_home: &Path,
511 commitment_id: &str,
512 evaluation: &Evaluation,
513) -> Result<(), vak_commit::LedgerError> {
514 CommitmentLedger::new(sessions_home).append(&Event::new(
515 commitment_id,
516 EventKind::CriterionEvaluated {
517 criterion_id: evaluation.criterion_id.clone(),
518 result: evaluation.result.clone(),
519 strength: evaluation.strength,
520 },
521 ))
522}
523
524/// Sweep commitments whose economics have run out.
525///
526/// Expiry is an explicit `Expired` verdict rather than a deletion, so the
527/// record says the work lapsed instead of quietly forgetting it existed.
528/// Budget exhaustion deliberately does **not** close anything: a human can
529/// raise the ceiling, and everything the commitment established is still
530/// good, so the scheduler simply stops offering it.
531pub fn sweep_expired(sessions_home: &Path) -> Vec<String> {
532 let ledger = CommitmentLedger::new(sessions_home);
533 let now = chrono::Utc::now();
534 let mut closed = Vec::new();
535 for commitment in ledger.open() {
536 if !commitment.is_expired(now) {
537 continue;
538 }
539 let strength = commitment.achieved_strength();
540 if ledger
541 .append(&Event::new(
542 &commitment.commitment_id,
543 EventKind::Closed {
544 verdict: Verdict::Expired,
545 strength,
546 evidence: Vec::new(),
547 note: "past its relevance window without closing".into(),
548 },
549 ))
550 .is_ok()
551 {
552 closed.push(commitment.commitment_id);
553 }
554 }
555 closed
556}
557
558/// One maintenance pass over the portfolio.
559///
560/// Runs on the server's ordinary tick, independently of whether the heartbeat
561/// is enabled: heartbeat is an opt-in *model* pass and costs tokens, while
562/// everything here is filesystem and clock work that costs none. Tying durable
563/// work's upkeep to an opt-in prober would mean a commitment stopped being
564/// durable the moment someone turned the prober off.
565///
566/// Everything it does is either a state transition the ledger already
567/// authorises or an evaluation the runtime performs itself. It never
568/// dispatches a model and never closes work as fulfilled.
569#[derive(Debug, Default, PartialEq)]
570pub struct Maintenance {
571 /// Closed `Expired` because their relevance window passed.
572 pub expired: Vec<String>,
573 /// Woken because a scheduled time arrived.
574 pub resumed: Vec<String>,
575 /// Woken because a predicate the runtime can check became true.
576 pub satisfied: Vec<String>,
577 /// Deferred questions whose escalation policy came due.
578 pub escalated: Vec<String>,
579}
580
581impl Maintenance {
582 pub fn is_empty(&self) -> bool {
583 self.expired.is_empty()
584 && self.resumed.is_empty()
585 && self.satisfied.is_empty()
586 && self.escalated.is_empty()
587 }
588
589 fn extend(&mut self, other: Maintenance) {
590 self.expired.extend(other.expired);
591 self.resumed.extend(other.resumed);
592 self.satisfied.extend(other.satisfied);
593 self.escalated.extend(other.escalated);
594 }
595}
596
597/// One upkeep pass over every Agent's portfolio.
598///
599/// Commitments live in the Agent that holds them
600/// (`<data home>/agents/<agent_id>/`, docs/design/64-agent-owned-platform.md),
601/// so a pass over one home would leave every other Agent's scheduled wakes
602/// and deadlines unkept. The data home's own ledger is included for work
603/// admitted without an Agent.
604pub async fn maintain_all(shared_data_home: &Path) -> Maintenance {
605 let mut homes = vec![shared_data_home.to_path_buf()];
606 if let Ok(entries) = std::fs::read_dir(shared_data_home.join("agents")) {
607 let mut agents: Vec<_> = entries
608 .flatten()
609 .map(|entry| entry.path())
610 .filter(|path| path.join("commitments.jsonl").is_file())
611 .collect();
612 agents.sort();
613 homes.extend(agents);
614 }
615 let mut report = Maintenance::default();
616 for home in homes {
617 report.extend(maintain(&home).await);
618 }
619 report
620}
621
622/// Advance whatever the clock and the workspace now permit, for the
623/// portfolio in one home. A predicate is checked against the workspace its
624/// commitment belongs to, never the caller's.
625pub async fn maintain(sessions_home: &Path) -> Maintenance {
626 let ledger = CommitmentLedger::new(sessions_home);
627 let now = chrono::Utc::now();
628 let mut report = Maintenance {
629 expired: sweep_expired(sessions_home),
630 ..Maintenance::default()
631 };
632
633 for commitment in ledger.open() {
634 let Some(suspension) = commitment.suspension.clone() else {
635 continue;
636 };
637 match suspension {
638 vak_commit::Suspension::Schedule { at: Some(at), .. } if now >= at => {
639 if ledger
640 .append(&Event::new(
641 &commitment.commitment_id,
642 EventKind::Resumed {
643 reason: "scheduled time reached".into(),
644 },
645 ))
646 .is_ok()
647 {
648 report.resumed.push(commitment.commitment_id.clone());
649 }
650 }
651 // A predicate is checked by the runtime for free. This is the
652 // path that lets "watch X and tell me when Y" cost nothing at all
653 // while Y stays false.
654 vak_commit::Suspension::Predicate { criterion } => {
655 let evaluation = WorkspaceEvaluator {
656 cwd: &commitment.spec.cwd,
657 }
658 .evaluate(&criterion)
659 .await;
660 if evaluation.passed()
661 && record_evaluation(sessions_home, &commitment.commitment_id, &evaluation)
662 .is_ok()
663 && ledger
664 .append(&Event::new(
665 &commitment.commitment_id,
666 EventKind::Resumed {
667 reason: format!("condition met: {}", criterion.statement),
668 },
669 ))
670 .is_ok()
671 {
672 report.satisfied.push(commitment.commitment_id.clone());
673 }
674 }
675 vak_commit::Suspension::Commitment { commitment_id } => {
676 if ledger
677 .get(&commitment_id)
678 .ok()
679 .flatten()
680 .is_some_and(|dependency| {
681 dependency
682 .closure
683 .as_ref()
684 .is_some_and(|closure| closure.verdict == Verdict::Fulfilled)
685 })
686 && ledger
687 .append(&Event::new(
688 &commitment.commitment_id,
689 EventKind::Resumed {
690 reason: format!("dependency {commitment_id} was fulfilled"),
691 },
692 ))
693 .is_ok()
694 {
695 report.resumed.push(commitment.commitment_id.clone());
696 }
697 }
698 vak_commit::Suspension::Human {
699 question_id,
700 escalation,
701 ..
702 } => {
703 if let Some(id) =
704 escalate_if_due(&ledger, &commitment, &question_id, &escalation, now)
705 {
706 report.escalated.push(id);
707 }
708 }
709 _ => {}
710 }
711 }
712 report
713}
714
715/// Apply a deferred question's escalation policy once its deadline passes.
716///
717/// A question with no policy waits forever by design; that is a decision, not
718/// a leak. What must never happen is a silent default standing in for consent
719/// on work that cannot be undone, so `AssumeConservative` is refused there
720/// even if a grant somehow carried it.
721fn escalate_if_due(
722 ledger: &CommitmentLedger,
723 commitment: &vak_commit::Commitment,
724 question_id: &str,
725 escalation: &vak_intent::Escalation,
726 now: chrono::DateTime<chrono::Utc>,
727) -> Option<String> {
728 let waited_hours = (now - commitment.updated_at).num_minutes() as f64 / 60.0;
729 let (after, verdict, note): (u32, Option<Verdict>, &str) = match escalation {
730 vak_intent::Escalation::WaitIndefinitely => return None,
731 vak_intent::Escalation::AssumeConservative { after_hours } => {
732 if !escalation.permitted_for(commitment.spec.reading.stakes) {
733 return None;
734 }
735 (
736 *after_hours,
737 None,
738 "no answer; taking the conservative branch",
739 )
740 }
741 vak_intent::Escalation::AbandonAfter { after_hours } => (
742 *after_hours,
743 Some(Verdict::Abandoned),
744 "no answer within the agreed window",
745 ),
746 vak_intent::Escalation::Reassign { after_hours, .. } => {
747 (*after_hours, None, "reassigned after no answer")
748 }
749 };
750 if waited_hours < f64::from(after) {
751 return None;
752 }
753 let event = match verdict {
754 Some(verdict) => Event::new(
755 &commitment.commitment_id,
756 EventKind::Closed {
757 verdict,
758 strength: commitment.achieved_strength(),
759 evidence: Vec::new(),
760 note: format!("{note} (question {question_id})"),
761 },
762 ),
763 None => Event::new(
764 &commitment.commitment_id,
765 EventKind::Resumed {
766 reason: format!("{note} (question {question_id})"),
767 },
768 ),
769 };
770 ledger
771 .append(&event)
772 .ok()
773 .map(|()| commitment.commitment_id.clone())
774}
775
776/// A criterion evaluator backed by the workspace filesystem.
777///
778/// Only the checks that need no permission gate live here: file existence and
779/// content. `Shell`, `ToolSucceeded` and `FlowCompleted` are permissioned
780/// effects and must cross the tool broker, so they return `Unknown` until a
781/// caller supplies a brokered evaluator — an honest "not determined" rather
782/// than a failure the work did not earn.
783///
784/// It runs in the server process, outside any sandbox, so a criterion may
785/// only name a path inside its commitment's workspace (invariant 10): an
786/// absolute path, a `..` step, or a symlink out of the tree is not checked
787/// at all, rather than turning a pass/fail into a probe of the machine.
788pub struct WorkspaceEvaluator<'a> {
789 pub cwd: &'a Path,
790}
791
792/// `path` inside `cwd`, or `None` when it names anything outside it.
793fn inside_workspace(cwd: &Path, relative: &Path) -> Option<std::path::PathBuf> {
794 if relative.is_absolute()
795 || relative.components().any(|component| {
796 !matches!(
797 component,
798 std::path::Component::Normal(_) | std::path::Component::CurDir
799 )
800 })
801 {
802 return None;
803 }
804 let root = cwd.canonicalize().ok()?;
805 let joined = root.join(relative);
806 let mut probe = joined.as_path();
807 loop {
808 if let Ok(real) = probe.canonicalize() {
809 return real.starts_with(&root).then_some(joined);
810 }
811 probe = probe.parent()?;
812 }
813}
814
815impl vak_commit::CriterionEvaluator for WorkspaceEvaluator<'_> {
816 async fn evaluate(&self, criterion: &WorkCriterion) -> Evaluation {
817 let outside = || {
818 Evaluation::unknown(
819 &criterion.criterion_id,
820 "names a path outside the workspace; not checked",
821 )
822 };
823 match &criterion.kind {
824 CriterionKind::FileExists { path } => {
825 let Some(resolved) = inside_workspace(self.cwd, path) else {
826 return outside();
827 };
828 if resolved.exists() {
829 Evaluation::observed(
830 criterion,
831 CriterionResult::Passed {
832 evidence: format!("{} exists", resolved.display()),
833 },
834 )
835 } else {
836 Evaluation::observed(
837 criterion,
838 CriterionResult::Failed {
839 reason: format!("{} does not exist", resolved.display()),
840 },
841 )
842 }
843 }
844 CriterionKind::FileContains { path, pattern } => {
845 let Some(resolved) = inside_workspace(self.cwd, path) else {
846 return outside();
847 };
848 match std::fs::read_to_string(&resolved) {
849 Ok(text) if text.contains(pattern) => Evaluation::observed(
850 criterion,
851 CriterionResult::Passed {
852 evidence: format!("{} contains the pattern", resolved.display()),
853 },
854 ),
855 Ok(_) => Evaluation::observed(
856 criterion,
857 CriterionResult::Failed {
858 reason: format!("{} does not contain the pattern", resolved.display()),
859 },
860 ),
861 // Unreadable is not the same as failing: the check could
862 // not run, and saying otherwise would be as dishonest as
863 // claiming it passed.
864 Err(error) => Evaluation::unknown(
865 &criterion.criterion_id,
866 format!("could not read {}: {error}", resolved.display()),
867 ),
868 }
869 }
870 _ => Evaluation::unknown(
871 &criterion.criterion_id,
872 "needs a brokered evaluator; not checked here",
873 ),
874 }
875 }
876}
877
878#[cfg(test)]
879#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
880mod tests {
881 use super::*;
882 use vak_commit::CriterionEvaluator;
883 use vak_intent::{Evidence, Horizon, Reading};
884
885 fn intent_with(evidence: Evidence, horizon: Horizon, confidence: f64) -> Intent {
886 intent_on(
887 evidence,
888 horizon,
889 confidence,
890 "t1.0",
891 vak_intent::Lineage::New,
892 )
893 }
894
895 /// A one-strand intent. `strand_id` is also the thread for a new strand
896 /// and a replacement; a continuation inherits its thread.
897 fn intent_on(
898 evidence: Evidence,
899 horizon: Horizon,
900 confidence: f64,
901 strand_id: &str,
902 lineage: vak_intent::Lineage,
903 ) -> Intent {
904 let reading = Reading {
905 horizon,
906 evidence,
907 confidence,
908 axis_confidence: vak_intent::Confidences {
909 act: confidence,
910 horizon: confidence,
911 stakes: confidence,
912 evidence: confidence,
913 },
914 ..Reading::general()
915 };
916 let engagement = vak_intent::derive(&reading, &vak_intent::Authority::default(), false);
917 let thread_id = lineage.continued_thread().unwrap_or(strand_id).to_string();
918 let mut intent = Intent::general(vak_intent::RESOLVER_VERSION);
919 intent.strands = vec![vak_intent::Strand {
920 strand_id: strand_id.into(),
921 thread_id,
922 text: String::new(),
923 reading: reading.clone(),
924 relation: vak_intent::StrandRelation::Independent,
925 lineage,
926 engagement: engagement.clone(),
927 }];
928 intent.reading = reading;
929 intent.engagement = engagement;
930 intent
931 }
932
933 fn begin_episode(
934 sessions_home: &Path,
935 config: &vak_config::Config,
936 intent: &Intent,
937 prompt: &str,
938 session_id: &str,
939 cwd: &Path,
940 ) -> Option<EpisodeHandle> {
941 let plan = plan_episodes(sessions_home, config, intent, chrono::Utc::now());
942 begin_episodes(
943 sessions_home,
944 config,
945 intent,
946 &plan,
947 prompt,
948 session_id,
949 cwd,
950 Some("local"),
951 )
952 .into_iter()
953 .next()
954 }
955
956 fn config() -> vak_config::Config {
957 vak_config::Config::default()
958 }
959
960 #[test]
961 fn a_turn_horizon_opens_no_commitment() {
962 let dir = tempfile::tempdir().unwrap();
963 let handle = begin_episode(
964 dir.path(),
965 &config(),
966 &intent_with(Evidence::None, Horizon::Turn, 0.9),
967 "fix the test",
968 "s1",
969 dir.path(),
970 );
971 assert!(handle.is_none());
972 assert!(CommitmentLedger::new(dir.path()).all().is_empty());
973 }
974
975 #[test]
976 fn a_durable_horizon_opens_one_and_records_an_episode() {
977 let dir = tempfile::tempdir().unwrap();
978 let handle = begin_episode(
979 dir.path(),
980 &config(),
981 &intent_with(Evidence::None, Horizon::Durable, 0.9),
982 "watch the cloud bill every day",
983 "s1",
984 dir.path(),
985 )
986 .expect("a durable turn opens a commitment");
987 let commitment = CommitmentLedger::new(dir.path())
988 .get(&handle.commitment_id)
989 .unwrap()
990 .unwrap();
991 assert_eq!(commitment.episodes.len(), 1);
992 assert_eq!(commitment.spec.objective, "watch the cloud bill every day");
993 }
994
995 /// A commitment on the durable thread, opened directly.
996 fn open_on_thread(ledger: &CommitmentLedger, cwd: &Path, thread: &str) -> String {
997 let mut spec = vak_commit::spec_from_reading(
998 "resume the migration",
999 vak_intent::Reading {
1000 horizon: Horizon::Durable,
1001 ..Reading::general()
1002 },
1003 Vec::new(),
1004 cwd.to_path_buf(),
1005 Economics::default(),
1006 );
1007 spec.thread_id = Some(thread.into());
1008 ledger.open_commitment(spec).unwrap()
1009 }
1010
1011 #[test]
1012 fn resuming_a_session_closes_its_orphaned_episode_before_starting_again() {
1013 let dir = tempfile::tempdir().unwrap();
1014 let ledger = CommitmentLedger::new(dir.path());
1015 let id = open_on_thread(&ledger, dir.path(), "t0.0");
1016 ledger
1017 .append(&Event::new(
1018 &id,
1019 EventKind::EpisodeStarted {
1020 episode_id: "orphan".into(),
1021 session_id: "s1".into(),
1022 },
1023 ))
1024 .unwrap();
1025 let intent = intent_on(
1026 Evidence::None,
1027 Horizon::Durable,
1028 0.9,
1029 "t1.0",
1030 vak_intent::Lineage::Continues {
1031 thread_id: "t0.0".into(),
1032 },
1033 );
1034 let next = begin_episode(
1035 dir.path(),
1036 &config(),
1037 &intent,
1038 "resume the migration",
1039 "s1",
1040 dir.path(),
1041 )
1042 .unwrap();
1043 assert_eq!(next.commitment_id, id);
1044 let commitment = ledger.get(&id).unwrap().unwrap();
1045 assert!(commitment.episodes[0].ended_at.is_some());
1046 assert!(matches!(
1047 commitment.episodes[0].advancement,
1048 Some(Advancement::Blocked { .. })
1049 ));
1050 assert_ne!(next.episode_id, "orphan");
1051 }
1052
1053 /// A strand that continues a durable thread works on that thread's
1054 /// commitment even when its own reading is a one-turn request: "now also
1055 /// check staging" is part of the job it continues.
1056 #[test]
1057 fn a_continuation_works_on_its_threads_commitment_whatever_its_own_horizon() {
1058 let dir = tempfile::tempdir().unwrap();
1059 let ledger = CommitmentLedger::new(dir.path());
1060 let id = open_on_thread(&ledger, dir.path(), "t0.0");
1061 let intent = intent_on(
1062 Evidence::None,
1063 Horizon::Turn,
1064 0.9,
1065 "t1.0",
1066 vak_intent::Lineage::Continues {
1067 thread_id: "t0.0".into(),
1068 },
1069 );
1070 let handle = begin_episode(
1071 dir.path(),
1072 &config(),
1073 &intent,
1074 "also staging",
1075 "s2",
1076 dir.path(),
1077 )
1078 .expect("the continuation joins the open commitment");
1079 assert_eq!(handle.commitment_id, id);
1080 assert_eq!(ledger.all().len(), 1, "no twin was opened");
1081 }
1082
1083 /// `/goal replace` on durable work opens its successor and supersedes the
1084 /// original, keyed by the replacement's own thread.
1085 #[test]
1086 fn a_replacement_supersedes_the_replaced_threads_commitment() {
1087 let dir = tempfile::tempdir().unwrap();
1088 let ledger = CommitmentLedger::new(dir.path());
1089 let old = open_on_thread(&ledger, dir.path(), "t0.0");
1090 let intent = intent_on(
1091 Evidence::None,
1092 Horizon::Turn,
1093 0.9,
1094 "t1.0",
1095 vak_intent::Lineage::Replaces {
1096 thread_id: "t0.0".into(),
1097 },
1098 );
1099 let handle = begin_episode(
1100 dir.path(),
1101 &config(),
1102 &intent,
1103 "do this instead",
1104 "s2",
1105 dir.path(),
1106 )
1107 .expect("the replacement opens a successor");
1108 assert_ne!(handle.commitment_id, old);
1109 let old = ledger.get(&old).unwrap().unwrap();
1110 assert_eq!(
1111 old.superseded_by.as_deref(),
1112 Some(handle.commitment_id.as_str())
1113 );
1114 let new = ledger.get(&handle.commitment_id).unwrap().unwrap();
1115 assert_eq!(new.spec.thread_id.as_deref(), Some("t1.0"));
1116 assert_eq!(new.spec.audience_id.as_deref(), Some("local"));
1117 }
1118
1119 /// Two turns that each ask for durable work open two commitments: strand
1120 /// ids carry the turn id, so the second can never mistake the first's
1121 /// thread for its own.
1122 #[test]
1123 fn separate_durable_requests_open_separate_commitments() {
1124 let dir = tempfile::tempdir().unwrap();
1125 for (turn, prompt) in [
1126 ("t1.0", "watch the bill daily"),
1127 ("t2.0", "watch the logs daily"),
1128 ] {
1129 let intent = intent_on(
1130 Evidence::None,
1131 Horizon::Durable,
1132 0.9,
1133 turn,
1134 vak_intent::Lineage::New,
1135 );
1136 begin_episode(dir.path(), &config(), &intent, prompt, "s1", dir.path()).unwrap();
1137 }
1138 assert_eq!(CommitmentLedger::new(dir.path()).open().len(), 2);
1139 }
1140
1141 /// Only an existing commitment can carry a grant, and only a live one is
1142 /// offered to the turn.
1143 #[test]
1144 fn the_plan_offers_only_live_envelopes_of_existing_commitments() {
1145 let dir = tempfile::tempdir().unwrap();
1146 let ledger = CommitmentLedger::new(dir.path());
1147 let id = open_on_thread(&ledger, dir.path(), "t0.0");
1148 let now = chrono::Utc::now();
1149 let envelope = vak_intent::Envelope {
1150 envelope_id: "env-1".into(),
1151 granted_by: "owner".into(),
1152 granted_at: now,
1153 expires_at: None,
1154 spend_limit_usd: Some(5.0),
1155 path_scope: vec!["docs/**".into()],
1156 tool_scope: Vec::new(),
1157 permission_ceiling: vak_intent::PermissionCeiling::WorkspaceWrite,
1158 escalation: vak_intent::Escalation::WaitIndefinitely,
1159 revoked_at: None,
1160 };
1161 ledger
1162 .append(&Event::new(
1163 &id,
1164 EventKind::EnvelopeGranted {
1165 envelope: Box::new(envelope),
1166 },
1167 ))
1168 .unwrap();
1169 let intent = intent_on(
1170 Evidence::None,
1171 Horizon::Turn,
1172 0.9,
1173 "t1.0",
1174 vak_intent::Lineage::Continues {
1175 thread_id: "t0.0".into(),
1176 },
1177 );
1178 let plan = plan_episodes(dir.path(), &config(), &intent, now);
1179 assert!(plan.envelopes().contains_key("t1.0"));
1180 assert_eq!(plan.enveloped_commitments(), vec![id.clone()]);
1181
1182 ledger
1183 .append(&Event::new(
1184 &id,
1185 EventKind::EnvelopeRevoked {
1186 envelope_id: "env-1".into(),
1187 by: "owner".into(),
1188 },
1189 ))
1190 .unwrap();
1191 let plan = plan_episodes(dir.path(), &config(), &intent, now);
1192 assert!(
1193 plan.envelopes().is_empty(),
1194 "a revoked grant is not offered"
1195 );
1196 }
1197
1198 /// The upkeep evaluator runs outside any sandbox, so a criterion naming a
1199 /// path outside the workspace is not checked at all.
1200 #[tokio::test]
1201 async fn the_workspace_evaluator_refuses_paths_outside_the_workspace() {
1202 let outer = tempfile::tempdir().unwrap();
1203 let workspace = outer.path().join("ws");
1204 std::fs::create_dir_all(&workspace).unwrap();
1205 std::fs::write(outer.path().join("secret.txt"), "hunter2").unwrap();
1206 #[cfg(unix)]
1207 std::os::unix::fs::symlink(outer.path(), workspace.join("up")).unwrap();
1208 let evaluator = WorkspaceEvaluator { cwd: &workspace };
1209 let mut paths = vec![
1210 "../secret.txt".to_string(),
1211 outer.path().join("secret.txt").display().to_string(),
1212 ];
1213 if cfg!(unix) {
1214 paths.push("up/secret.txt".into());
1215 }
1216 for path in paths {
1217 let evaluation = evaluator
1218 .evaluate(&WorkCriterion {
1219 criterion_id: "leak".into(),
1220 statement: "probe".into(),
1221 kind: CriterionKind::FileContains {
1222 path: path.clone().into(),
1223 pattern: "hunter2".into(),
1224 },
1225 required: true,
1226 })
1227 .await;
1228 assert!(
1229 matches!(evaluation.result, CriterionResult::Unknown { .. }),
1230 "{path} was checked: {:?}",
1231 evaluation.result
1232 );
1233 }
1234 }
1235
1236 /// Upkeep reaches every Agent's portfolio, not only the server's own.
1237 #[tokio::test]
1238 async fn upkeep_covers_every_agents_ledger() {
1239 let data = tempfile::tempdir().unwrap();
1240 let agent_home = data.path().join("agents").join("helper");
1241 std::fs::create_dir_all(&agent_home).unwrap();
1242 let ledger = CommitmentLedger::new(&agent_home);
1243 let id = open_on_thread(&ledger, data.path(), "t0.0");
1244 ledger
1245 .append(&Event::new(
1246 &id,
1247 EventKind::Suspended {
1248 suspension: vak_commit::Suspension::Schedule {
1249 at: Some(chrono::Utc::now() - chrono::Duration::minutes(1)),
1250 cron: None,
1251 },
1252 },
1253 ))
1254 .unwrap();
1255 assert_eq!(maintain_all(data.path()).await.resumed, vec![id]);
1256 }
1257
1258 /// A stray recurrence-ish word must not leave a month-long obligation
1259 /// behind. Weak horizon evidence declines to open one.
1260 #[test]
1261 fn a_weak_horizon_reading_opens_nothing() {
1262 let dir = tempfile::tempdir().unwrap();
1263 let handle = begin_episode(
1264 dir.path(),
1265 &config(),
1266 &intent_with(Evidence::None, Horizon::Durable, 0.3),
1267 "maybe keep an eye on things",
1268 "s1",
1269 dir.path(),
1270 );
1271 assert!(handle.is_none());
1272 }
1273
1274 #[test]
1275 fn disabling_commitments_opens_nothing() {
1276 let dir = tempfile::tempdir().unwrap();
1277 let mut config = config();
1278 config.commitment.enabled = false;
1279 assert!(
1280 begin_episode(
1281 dir.path(),
1282 &config,
1283 &intent_with(Evidence::None, Horizon::Durable, 0.9),
1284 "watch the bill every day",
1285 "s1",
1286 dir.path(),
1287 )
1288 .is_none()
1289 );
1290 }
1291
1292 /// The runtime may only propose criteria it could also check. Guessing a
1293 /// shell command would manufacture `Observed` evidence from a guess.
1294 #[test]
1295 fn seeded_criteria_are_never_stronger_than_asserted() {
1296 for evidence in [Evidence::Verified, Evidence::Audited] {
1297 let intent = intent_with(evidence, Horizon::Durable, 0.9);
1298 for criterion in seed_criteria(&intent, "do the thing") {
1299 assert_eq!(
1300 vak_commit::strength_of(&criterion.kind),
1301 Satisfaction::Asserted
1302 );
1303 }
1304 }
1305 // And nothing at all is seeded when no proof is owed.
1306 assert!(seed_criteria(&intent_with(Evidence::None, Horizon::Durable, 0.9), "x").is_empty());
1307 }
1308
1309 /// Which means a seeded commitment cannot close itself: it stays visibly
1310 /// open until a checkable criterion or a human attestation arrives.
1311 #[test]
1312 fn a_seeded_verified_commitment_cannot_be_closed_by_the_seed_alone() {
1313 let dir = tempfile::tempdir().unwrap();
1314 let handle = begin_episode(
1315 dir.path(),
1316 &config(),
1317 &intent_with(Evidence::Verified, Horizon::Durable, 0.9),
1318 "migrate the schema and prove it works",
1319 "s1",
1320 dir.path(),
1321 )
1322 .unwrap();
1323 let ledger = CommitmentLedger::new(dir.path());
1324 ledger
1325 .append(&Event::new(
1326 &handle.commitment_id,
1327 EventKind::CriterionEvaluated {
1328 criterion_id: "objective".into(),
1329 result: CriterionResult::Passed {
1330 evidence: "I believe this is done".into(),
1331 },
1332 strength: Satisfaction::Asserted,
1333 },
1334 ))
1335 .unwrap();
1336 let refused = ledger.append(&Event::new(
1337 &handle.commitment_id,
1338 EventKind::Closed {
1339 verdict: Verdict::Fulfilled,
1340 strength: Satisfaction::Asserted,
1341 evidence: Vec::new(),
1342 note: "done".into(),
1343 },
1344 ));
1345 assert!(refused.is_err());
1346 }
1347
1348 #[test]
1349 fn exhausting_the_turn_budget_counts_as_a_stall_but_answering_does_not() {
1350 let stalled = classify(&vak_agent::TurnOutcome::MaxTurnsReached, 12, Vec::new());
1351 assert!(stalled.is_stall());
1352
1353 let answered = classify(
1354 &vak_agent::TurnOutcome::Completed {
1355 response: vak_llm::AssistantMessage {
1356 content: vec![vak_llm::ContentBlock::Text {
1357 text: "Here is a substantive answer that reduced uncertainty a lot.".into(),
1358 }],
1359 ..vak_llm::AssistantMessage::empty("test-model")
1360 },
1361 },
1362 0,
1363 Vec::new(),
1364 );
1365 assert!(!answered.is_stall());
1366 assert!(matches!(answered, Advancement::Learned { .. }));
1367
1368 let tool_only = classify(
1369 &vak_agent::TurnOutcome::Completed {
1370 response: vak_llm::AssistantMessage::empty("test-model"),
1371 },
1372 12,
1373 Vec::new(),
1374 );
1375 assert!(tool_only.is_stall());
1376 }
1377
1378 #[test]
1379 fn moving_a_criterion_is_advancement_whatever_else_happened() {
1380 let advancement = classify(
1381 &vak_agent::TurnOutcome::MaxTurnsReached,
1382 0,
1383 vec!["tests-pass".into()],
1384 );
1385 assert!(matches!(advancement, Advancement::Advanced { .. }));
1386 assert!(!advancement.is_stall());
1387 }
1388
1389 #[tokio::test]
1390 async fn the_workspace_evaluator_observes_files_and_abstains_on_the_rest() {
1391 let dir = tempfile::tempdir().unwrap();
1392 std::fs::write(dir.path().join("out.txt"), "all good").unwrap();
1393 let evaluator = WorkspaceEvaluator { cwd: dir.path() };
1394
1395 let exists = evaluator
1396 .evaluate(&WorkCriterion {
1397 criterion_id: "c1".into(),
1398 statement: "output exists".into(),
1399 kind: CriterionKind::FileExists {
1400 path: "out.txt".into(),
1401 },
1402 required: true,
1403 })
1404 .await;
1405 assert!(exists.passed());
1406 assert_eq!(exists.strength, Satisfaction::Observed);
1407
1408 // A permissioned check abstains rather than guessing.
1409 let shell = evaluator
1410 .evaluate(&WorkCriterion {
1411 criterion_id: "c2".into(),
1412 statement: "tests pass".into(),
1413 kind: CriterionKind::Shell {
1414 command: "cargo test".into(),
1415 },
1416 required: true,
1417 })
1418 .await;
1419 assert!(!shell.passed());
1420 assert!(matches!(shell.result, CriterionResult::Unknown { .. }));
1421 }
1422
1423 #[tokio::test]
1424 async fn a_predicate_suspension_wakes_when_the_condition_becomes_true() {
1425 let dir = tempfile::tempdir().unwrap();
1426 let ledger = CommitmentLedger::new(dir.path());
1427 let handle = begin_episode(
1428 dir.path(),
1429 &config(),
1430 &intent_with(Evidence::None, Horizon::Durable, 0.9),
1431 "tell me when the report lands, every day",
1432 "s1",
1433 dir.path(),
1434 )
1435 .unwrap();
1436 let criterion = WorkCriterion {
1437 criterion_id: "landed".into(),
1438 statement: "report.csv exists".into(),
1439 kind: CriterionKind::FileExists {
1440 path: "report.csv".into(),
1441 },
1442 required: true,
1443 };
1444 ledger
1445 .append(&Event::new(
1446 &handle.commitment_id,
1447 EventKind::Suspended {
1448 suspension: vak_commit::Suspension::Predicate {
1449 criterion: criterion.clone(),
1450 },
1451 },
1452 ))
1453 .unwrap();
1454

455 vak_agent::TurnOutcome::MaxTurnsReached => Advancement::Stalled {
456 reason: "exhausted the turn budget without moving a criterion".into(),
457 },
458 }
459}
460
461/// Close out this turn's episode.
462pub fn end_episode(
463 sessions_home: &Path,
464 handle: &EpisodeHandle,
465 advancement: Advancement,
466 spend_usd: f64,
467) {
468 let ledger = CommitmentLedger::new(sessions_home);
469 if let Err(error) = ledger.append(&Event::new(
470 &handle.commitment_id,
471 EventKind::EpisodeEnded {
472 episode_id: handle.episode_id.clone(),
473 advancement,
474 spend_usd,
475 },
476 )) {
477 eprintln!("[commit] could not close the episode: {error}");
478 }
479}
480
481/// Suspend a commitment on a question nobody here can answer.
482///
483/// This is the `Defer` half of the human-in-the-loop contract: an unattended
484/// surface used to fail closed unconditionally, which is right for a one-shot
485/// turn and destroys month-long work that merely needed to wait.
486pub fn defer_for_human(
487 sessions_home: &Path,
488 commitment_id: &str,
489 question: &str,
490 addressed_to: Option<String>,
491 escalation: vak_intent::Escalation,
492) -> Result<String, vak_commit::LedgerError> {
493 let question_id = uuid::Uuid::now_v7().to_string();
494 CommitmentLedger::new(sessions_home).append(&Event::new(
495 commitment_id,
496 EventKind::Suspended {
497 suspension: vak_commit::Suspension::Human {
498 question_id: question_id.clone(),
499 question: question.to_string(),
500 addressed_to,
501 escalation,
502 },
503 },
504 ))?;
505 Ok(question_id)
506}
507
508/// Record a criterion evaluation the runtime performed.
509pub fn record_evaluation(
510 sessions_home: &Path,
511 commitment_id: &str,
512 evaluation: &Evaluation,
513) -> Result<(), vak_commit::LedgerError> {
514 CommitmentLedger::new(sessions_home).append(&Event::new(
515 commitment_id,
516 EventKind::CriterionEvaluated {
517 criterion_id: evaluation.criterion_id.clone(),
518 result: evaluation.result.clone(),
519 strength: evaluation.strength,
520 },
521 ))
522}
523
524/// Sweep commitments whose economics have run out.
525///
526/// Expiry is an explicit `Expired` verdict rather than a deletion, so the
527/// record says the work lapsed instead of quietly forgetting it existed.
528/// Budget exhaustion deliberately does **not** close anything: a human can
529/// raise the ceiling, and everything the commitment established is still
530/// good, so the scheduler simply stops offering it.
531pub fn sweep_expired(sessions_home: &Path) -> Vec<String> {
532 let ledger = CommitmentLedger::new(sessions_home);
533 let now = chrono::Utc::now();
534 let mut closed = Vec::new();
535 for commitment in ledger.open() {
536 if !commitment.is_expired(now) {
537 continue;
538 }
539 let strength = commitment.achieved_strength();
540 if ledger
541 .append(&Event::new(
542 &commitment.commitment_id,
543 EventKind::Closed {
544 verdict: Verdict::Expired,
545 strength,
546 evidence: Vec::new(),
547 note: "past its relevance window without closing".into(),
548 },
549 ))
550 .is_ok()
551 {
552 closed.push(commitment.commitment_id);
553 }
554 }
555 closed
556}
557
558/// One maintenance pass over the portfolio.
559///
560/// Runs on the server's ordinary tick, independently of whether the heartbeat
561/// is enabled: heartbeat is an opt-in *model* pass and costs tokens, while
562/// everything here is filesystem and clock work that costs none. Tying durable
563/// work's upkeep to an opt-in prober would mean a commitment stopped being
564/// durable the moment someone turned the prober off.
565///
566/// Everything it does is either a state transition the ledger already
567/// authorises or an evaluation the runtime performs itself. It never
568/// dispatches a model and never closes work as fulfilled.
569#[derive(Debug, Default, PartialEq)]
570pub struct Maintenance {
571 /// Closed `Expired` because their relevance window passed.
572 pub expired: Vec<String>,
573 /// Woken because a scheduled time arrived.
574 pub resumed: Vec<String>,
575 /// Woken because a predicate the runtime can check became true.
576 pub satisfied: Vec<String>,
577 /// Deferred questions whose escalation policy came due.
578 pub escalated: Vec<String>,
579}
580
581impl Maintenance {
582 pub fn is_empty(&self) -> bool {
583 self.expired.is_empty()
584 && self.resumed.is_empty()
585 && self.satisfied.is_empty()
586 && self.escalated.is_empty()
587 }
588
589 fn extend(&mut self, other: Maintenance) {
590 self.expired.extend(other.expired);
591 self.resumed.extend(other.resumed);
592 self.satisfied.extend(other.satisfied);
593 self.escalated.extend(other.escalated);
594 }
595}
596
597/// One upkeep pass over every Agent's portfolio.
598///
599/// Commitments live in the Agent that holds them
600/// (`<data home>/agents/<agent_id>/`, docs/design/64-agent-owned-platform.md),
601/// so a pass over one home would leave every other Agent's scheduled wakes
602/// and deadlines unkept. The data home's own ledger is included for work
603/// admitted without an Agent.
604pub async fn maintain_all(shared_data_home: &Path) -> Maintenance {
605 let mut homes = vec![shared_data_home.to_path_buf()];
606 if let Ok(entries) = std::fs::read_dir(shared_data_home.join("agents")) {
607 let mut agents: Vec<_> = entries
608 .flatten()
609 .map(|entry| entry.path())
610 .filter(|path| path.join("commitments.jsonl").is_file())
611 .collect();
612 agents.sort();
613 homes.extend(agents);
614 }
615 let mut report = Maintenance::default();
616 for home in homes {
617 report.extend(maintain(&home).await);
618 }
619 report
620}
621
622/// Advance whatever the clock and the workspace now permit, for the
623/// portfolio in one home. A predicate is checked against the workspace its
624/// commitment belongs to, never the caller's.
625pub async fn maintain(sessions_home: &Path) -> Maintenance {
626 let ledger = CommitmentLedger::new(sessions_home);
627 let now = chrono::Utc::now();
628 let mut report = Maintenance {
629 expired: sweep_expired(sessions_home),
630 ..Maintenance::default()
631 };
632
633 for commitment in ledger.open() {
634 let Some(suspension) = commitment.suspension.clone() else {
635 continue;
636 };
637 match suspension {
638 vak_commit::Suspension::Schedule { at: Some(at), .. } if now >= at => {
639 if ledger
640 .append(&Event::new(
641 &commitment.commitment_id,
642 EventKind::Resumed {
643 reason: "scheduled time reached".into(),
644 },
645 ))
646 .is_ok()
647 {
648 report.resumed.push(commitment.commitment_id.clone());
649 }
650 }
651 // A predicate is checked by the runtime for free. This is the
652 // path that lets "watch X and tell me when Y" cost nothing at all
653 // while Y stays false.
654 vak_commit::Suspension::Predicate { criterion } => {
655 let evaluation = WorkspaceEvaluator {
656 cwd: &commitment.spec.cwd,
657 }
658 .evaluate(&criterion)
659 .await;
660 if evaluation.passed()
661 && record_evaluation(sessions_home, &commitment.commitment_id, &evaluation)
662 .is_ok()
663 && ledger
664 .append(&Event::new(
665 &commitment.commitment_id,
666 EventKind::Resumed {
667 reason: format!("condition met: {}", criterion.statement),
668 },
669 ))
670 .is_ok()
671 {
672 report.satisfied.push(commitment.commitment_id.clone());
673 }
674 }
675 vak_commit::Suspension::Commitment { commitment_id } => {
676 if ledger
677 .get(&commitment_id)
678 .ok()
679 .flatten()
680 .is_some_and(|dependency| {
681 dependency
682 .closure
683 .as_ref()
684 .is_some_and(|closure| closure.verdict == Verdict::Fulfilled)
685 })
686 && ledger
687 .append(&Event::new(
688 &commitment.commitment_id,
689 EventKind::Resumed {
690 reason: format!("dependency {commitment_id} was fulfilled"),
691 },
692 ))
693 .is_ok()
694 {
695 report.resumed.push(commitment.commitment_id.clone());
696 }
697 }
698 vak_commit::Suspension::Human {
699 question_id,
700 escalation,
701 ..
702 } => {
703 if let Some(id) =
704 escalate_if_due(&ledger, &commitment, &question_id, &escalation, now)
705 {
706 report.escalated.push(id);
707 }
708 }
709 _ => {}
710 }
711 }
712 report
713}
714
715/// Apply a deferred question's escalation policy once its deadline passes.
716///
717/// A question with no policy waits forever by design; that is a decision, not
718/// a leak. What must never happen is a silent default standing in for consent
719/// on work that cannot be undone, so `AssumeConservative` is refused there
720/// even if a grant somehow carried it.
721fn escalate_if_due(
722 ledger: &CommitmentLedger,
723 commitment: &vak_commit::Commitment,
724 question_id: &str,
725 escalation: &vak_intent::Escalation,
726 now: chrono::DateTime<chrono::Utc>,
727) -> Option<String> {
728 let waited_hours = (now - commitment.updated_at).num_minutes() as f64 / 60.0;
729 let (after, verdict, note): (u32, Option<Verdict>, &str) = match escalation {
730 vak_intent::Escalation::WaitIndefinitely => return None,
731 vak_intent::Escalation::AssumeConservative { after_hours } => {
732 if !escalation.permitted_for(commitment.spec.reading.stakes) {
733 return None;
734 }
735 (
736 *after_hours,
737 None,
738 "no answer; taking the conservative branch",
739 )
740 }
741 vak_intent::Escalation::AbandonAfter { after_hours } => (
742 *after_hours,
743 Some(Verdict::Abandoned),
744 "no answer within the agreed window",
745 ),
746 vak_intent::Escalation::Reassign { after_hours, .. } => {
747 (*after_hours, None, "reassigned after no answer")
748 }
749 };
750 if waited_hours < f64::from(after) {
751 return None;
752 }
753 let event = match verdict {
754 Some(verdict) => Event::new(
755 &commitment.commitment_id,
756 EventKind::Closed {
757 verdict,
758 strength: commitment.achieved_strength(),
759 evidence: Vec::new(),
760 note: format!("{note} (question {question_id})"),
761 },
762 ),
763 None => Event::new(
764 &commitment.commitment_id,
765 EventKind::Resumed {
766 reason: format!("{note} (question {question_id})"),
767 },
768 ),
769 };
770 ledger
771 .append(&event)
772 .ok()
773 .map(|()| commitment.commitment_id.clone())
774}
775
776/// A criterion evaluator backed by the workspace filesystem.
777///
778/// Only the checks that need no permission gate live here: file existence and
779/// content. `Shell`, `ToolSucceeded` and `FlowCompleted` are permissioned
780/// effects and must cross the tool broker, so they return `Unknown` until a
781/// caller supplies a brokered evaluator — an honest "not determined" rather
782/// than a failure the work did not earn.
783///
784/// It runs in the server process, outside any sandbox, so a criterion may
785/// only name a path inside its commitment's workspace (invariant 10): an
786/// absolute path, a `..` step, or a symlink out of the tree is not checked
787/// at all, rather than turning a pass/fail into a probe of the machine.
788pub struct WorkspaceEvaluator<'a> {
789 pub cwd: &'a Path,
790}
791
792/// `path` inside `cwd`, or `None` when it names anything outside it.
793fn inside_workspace(cwd: &Path, relative: &Path) -> Option<std::path::PathBuf> {
794 if relative.is_absolute()
795 || relative.components().any(|component| {
796 !matches!(
797 component,
798 std::path::Component::Normal(_) | std::path::Component::CurDir
799 )
800 })
801 {
802 return None;
803 }
804 let root = cwd.canonicalize().ok()?;
805 let joined = root.join(relative);
806 let mut probe = joined.as_path();
807 loop {
808 if let Ok(real) = probe.canonicalize() {
809 return real.starts_with(&root).then_some(joined);
810 }
811 probe = probe.parent()?;
812 }
813}
814
815impl vak_commit::CriterionEvaluator for WorkspaceEvaluator<'_> {
816 async fn evaluate(&self, criterion: &WorkCriterion) -> Evaluation {
817 let outside = || {
818 Evaluation::unknown(
819 &criterion.criterion_id,
820 "names a path outside the workspace; not checked",
821 )
822 };
823 match &criterion.kind {
824 CriterionKind::FileExists { path } => {
825 let Some(resolved) = inside_workspace(self.cwd, path) else {
826 return outside();
827 };
828 if resolved.exists() {
829 Evaluation::observed(
830 criterion,
831 CriterionResult::Passed {
832 evidence: format!("{} exists", resolved.display()),
833 },
834 )
835 } else {
836 Evaluation::observed(
837 criterion,
838 CriterionResult::Failed {
839 reason: format!("{} does not exist", resolved.display()),
840 },
841 )
842 }
843 }
844 CriterionKind::FileContains { path, pattern } => {
845 let Some(resolved) = inside_workspace(self.cwd, path) else {
846 return outside();
847 };
848 match std::fs::read_to_string(&resolved) {
849 Ok(text) if text.contains(pattern) => Evaluation::observed(
850 criterion,
851 CriterionResult::Passed {
852 evidence: format!("{} contains the pattern", resolved.display()),
853 },
854 ),
855 Ok(_) => Evaluation::observed(
856 criterion,
857 CriterionResult::Failed {
858 reason: format!("{} does not contain the pattern", resolved.display()),
859 },
860 ),
861 // Unreadable is not the same as failing: the check could
862 // not run, and saying otherwise would be as dishonest as
863 // claiming it passed.
864 Err(error) => Evaluation::unknown(
865 &criterion.criterion_id,
866 format!("could not read {}: {error}", resolved.display()),
867 ),
868 }
869 }
870 _ => Evaluation::unknown(
871 &criterion.criterion_id,
872 "needs a brokered evaluator; not checked here",
873 ),
874 }
875 }
876}
877
878#[cfg(test)]
879#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
880mod tests {
881 use super::*;
882 use vak_commit::CriterionEvaluator;
883 use vak_intent::{Evidence, Horizon, Reading};
884
885 fn intent_with(evidence: Evidence, horizon: Horizon, confidence: f64) -> Intent {
886 intent_on(
887 evidence,
888 horizon,
889 confidence,
890 "t1.0",
891 vak_intent::Lineage::New,
892 )
893 }
894
895 /// A one-strand intent. `strand_id` is also the thread for a new strand
896 /// and a replacement; a continuation inherits its thread.
897 fn intent_on(
898 evidence: Evidence,
899 horizon: Horizon,
900 confidence: f64,
901 strand_id: &str,
902 lineage: vak_intent::Lineage,
903 ) -> Intent {
904 let reading = Reading {
905 horizon,
906 evidence,
907 confidence,
908 axis_confidence: vak_intent::Confidences {
909 act: confidence,
910 horizon: confidence,
911 stakes: confidence,
912 evidence: confidence,
913 },
914 ..Reading::general()
915 };
916 let engagement = vak_intent::derive(&reading, &vak_intent::Authority::default(), false);
917 let thread_id = lineage.continued_thread().unwrap_or(strand_id).to_string();
918 let mut intent = Intent::general(vak_intent::RESOLVER_VERSION);
919 intent.strands = vec![vak_intent::Strand {
920 strand_id: strand_id.into(),
921 thread_id,
922 text: String::new(),
923 reading: reading.clone(),
924 relation: vak_intent::StrandRelation::Independent,
925 lineage,
926 engagement: engagement.clone(),
927 }];
928 intent.reading = reading;
929 intent.engagement = engagement;
930 intent
931 }
932
933 fn begin_episode(
934 sessions_home: &Path,
935 config: &vak_config::Config,
936 intent: &Intent,
937 prompt: &str,
938 session_id: &str,
939 cwd: &Path,
940 ) -> Option<EpisodeHandle> {
941 let plan = plan_episodes(sessions_home, config, intent, chrono::Utc::now());
942 begin_episodes(
943 sessions_home,
944 config,
945 intent,
946 &plan,
947 prompt,
948 session_id,
949 cwd,
950 Some("local"),
951 )
952 .into_iter()
953 .next()
954 }
955
956 fn config() -> vak_config::Config {
957 vak_config::Config::default()
958 }
959
960 #[test]
961 fn a_turn_horizon_opens_no_commitment() {
962 let dir = tempfile::tempdir().unwrap();
963 let handle = begin_episode(
964 dir.path(),
965 &config(),
966 &intent_with(Evidence::None, Horizon::Turn, 0.9),
967 "fix the test",
968 "s1",
969 dir.path(),
970 );
971 assert!(handle.is_none());
972 assert!(CommitmentLedger::new(dir.path()).all().is_empty());
973 }
974
975 #[test]
976 fn a_durable_horizon_opens_one_and_records_an_episode() {
977 let dir = tempfile::tempdir().unwrap();
978 let handle = begin_episode(
979 dir.path(),
980 &config(),
981 &intent_with(Evidence::None, Horizon::Durable, 0.9),
982 "watch the cloud bill every day",
983 "s1",
984 dir.path(),
985 )
986 .expect("a durable turn opens a commitment");
987 let commitment = CommitmentLedger::new(dir.path())
988 .get(&handle.commitment_id)
989 .unwrap()
990 .unwrap();
991 assert_eq!(commitment.episodes.len(), 1);
992 assert_eq!(commitment.spec.objective, "watch the cloud bill every day");
993 }
994
995 /// A commitment on the durable thread, opened directly.
996 fn open_on_thread(ledger: &CommitmentLedger, cwd: &Path, thread: &str) -> String {
997 let mut spec = vak_commit::spec_from_reading(
998 "resume the migration",
999 vak_intent::Reading {
1000 horizon: Horizon::Durable,
1001 ..Reading::general()
1002 },
1003 Vec::new(),
1004 cwd.to_path_buf(),
1005 Economics::default(),
1006 );
1007 spec.thread_id = Some(thread.into());
1008 ledger.open_commitment(spec).unwrap()
1009 }
1010
1011 #[test]
1012 fn resuming_a_session_closes_its_orphaned_episode_before_starting_again() {
1013 let dir = tempfile::tempdir().unwrap();
1014 let ledger = CommitmentLedger::new(dir.path());
1015 let id = open_on_thread(&ledger, dir.path(), "t0.0");
1016 ledger
1017 .append(&Event::new(
1018 &id,
1019 EventKind::EpisodeStarted {
1020 episode_id: "orphan".into(),
1021 session_id: "s1".into(),
1022 },
1023 ))
1024 .unwrap();
1025 let intent = intent_on(
1026 Evidence::None,
1027 Horizon::Durable,
1028 0.9,
1029 "t1.0",
1030 vak_intent::Lineage::Continues {
1031 thread_id: "t0.0".into(),
1032 },
1033 );
1034 let next = begin_episode(
1035 dir.path(),
1036 &config(),
1037 &intent,
1038 "resume the migration",
1039 "s1",
1040 dir.path(),
1041 )
1042 .unwrap();
1043 assert_eq!(next.commitment_id, id);
1044 let commitment = ledger.get(&id).unwrap().unwrap();
1045 assert!(commitment.episodes[0].ended_at.is_some());
1046 assert!(matches!(
1047 commitment.episodes[0].advancement,
1048 Some(Advancement::Blocked { .. })
1049 ));
1050 assert_ne!(next.episode_id, "orphan");
1051 }
1052
1053 /// A strand that continues a durable thread works on that thread's
1054 /// commitment even when its own reading is a one-turn request: "now also
1055 /// check staging" is part of the job it continues.
1056 #[test]
1057 fn a_continuation_works_on_its_threads_commitment_whatever_its_own_horizon() {
1058 let dir = tempfile::tempdir().unwrap();
1059 let ledger = CommitmentLedger::new(dir.path());
1060 let id = open_on_thread(&ledger, dir.path(), "t0.0");
1061 let intent = intent_on(
1062 Evidence::None,
1063 Horizon::Turn,
1064 0.9,
1065 "t1.0",
1066 vak_intent::Lineage::Continues {
1067 thread_id: "t0.0".into(),
1068 },
1069 );
1070 let handle = begin_episode(
1071 dir.path(),
1072 &config(),
1073 &intent,
1074 "also staging",
1075 "s2",
1076 dir.path(),
1077 )
1078 .expect("the continuation joins the open commitment");
1079 assert_eq!(handle.commitment_id, id);
1080 assert_eq!(ledger.all().len(), 1, "no twin was opened");
1081 }
1082
1083 /// `/goal replace` on durable work opens its successor and supersedes the
1084 /// original, keyed by the replacement's own thread.
1085 #[test]
1086 fn a_replacement_supersedes_the_replaced_threads_commitment() {
1087 let dir = tempfile::tempdir().unwrap();
1088 let ledger = CommitmentLedger::new(dir.path());
1089 let old = open_on_thread(&ledger, dir.path(), "t0.0");
1090 let intent = intent_on(
1091 Evidence::None,
1092 Horizon::Turn,
1093 0.9,
1094 "t1.0",
1095 vak_intent::Lineage::Replaces {
1096 thread_id: "t0.0".into(),
1097 },
1098 );
1099 let handle = begin_episode(
1100 dir.path(),
1101 &config(),
1102 &intent,
1103 "do this instead",
1104 "s2",
1105 dir.path(),
1106 )
1107 .expect("the replacement opens a successor");
1108 assert_ne!(handle.commitment_id, old);
1109 let old = ledger.get(&old).unwrap().unwrap();
1110 assert_eq!(
1111 old.superseded_by.as_deref(),
1112 Some(handle.commitment_id.as_str())
1113 );
1114 let new = ledger.get(&handle.commitment_id).unwrap().unwrap();
1115 assert_eq!(new.spec.thread_id.as_deref(), Some("t1.0"));
1116 assert_eq!(new.spec.audience_id.as_deref(), Some("local"));
1117 }
1118
1119 /// Two turns that each ask for durable work open two commitments: strand
1120 /// ids carry the turn id, so the second can never mistake the first's
1121 /// thread for its own.
1122 #[test]
1123 fn separate_durable_requests_open_separate_commitments() {
1124 let dir = tempfile::tempdir().unwrap();
1125 for (turn, prompt) in [
1126 ("t1.0", "watch the bill daily"),
1127 ("t2.0", "watch the logs daily"),
1128 ] {
1129 let intent = intent_on(
1130 Evidence::None,
1131 Horizon::Durable,
1132 0.9,
1133 turn,
1134 vak_intent::Lineage::New,
1135 );
1136 begin_episode(dir.path(), &config(), &intent, prompt, "s1", dir.path()).unwrap();
1137 }
1138 assert_eq!(CommitmentLedger::new(dir.path()).open().len(), 2);
1139 }
1140
1141 /// Only an existing commitment can carry a grant, and only a live one is
1142 /// offered to the turn.
1143 #[test]
1144 fn the_plan_offers_only_live_envelopes_of_existing_commitments() {
1145 let dir = tempfile::tempdir().unwrap();
1146 let ledger = CommitmentLedger::new(dir.path());
1147 let id = open_on_thread(&ledger, dir.path(), "t0.0");
1148 let now = chrono::Utc::now();
1149 let envelope = vak_intent::Envelope {
1150 envelope_id: "env-1".into(),
1151 granted_by: "owner".into(),
1152 granted_at: now,
1153 expires_at: None,
1154 spend_limit_usd: Some(5.0),
1155 path_scope: vec!["docs/**".into()],
1156 tool_scope: Vec::new(),
1157 permission_ceiling: vak_intent::PermissionCeiling::WorkspaceWrite,
1158 escalation: vak_intent::Escalation::WaitIndefinitely,
1159 revoked_at: None,
1160 };
1161 ledger
1162 .append(&Event::new(
1163 &id,
1164 EventKind::EnvelopeGranted {
1165 envelope: Box::new(envelope),
1166 },
1167 ))
1168 .unwrap();
1169 let intent = intent_on(
1170 Evidence::None,
1171 Horizon::Turn,
1172 0.9,
1173 "t1.0",
1174 vak_intent::Lineage::Continues {
1175 thread_id: "t0.0".into(),
1176 },
1177 );
1178 let plan = plan_episodes(dir.path(), &config(), &intent, now);
1179 assert!(plan.envelopes().contains_key("t1.0"));
1180 assert_eq!(plan.enveloped_commitments(), vec![id.clone()]);
1181
1182 ledger
1183 .append(&Event::new(
1184 &id,
1185 EventKind::EnvelopeRevoked {
1186 envelope_id: "env-1".into(),
1187 by: "owner".into(),
1188 },
1189 ))
1190 .unwrap();
1191 let plan = plan_episodes(dir.path(), &config(), &intent, now);
1192 assert!(
1193 plan.envelopes().is_empty(),
1194 "a revoked grant is not offered"
1195 );
1196 }
1197
1198 /// The upkeep evaluator runs outside any sandbox, so a criterion naming a
1199 /// path outside the workspace is not checked at all.
1200 #[tokio::test]
1201 async fn the_workspace_evaluator_refuses_paths_outside_the_workspace() {
1202 let outer = tempfile::tempdir().unwrap();
1203 let workspace = outer.path().join("ws");
1204 std::fs::create_dir_all(&workspace).unwrap();
1205 std::fs::write(outer.path().join("secret.txt"), "hunter2").unwrap();
1206 #[cfg(unix)]
1207 std::os::unix::fs::symlink(outer.path(), workspace.join("up")).unwrap();
1208 let evaluator = WorkspaceEvaluator { cwd: &workspace };
1209 let mut paths = vec![
1210 "../secret.txt".to_string(),
1211 outer.path().join("secret.txt").display().to_string(),
1212 ];
1213 if cfg!(unix) {
1214 paths.push("up/secret.txt".into());
1215 }
1216 for path in paths {
1217 let evaluation = evaluator
1218 .evaluate(&WorkCriterion {
1219 criterion_id: "leak".into(),
1220 statement: "probe".into(),
1221 kind: CriterionKind::FileContains {
1222 path: path.clone().into(),
1223 pattern: "hunter2".into(),
1224 },
1225 required: true,
1226 })
1227 .await;
1228 assert!(
1229 matches!(evaluation.result, CriterionResult::Unknown { .. }),
1230 "{path} was checked: {:?}",
1231 evaluation.result
1232 );
1233 }
1234 }
1235
1236 /// Upkeep reaches every Agent's portfolio, not only the server's own.
1237 #[tokio::test]
1238 async fn upkeep_covers_every_agents_ledger() {
1239 let data = tempfile::tempdir().unwrap();
1240 let agent_home = data.path().join("agents").join("helper");
1241 std::fs::create_dir_all(&agent_home).unwrap();
1242 let ledger = CommitmentLedger::new(&agent_home);
1243 let id = open_on_thread(&ledger, data.path(), "t0.0");
1244 ledger
1245 .append(&Event::new(
1246 &id,
1247 EventKind::Suspended {
1248 suspension: vak_commit::Suspension::Schedule {
1249 at: Some(chrono::Utc::now() - chrono::Duration::minutes(1)),
1250 cron: None,
1251 },
1252 },
1253 ))
1254 .unwrap();
1255 assert_eq!(maintain_all(data.path()).await.resumed, vec![id]);
1256 }
1257
1258 /// A stray recurrence-ish word must not leave a month-long obligation
1259 /// behind. Weak horizon evidence declines to open one.
1260 #[test]
1261 fn a_weak_horizon_reading_opens_nothing() {
1262 let dir = tempfile::tempdir().unwrap();
1263 let handle = begin_episode(
1264 dir.path(),
1265 &config(),
1266 &intent_with(Evidence::None, Horizon::Durable, 0.3),
1267 "maybe keep an eye on things",
1268 "s1",
1269 dir.path(),
1270 );
1271 assert!(handle.is_none());
1272 }
1273
1274 #[test]
1275 fn disabling_commitments_opens_nothing() {
1276 let dir = tempfile::tempdir().unwrap();
1277 let mut config = config();
1278 config.commitment.enabled = false;
1279 assert!(
1280 begin_episode(
1281 dir.path(),
1282 &config,
1283 &intent_with(Evidence::None, Horizon::Durable, 0.9),
1284 "watch the bill every day",
1285 "s1",
1286 dir.path(),
1287 )
1288 .is_none()
1289 );
1290 }
1291
1292 /// The runtime may only propose criteria it could also check. Guessing a
1293 /// shell command would manufacture `Observed` evidence from a guess.
1294 #[test]
1295 fn seeded_criteria_are_never_stronger_than_asserted() {
1296 for evidence in [Evidence::Verified, Evidence::Audited] {
1297 let intent = intent_with(evidence, Horizon::Durable, 0.9);
1298 for criterion in seed_criteria(&intent, "do the thing") {
1299 assert_eq!(
1300 vak_commit::strength_of(&criterion.kind),
1301 Satisfaction::Asserted
1302 );
1303 }
1304 }
1305 // And nothing at all is seeded when no proof is owed.
1306 assert!(seed_criteria(&intent_with(Evidence::None, Horizon::Durable, 0.9), "x").is_empty());
1307 }
1308
1309 /// Which means a seeded commitment cannot close itself: it stays visibly
1310 /// open until a checkable criterion or a human attestation arrives.
1311 #[test]
1312 fn a_seeded_verified_commitment_cannot_be_closed_by_the_seed_alone() {
1313 let dir = tempfile::tempdir().unwrap();
1314 let handle = begin_episode(
1315 dir.path(),
1316 &config(),
1317 &intent_with(Evidence::Verified, Horizon::Durable, 0.9),
1318 "migrate the schema and prove it works",
1319 "s1",
1320 dir.path(),
1321 )
1322 .unwrap();
1323 let ledger = CommitmentLedger::new(dir.path());
1324 ledger
1325 .append(&Event::new(
1326 &handle.commitment_id,
1327 EventKind::CriterionEvaluated {
1328 criterion_id: "objective".into(),
1329 result: CriterionResult::Passed {
1330 evidence: "I believe this is done".into(),
1331 },
1332 strength: Satisfaction::Asserted,
1333 },
1334 ))
1335 .unwrap();
1336 let refused = ledger.append(&Event::new(
1337 &handle.commitment_id,
1338 EventKind::Closed {
1339 verdict: Verdict::Fulfilled,
1340 strength: Satisfaction::Asserted,
1341 evidence: Vec::new(),
1342 note: "done".into(),
1343 },
1344 ));
1345 assert!(refused.is_err());
1346 }
1347
1348 #[test]
1349 fn exhausting_the_turn_budget_counts_as_a_stall_but_answering_does_not() {
1350 let stalled = classify(&vak_agent::TurnOutcome::MaxTurnsReached, 12, Vec::new());
1351 assert!(stalled.is_stall());
1352
1353 let answered = classify(
1354 &vak_agent::TurnOutcome::Completed {
1355 response: vak_llm::AssistantMessage {
1356 content: vec![vak_llm::ContentBlock::Text {
1357 text: "Here is a substantive answer that reduced uncertainty a lot.".into(),
1358 }],
1359 ..vak_llm::AssistantMessage::empty("test-model")
1360 },
1361 },
1362 0,
1363 Vec::new(),
1364 );
1365 assert!(!answered.is_stall());
1366 assert!(matches!(answered, Advancement::Learned { .. }));
1367
1368 let tool_only = classify(
1369 &vak_agent::TurnOutcome::Completed {
1370 response: vak_llm::AssistantMessage::empty("test-model"),
1371 },
1372 12,
1373 Vec::new(),
1374 );
1375 assert!(tool_only.is_stall());
1376 }
1377
1378 #[test]
1379 fn moving_a_criterion_is_advancement_whatever_else_happened() {
1380 let advancement = classify(
1381 &vak_agent::TurnOutcome::MaxTurnsReached,
1382 0,
1383 vec!["tests-pass".into()],
1384 );
1385 assert!(matches!(advancement, Advancement::Advanced { .. }));
1386 assert!(!advancement.is_stall());
1387 }
1388
1389 #[tokio::test]
1390 async fn the_workspace_evaluator_observes_files_and_abstains_on_the_rest() {
1391 let dir = tempfile::tempdir().unwrap();
1392 std::fs::write(dir.path().join("out.txt"), "all good").unwrap();
1393 let evaluator = WorkspaceEvaluator { cwd: dir.path() };
1394
1395 let exists = evaluator
1396 .evaluate(&WorkCriterion {
1397 criterion_id: "c1".into(),
1398 statement: "output exists".into(),
1399 kind: CriterionKind::FileExists {
1400 path: "out.txt".into(),
1401 },
1402 required: true,
1403 })
1404 .await;
1405 assert!(exists.passed());
1406 assert_eq!(exists.strength, Satisfaction::Observed);
1407
1408 // A permissioned check abstains rather than guessing.
1409 let shell = evaluator
1410 .evaluate(&WorkCriterion {
1411 criterion_id: "c2".into(),
1412 statement: "tests pass".into(),
1413 kind: CriterionKind::Shell {
1414 command: "cargo test".into(),
1415 },
1416 required: true,
1417 })
1418 .await;
1419 assert!(!shell.passed());
1420 assert!(matches!(shell.result, CriterionResult::Unknown { .. }));
1421 }
1422
1423 #[tokio::test]
1424 async fn a_predicate_suspension_wakes_when_the_condition_becomes_true() {
1425 let dir = tempfile::tempdir().unwrap();
1426 let ledger = CommitmentLedger::new(dir.path());
1427 let handle = begin_episode(
1428 dir.path(),
1429 &config(),
1430 &intent_with(Evidence::None, Horizon::Durable, 0.9),
1431 "tell me when the report lands, every day",
1432 "s1",
1433 dir.path(),
1434 )
1435 .unwrap();
1436 let criterion = WorkCriterion {
1437 criterion_id: "landed".into(),
1438 statement: "report.csv exists".into(),
1439 kind: CriterionKind::FileExists {
1440 path: "report.csv".into(),
1441 },
1442 required: true,
1443 };
1444 ledger
1445 .append(&Event::new(
1446 &handle.commitment_id,
1447 EventKind::Suspended {
1448 suspension: vak_commit::Suspension::Predicate {
1449 criterion: criterion.clone(),
1450 },
1451 },
1452 ))
1453 .unwrap();
1454

Indexing the workspace…