Vakyartha documentation

5457 StatusCode::ACCEPTED,
5458 Json(serde_json::json!({
5459 "request_id": evaluation.request.request_id,
5460 "decision": evaluation.decision.as_str(),
5461 "state": "status_requested",
5462 })),
5463 )
5464 .into_response();
5465 }
5466 vak_intent::InterventionKind::Pause => {
5467 handle
5468 .admissions
5469 .lock()
5470 .unwrap_or_else(std::sync::PoisonError::into_inner)
5471 .remove(&request_id);
5472 handle.steering.pause();
5473 record_control_activity(&handle, "Run paused", "pause");
5474 return (
5475 StatusCode::ACCEPTED,
5476 Json(serde_json::json!({
5477 "request_id": evaluation.request.request_id,
5478 "decision": evaluation.decision.as_str(),
5479 "state": "paused",
5480 "reason": evaluation.reason,
5481 })),
5482 )
5483 .into_response();
5484 }
5485 vak_intent::InterventionKind::Resume => {
5486 handle
5487 .admissions
5488 .lock()
5489 .unwrap_or_else(std::sync::PoisonError::into_inner)
5490 .remove(&request_id);
5491 handle.steering.resume();
5492 record_control_activity(&handle, "Run resumed", "resume");
5493 return (
5494 StatusCode::ACCEPTED,
5495 Json(serde_json::json!({
5496 "request_id": evaluation.request.request_id,
5497 "decision": evaluation.decision.as_str(),
5498 "state": "resumed",
5499 "reason": evaluation.reason,
5500 })),
5501 )
5502 .into_response();
5503 }
5504 vak_intent::InterventionKind::Cancel => {
5505 handle
5506 .admissions
5507 .lock()
5508 .unwrap_or_else(std::sync::PoisonError::into_inner)
5509 .remove(&request_id);
5510 handle
5511 .cancel
5512 .lock()
5513 .unwrap_or_else(std::sync::PoisonError::into_inner)
5514 .cancel();
5515 record_control_activity(&handle, "Run cancelled", "cancel");
5516 return (
5517 StatusCode::ACCEPTED,
5518 Json(serde_json::json!({
5519 "request_id": evaluation.request.request_id,
5520 "decision": evaluation.decision.as_str(),
5521 "state": "cancelled",
5522 "reason": evaluation.reason,
5523 })),
5524 )
5525 .into_response();
5526 }
5527 _ => {}
5528 }
5529 let usable: Vec<&RunAttachment> = body
5530 .attachments
5531 .iter()
5532 .filter(|a| !a.data.trim().is_empty())
5533 .collect();
5534 let message = if usable.is_empty() {
5535 vak_llm::Message::user_text(body.text.clone())
5536 } else {
5537 let mut blocks = vec![vak_llm::ContentBlock::text(body.text.clone())];
5538 for a in usable {
5539 blocks.push(vak_llm::ContentBlock::image_base64(
5540 a.mime.clone(),
5541 a.data.trim().to_string(),
5542 ));
5543 }
5544 vak_llm::Message {
5545 role: vak_llm::Role::User,
5546 content: blocks,
5547 }
5548 };
5549
5550 // Admit at the same busy boundary `/run` uses (docs/design/64, "Request
5551 // durability and delivery"): a steer that lands on an IDLE session is a
5552 // fresh admission and starts its own chain, rather than sitting in a
5553 // queue nothing is left to drain (finding 1b). A steer is never a
5554 // restricted (goal/managed/auto) request, so this never rejects.
5555 match admit_or_queue(&handle, false, message.clone()) {
5556 Admission::RejectedBusy => {
5557 unreachable!("send_steering never admits a restricted request kind")
5558 }
5559 Admission::Queued => {
5560 // The ledger was busy: the durable "Intervention queued" /
5561 // "accepted" activity recorded above already covers this
5562 // request. If it landed in the buffer rather than the ledger
5563 // (session was busy at that check too), the busy leg's own
5564 // `http_settle` flush removes this admission guard when it
5565 // settles; nothing here needs to.
5566 (
5567 StatusCode::ACCEPTED,
5568 Json(serde_json::json!({
5569 "request_id": evaluation.request.request_id,
5570 "decision": evaluation.decision.as_str(),
5571 "state": "steering_queued",
5572 })),
5573 )
5574 .into_response()
5575 }
5576 Admission::Started(taken) => {
5577 // The activity recorded above is durable now (whether it was
5578 // written straight into the ledger or is about to be, via this
5579 // very chain) — the in-memory admission guard can be released.
5580 handle
5581 .admissions
5582 .lock()
5583 .unwrap_or_else(std::sync::PoisonError::into_inner)
5584 .remove(&request_id);
5585 let core = handle.core.clone();
5586 let preview_intent = core.resolve_turn_intent(&taken, &message);
5587 let mut preview_outcome =
5588 vak_intent::OutcomeSpec::from_intent(&body.text, &preview_intent);
5589 preview_outcome.evidence_max_age_secs = Some(core.effective_evidence_max_age_secs());
5590 *handle
5591 .intent
5592 .lock()
5593 .unwrap_or_else(std::sync::PoisonError::into_inner) =
5594 Some(vak_session::types::IntentRecord {
5595 reading: preview_intent.reading,
5596 strands: preview_intent.strands,
5597 engagement: preview_intent.engagement,
5598 provenance: preview_intent.provenance,
5599 outcome: Some(preview_outcome),
5600 model_visible: None,
5601 commitment_id: None,
5602 strand_commitments: Default::default(),
5603 });
5604 wait_for_external_subscriber(&handle).await;
5605 spawn_http_turn_chain(
5606 &state,
5607 handle,
5608 core,
5609 taken,
5610 TurnStart::message(message),
5611 None,
5612 );
5613 (
5614 StatusCode::ACCEPTED,
5615 Json(serde_json::json!({
5616 "request_id": evaluation.request.request_id,
5617 "decision": evaluation.decision.as_str(),
5618 "state": "started",
5619 })),
5620 )
5621 .into_response()
5622 }
5623 }
5624}
5625
5626async fn cancel_run(State(state): State<AppState>, Path(id): Path<String>) -> StatusCode {
5627 let Some(handle) = state.get(&id) else {
5628 return StatusCode::NOT_FOUND;
5629 };
5630 handle
5631 .cancel
5632 .lock()
5633 .unwrap_or_else(std::sync::PoisonError::into_inner)
5634 .cancel();
5635 record_control_activity(&handle, "Run cancelled", "cancel");
5636 deny_pending_approvals(&handle);
5637 // A stop means stop: whatever was queued for a continuation leg is
5638 // discarded rather than silently running as the "next" turn once the
5639 // cancelled run unwinds (finding 1c). Input that arrives AFTER this
5640 // drain but before the run actually unwinds is a fresh push into the
5641 // same queue and is unaffected — it becomes the next leg of the chain,
5642 // same as any other steering.
5643 let discarded = handle.steering.drain(vak_agent::DrainMode::All);
5644 if !discarded.is_empty() {
5645 record_activity_or_buffer(
5646 &handle,
5647 vak_session::ActivityRecord {
5648 activity_id: format!("cancel-discard-{}", uuid::Uuid::now_v7()),
5649 turn: None,
5650 kind: vak_session::ActivityKind::Diagnostic,
5651 status: vak_session::ActivityStatus::Cancelled,
5652 label: "Queued input discarded by stop".into(),
5653 detail: Some(format!(
5654 "{} queued message{} discarded",
5655 discarded.len(),
5656 if discarded.len() == 1 { "" } else { "s" }
5657 )),
5658 data: std::collections::BTreeMap::new(),
5659 },
5660 );
5661 }
5662 // No synthesized `RunFinished` here: the run's own settle path
5663 // (`run_turn_chain`) emits the terminal event once it actually stops.
5664 // Broadcasting one from here raced the real one — cancel_run's
5665 // synthetic event could arrive, mark the chat finished on the client,
5666 // and then the run's genuine `RunFinished` landed afterward and
5667 // un-terminated it.
5668 StatusCode::ACCEPTED
5669}
5670
5671async fn pause_run(State(state): State<AppState>, Path(id): Path<String>) -> StatusCode {
5672 let Some(handle) = state.get(&id) else {
5673 return StatusCode::NOT_FOUND;
5674 };
5675 handle.steering.pause();
5676 record_control_activity(&handle, "Run paused", "pause");
5677 StatusCode::ACCEPTED
5678}
5679
5680async fn resume_run(State(state): State<AppState>, Path(id): Path<String>) -> StatusCode {
5681 let Some(handle) = state.get(&id) else {
5682 return StatusCode::NOT_FOUND;
5683 };
5684 handle.steering.resume();
5685 record_control_activity(&handle, "Run resumed", "resume");
5686 StatusCode::ACCEPTED
5687}
5688
5689async fn control_state(
5690 State(state): State<AppState>,
5691 Path(id): Path<String>,
5692) -> axum::response::Response {
5693 let Some(handle) = state.get(&id) else {
5694 return StatusCode::NOT_FOUND.into_response();
5695 };
5696 let running = handle
5697 .session
5698 .lock()
5699 .map(|session| session.is_none())
5700 .unwrap_or(false);
5701 let revision = handle
5702 .session
5703 .lock()
5704 .ok()
5705 .and_then(|guard| {
5706 guard.as_ref().and_then(|session| {
5707 session.chain_to_root().iter().rev().find_map(|entry| {
5708 if let vak_session::EntryPayload::Intent(record) = &entry.payload {
5709 record.outcome.as_ref().map(|outcome| outcome.revision)
5710 } else {
5711 None
5712 }
5713 })
5714 })
5715 })
5716 .or_else(|| {
5717 handle.intent.lock().ok().and_then(|guard| {
5718 guard
5719 .as_ref()?
5720 .outcome
5721 .as_ref()
5722 .map(|outcome| outcome.revision)
5723 })
5724 })
5725 .unwrap_or(0);
5726 Json(serde_json::json!({
5727 "running": running,
5728 "paused": handle.steering.is_paused(),
5729 "revision": revision,
5730 }))
5731 .into_response()
5732}
5733
5734#[derive(serde::Deserialize)]
5735struct PlanChangeBody {
5736 text: String,
5737 #[serde(default = "default_intervention_source")]
5738 source: String,
5739 #[serde(default)]
5740 target_revision: Option<u64>,
5741}
5742
5743fn apply_plan_change(
5744 mut outcome: vak_intent::OutcomeSpec,
5745 request: &vak_intent::InterventionRequest,
5746) -> (vak_intent::OutcomeSpec, (String, String)) {
5747 let before = outcome
5748 .requirements
5749 .iter()
5750 .map(|requirement| requirement.id.clone())
5751 .collect::<Vec<_>>();
5752 match request.kind {
5753 vak_intent::InterventionKind::AddRequirement => {
5754 outcome.requirements.push(vak_intent::OutcomeRequirement {
5755 id: format!("intervention-{}", request.request_id),
5756 kind: vak_intent::RequirementKind::Deliverable,
5757 description: request.text.clone(),
5758 origin: vak_intent::RequirementOrigin::Explicit,
5759 importance: vak_intent::RequirementImportance::Must,
5760 target: None,
5761 })
5762 }
5763 vak_intent::InterventionKind::RemoveRequirement => {
5764 if let Some(target) = request.text.split_whitespace().last() {
5765 outcome
5766 .requirements
5767 .retain(|requirement| requirement.id != target);
5768 }
5769 }
5770 vak_intent::InterventionKind::Replan | vak_intent::InterventionKind::Reprioritize => {
5771 outcome.assumptions.push(request.text.clone());
5772 }
5773 _ => {}
5774 }
5775 outcome.revision = outcome.revision.saturating_add(1);
5776 let after = outcome
5777 .requirements
5778 .iter()
5779 .map(|requirement| requirement.id.clone())
5780 .collect::<Vec<_>>();
5781 (outcome, (before.join(","), after.join(",")))
5782}
5783
5784async fn plan_change(
5785 State(state): State<AppState>,
5786 Path(id): Path<String>,
5787 Json(body): Json<PlanChangeBody>,
5788) -> axum::response::Response {
5789 let Some(handle) = state.get(&id) else {
5790 return StatusCode::NOT_FOUND.into_response();
5791 };
5792 let (kind, command_text) = intervention_of(&body.text);
5793 if !matches!(
5794 kind,
5795 vak_intent::InterventionKind::Replan
5796 | vak_intent::InterventionKind::Reprioritize
5797 | vak_intent::InterventionKind::AddRequirement
5798 | vak_intent::InterventionKind::RemoveRequirement
5799 ) {
5800 return (StatusCode::BAD_REQUEST, "request is not a plan change").into_response();
5801 }
5802 let revision = handle
5803 .session
5804 .lock()
5805 .ok()
5806 .and_then(|guard| {
5807 guard.as_ref().and_then(|session| {
5808 session.chain_to_root().iter().rev().find_map(|entry| {
5809 if let vak_session::EntryPayload::Intent(record) = &entry.payload {
5810 record.outcome.as_ref().map(|outcome| outcome.revision)
5811 } else {
5812 None
5813 }
5814 })
5815 })
5816 })
5817 .or_else(|| {
5818 handle.intent.lock().ok().and_then(|guard| {
5819 guard
5820 .as_ref()
5821 .and_then(|record| record.outcome.as_ref().map(|outcome| outcome.revision))
5822 })
5823 })
5824 .unwrap_or(0);
5825 if body
5826 .target_revision
5827 .is_some_and(|target| target != revision)
5828 {
5829 return (
5830 StatusCode::CONFLICT,
5831 Json(serde_json::json!({
5832 "decision": "rejected",
5833 "reason": "plan revision is stale",
5834 "revision": revision,
5835 })),
5836 )
5837 .into_response();
5838 }
5839 let evaluation = vak_intent::evaluate_intervention(vak_intent::InterventionRequest {
5840 request_id: format!("plan-change-{}", uuid::Uuid::now_v7()),
5841 kind,
5842 text: command_text,
5843 source: control_source_for(&body.source, state.core.surface().slug(), &id),
5844 target_revision: Some(revision),
5845 target_session_id: Some(id.clone()),
5846 target_parent_session_id: target_parent_of(&handle),
5847 });
5848 let mut admitted_revision = None;
5849 let mut requirement_diff = None;
5850 if evaluation.decision == vak_intent::InterventionDecision::Queued
5851 && let Ok(mut guard) = handle.session.lock()
5852 && let Some(session) = guard.as_mut()
5853 && let Some(record) = session.chain_to_root().iter().rev().find_map(|entry| {
5854 if let vak_session::EntryPayload::Intent(record) = &entry.payload {
5855 Some(record.as_ref())
5856 } else {
5857 None
5858 }
5859 })
5860 && let Some(outcome) = record.outcome.clone()
5861 {
5862 let (outcome, diff) = apply_plan_change(outcome, &evaluation.request);
5863 requirement_diff = Some(diff);
5864 let update = vak_session::types::IntentRecord {
5865 reading: record.reading.clone(),
5866 strands: record.strands.clone(),
5867 engagement: record.engagement.clone(),
5868 provenance: record.provenance.clone(),
5869 outcome: Some(outcome),
5870 model_visible: None,
5871 commitment_id: record.commitment_id.clone(),
5872 strand_commitments: record.strand_commitments.clone(),
5873 };
5874 if session.append_intent(update.clone()).is_ok() {
5875 *handle
5876 .intent
5877 .lock()
5878 .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(update);
5879 admitted_revision = Some(revision + 1);
5880 }
5881 }
5882 if evaluation.decision == vak_intent::InterventionDecision::Queued
5883 && admitted_revision.is_none()
5884 && let Some(mut record) = handle.intent.lock().ok().and_then(|guard| guard.clone())
5885 && let Some(outcome) = record.outcome.take()
5886 {
5887 let (outcome, diff) = apply_plan_change(outcome, &evaluation.request);
5888 requirement_diff = Some(diff);
5889 let update = vak_session::types::IntentRecord {
5890 reading: record.reading,
5891 strands: record.strands,
5892 engagement: record.engagement,
5893 provenance: record.provenance,
5894 outcome: Some(outcome.clone()),
5895 model_visible: None,
5896 commitment_id: record.commitment_id,
5897 strand_commitments: record.strand_commitments,
5898 };
5899 *handle
5900 .intent
5901 .lock()
5902 .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(update.clone());
5903 handle.steering.push_outcome_update(update);
5904 admitted_revision = Some(outcome.revision);
5905 }
5906 record_activity_or_buffer(
5907 &handle,
5908 vak_session::ActivityRecord {
5909 activity_id: evaluation.request.request_id.clone(),
5910 turn: None,
5911 kind: vak_session::ActivityKind::Diagnostic,
5912 status: if evaluation.decision == vak_intent::InterventionDecision::RequiresHuman {
5913 vak_session::ActivityStatus::Denied
5914 } else {
5915 vak_session::ActivityStatus::Pending
5916 },
5917 label: "Plan change evaluated".into(),
5918 detail: Some(evaluation.reason.clone()),
5919 data: std::collections::BTreeMap::from([
5920 ("decision".into(), evaluation.decision.as_str().into()),
5921 ("source".into(), body.source),
5922 (
5923 "revision".into(),
5924 admitted_revision.unwrap_or(revision).to_string(),
5925 ),
5926 (
5927 "requirements_before".into(),
5928 requirement_diff
5929 .as_ref()
5930 .map(|diff| diff.0.clone())
5931 .unwrap_or_else(String::new),
5932 ),
5933 (
5934 "requirements_after".into(),
5935 requirement_diff
5936 .as_ref()
5937 .map(|diff| diff.1.clone())
5938 .unwrap_or_else(String::new),
5939 ),
5940 ("request".into(), body.text),
5941 ]),
5942 },
5943 );
5944 let status = if evaluation.decision == vak_intent::InterventionDecision::RequiresHuman {
5945 StatusCode::CONFLICT
5946 } else {
5947 handle
5948 .steering
5949 .push_steering(evaluation.request.text.clone());
5950 StatusCode::ACCEPTED
5951 };
5952 (
5953 status,
5954 Json(serde_json::json!({
5955 "request_id": evaluation.request.request_id,
5956 "decision": evaluation.decision.as_str(),
5957 "revision": admitted_revision.unwrap_or(revision),
5958 "reason": evaluation.reason,
5959 })),
5960 )
5961 .into_response()
5962}
5963
5964pub(crate) fn record_control_activity(handle: &SessionHandle, label: &str, control: &str) {
5965 record_activity_or_buffer(
5966 handle,
5967 vak_session::ActivityRecord {
5968 activity_id: format!("control-{}", uuid::Uuid::now_v7()),
5969 turn: None,
5970 kind: vak_session::ActivityKind::Diagnostic,
5971 status: vak_session::ActivityStatus::Succeeded,
5972 label: label.into(),
5973 detail: Some("operator control-plane request".into()),
5974 data: std::collections::BTreeMap::from([
5975 ("control".into(), control.into()),
5976 ("source".into(), "human".into()),
5977 ]),
5978 },
5979 );
5980}
5981
5982fn record_activity_or_buffer(handle: &SessionHandle, activity: vak_session::ActivityRecord) {
5983 if let Ok(mut session) = handle.session.lock()
5984 && let Some(session) = session.as_mut()
5985 {
5986 let _ = session.append_activity(activity);
5987 } else if let Ok(mut activities) = handle.activity_buffer.lock() {
5988 activities.push(activity);
5989 }
5990}
5991
5992fn record_routing_data(
5993 data: &mut std::collections::BTreeMap<String, String>,
5994 routing: &RoutingEnvelope,
5995) {
5996 if let Some(value) = routing.message_id.as_deref() {
5997 data.insert("message_id".into(), value.into());
5998 }
5999 if let Some(value) = routing.conversation_id.as_deref() {
6000 data.insert("conversation_id".into(), value.into());
6001 }
6002 if let Some(value) = routing.target_work_id.as_deref() {
6003 data.insert("target_work_id".into(), value.into());
6004 }
6005 if let Some(value) = routing.target_result_id.as_deref() {
6006 data.insert("target_result_id".into(), value.into());
6007 }
6008 if let Some(value) = routing.relation.as_deref() {
6009 data.insert("relation".into(), value.into());
6010 }
6011 if let Some(value) = routing.outcome_revision {
6012 data.insert("outcome_revision".into(), value.to_string());
6013 }
6014 if let Some(value) = routing.provenance.as_deref() {
6015 data.insert("routing_provenance".into(), value.into());
6016 }
6017}
6018
6019/// The session that dispatched this one, from its own header: what "an agent
6020/// may control only its own children" is checked against. Unknown while the
6021/// ledger is out on a running turn, and unknown refuses an agent's control.
6022fn target_parent_of(handle: &SessionHandle) -> Option<String> {
6023 handle.session.lock().ok().and_then(|guard| {
6024 guard.as_ref().and_then(|log| {
6025 log.header()
6026 .and_then(|header| header.parent_session_id.clone())
6027 })
6028 })
6029}
6030
6031fn routing_revision_is_current(handle: &SessionHandle, expected: u64) -> bool {
6032 handle.intent.lock().ok().is_some_and(|record| {
6033 record
6034 .as_ref()
6035 .and_then(|intent| intent.outcome.as_ref())
6036 .is_some_and(|outcome| outcome.revision == expected)
6037 })
6038}
6039
6040fn routing_summary(routing: Option<&RoutingEnvelope>) -> String {
6041 let Some(routing) = routing else {
6042 return String::new();
6043 };
6044 let mut data = std::collections::BTreeMap::new();
6045 record_routing_data(&mut data, routing);
6046 serde_json::to_string(&data).unwrap_or_default()
6047}
6048
6049fn deny_pending_approvals(handle: &SessionHandle) {
6050 let requests: Vec<ApprovalRequest> = handle
6051 .pending
6052 .lock()
6053 .unwrap_or_else(std::sync::PoisonError::into_inner)
6054 .drain()
6055 .map(|(_, request)| request)
6056 .collect();
6057 for request in requests {
6058 request.respond(false);
6059 }
6060}
6061
6062// ---- Worker control plane -------------------------------------------------
6063//
6064// Children already stream lifecycle/tool events into the parent session's
6065// SSE channel; these endpoints add the missing half: listing, steering, and
6066// stopping from a remote surface. Scope-checked against the parent so one
6067// session can never touch another's child.
6068
6069async fn list_workers(
6070 State(state): State<AppState>,
6071 Path(id): Path<String>,
6072) -> Json<serde_json::Value> {
6073 let children = state.core.workers().active_for(&id);
6074 Json(serde_json::json!({ "workers": children }))
6075}
6076
6077#[derive(serde::Deserialize)]
6078struct WorkerSteerBody {
6079 text: String,
6080}
6081
6082async fn steer_worker(
6083 State(state): State<AppState>,
6084 Path((id, child)): Path<(String, String)>,
6085 Json(body): Json<WorkerSteerBody>,
6086) -> StatusCode {
6087 if state.core.workers().parent_of(&child).as_deref() != Some(id.as_str()) {
6088 return StatusCode::NOT_FOUND;
6089 }
6090 if state.core.workers().steer(&child, &body.text) {
6091 StatusCode::ACCEPTED
6092 } else {
6093 StatusCode::CONFLICT
6094 }
6095}
6096
6097async fn stop_worker(
6098 State(state): State<AppState>,
6099 Path((id, child)): Path<(String, String)>,
6100) -> StatusCode {
6101 if state.core.workers().parent_of(&child).as_deref() != Some(id.as_str()) {
6102 return StatusCode::NOT_FOUND;
6103 }
6104 if state.core.workers().stop(&child) {
6105 StatusCode::ACCEPTED
6106 } else {
6107 StatusCode::CONFLICT
6108 }
6109}
6110
6111#[derive(serde::Deserialize)]
6112struct ApprovalBody {
6113 approve: bool,
6114 /// "…and don't ask again for calls like this one". Derives the narrowest
6115 /// rule that covers this call and persists it to
6116 /// `.vak/permissions.local.toml`. Only meaningful alongside
6117 /// `approve: true` — remembering a refusal would be a deny rule, which
6118 /// is a different and much heavier decision than answering one gate.
6119 #[serde(default)]
6120 remember: bool,
6121}
6122
6123#[derive(serde::Deserialize)]
6124struct OutcomeReviewBody {
6125 verdict: String,
6126 #[serde(default)]
6127 turn: Option<usize>,
6128 #[serde(default)]
6129 note: Option<String>,
6130}
6131
6132/// Record an operator's review of a computed outcome. This is deliberately
6133/// separate from approval: reviewing a result never authorizes a tool call.
6134async fn record_outcome_review(
6135 State(state): State<AppState>,
6136 Path(id): Path<String>,
6137 Json(body): Json<OutcomeReviewBody>,
6138) -> axum::response::Response {
6139 use axum::response::IntoResponse;
6140 if !matches!(
6141 body.verdict.as_str(),
6142 "accepted" | "needs_work" | "rejected"
6143 ) {
6144 return (StatusCode::BAD_REQUEST, "invalid outcome review verdict").into_response();
6145 }
6146 let Some(handle) = state.get(&id) else {
6147 return StatusCode::NOT_FOUND.into_response();
6148 };
6149 let Ok(mut guard) = handle.session.lock() else {
6150 return StatusCode::CONFLICT.into_response();
6151 };
6152 let Some(session) = guard.as_mut() else {
6153 return StatusCode::NOT_FOUND.into_response();
6154 };
6155 let reviewed_turn = body.turn.or_else(|| {
6156 session
6157 .chain_to_root()
6158 .iter()
6159 .filter_map(|entry| match &entry.payload {
6160 vak_session::EntryPayload::Activity(activity)
6161 if activity.label == "Outcome evaluation" =>
6162 {
6163 activity.turn
6164 }
6165 _ => None,
6166 })
6167 .next_back()
6168 });
6169 if body.turn.is_some() && reviewed_turn.is_none() {
6170 return (StatusCode::NOT_FOUND, "outcome evaluation turn not found").into_response();
6171 }
6172 let activity = vak_session::ActivityRecord {
6173 activity_id: format!("outcome-review-{}", uuid::Uuid::now_v7()),
6174 turn: reviewed_turn,
6175 kind: vak_session::ActivityKind::Diagnostic,
6176 status: vak_session::ActivityStatus::Succeeded,
6177 label: "Outcome review".into(),
6178 detail: body.note,
6179 data: std::collections::BTreeMap::from([("verdict".into(), body.verdict)]),
6180 };
6181 if let Err(error) = session.append_activity(activity) {
6182 return (StatusCode::INTERNAL_SERVER_ERROR, error.to_string()).into_response();
6183 }
6184 Json(serde_json::json!({ "recorded": true })).into_response()
6185}
6186
6187/// `POST /sessions/{id}/approvals/{req_id}` — answer one gate.
6188///
6189/// `remember` is the mechanism `docs/design/08-permissions.md` has described
6190/// since the engine shipped and nothing ever called: `Core::learn_allow_rule`
6191/// existed, was tested, wrote a validated file, and had no caller on any
6192/// surface, because the interactive approver its comment referred to was
6193/// never built. This is that caller.
6194///
6195/// Remembering never blocks the answer. The gate is resolved first; a
6196/// failure to derive or persist a rule is reported alongside a successful
6197/// approval, because the run is already waiting and a bookkeeping problem
6198/// must not become a denial.
6199async fn answer_approval(
6200 State(state): State<AppState>,
6201 Path((id, req_id)): Path<(String, String)>,
6202 Json(body): Json<ApprovalBody>,
6203) -> axum::response::Response {
6204 use axum::response::IntoResponse;
6205 // Look the request up in THIS session's pending map only: approvals
6206 // are never resolvable across sessions.
6207 let Some(handle) = state.get(&id) else {
6208 return StatusCode::NOT_FOUND.into_response();
6209 };
6210 let Some(req) = handle
6211 .pending
6212 .lock()
6213 .unwrap_or_else(std::sync::PoisonError::into_inner)
6214 .remove(&req_id)
6215 else {
6216 return StatusCode::NOT_FOUND.into_response();
6217 };
6218 let tool = req.tool.clone();
6219 let args_json = req.args_json.clone();
6220 *req.answered_by
6221 .lock()
6222 .unwrap_or_else(std::sync::PoisonError::into_inner) =
6223 Some(("operator".into(), "You".into()));
6224 req.respond(body.approve);
6225
6226 let mut learned: Option<String> = None;
6227 let mut learn_error: Option<String> = None;
6228 if body.remember && body.approve {
6229 match serde_json::from_str::<serde_json::Value>(&args_json) {
6230 // The gate's own core, not the server's: a learned rule belongs
6231 // to the workspace whose call raised it.
6232 Ok(args) => match handle.core.learn_from_call(&tool, &args) {
6233 Ok(spec) => {
6234 vak_core::security_events::record(
6235 &state.core.sessions_home(),
6236 vak_core::security_events::EventKind::ConfigChange,
6237 "permission_rule_learned",
6238 &format!("session={id} rule={spec}"),
6239 None,
6240 );
6241 learned = Some(spec);
6242 }
6243 Err(error) => learn_error = Some(error.to_string()),
6244 },
6245 Err(error) => learn_error = Some(format!("unreadable tool arguments: {error}")),
6246 }
6247 }
6248 (
6249 StatusCode::OK,
6250 Json(serde_json::json!({
6251 "approved": body.approve,
6252 "learned_rule": learned,
6253 "learn_error": learn_error,
6254 })),
6255 )
6256 .into_response()
6257}
6258
6259/// Dispatch forensics (docs/design/42-managed-work-contracts.md): the session's work
6260/// receipts, newest last.
6261async fn session_receipts(
6262 State(state): State<AppState>,
6263 Path(id): Path<String>,
6264) -> Result<Json<Vec<vak_llm::WorkReceipt>>, StatusCode> {
6265 let Some(handle) = state.get(&id) else {
6266 return Err(StatusCode::NOT_FOUND);
6267 };
6268 let Ok(session) = handle.session.lock() else {
6269 return Err(StatusCode::NOT_FOUND);
6270 };
6271 match session.as_ref() {
6272 Some(log) => Ok(Json(log.receipts().into_iter().cloned().collect())),
6273 None => Err(StatusCode::NOT_FOUND),
6274 }
6275}
6276
6277async fn session_work(
6278 State(state): State<AppState>,
6279 Path(id): Path<String>,
6280) -> Result<Json<Option<vak_session::work::WorkProjection>>, StatusCode> {
6281 if let Some(handle) = state.get(&id) {
6282 let guard = handle
6283 .session
6284 .lock()
6285 .unwrap_or_else(std::sync::PoisonError::into_inner);
6286 let Some(session) = guard.as_ref() else {
6287 return Err(StatusCode::NOT_FOUND);
6288 };
6289 return session
6290 .work_projection()
6291 .map(Json)
6292 .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR);
6293 }
6294 let Some(session) = open_historical_session(&state, &id) else {
6295 return Err(StatusCode::NOT_FOUND);
6296 };
6297 session
6298 .work_projection()
6299 .map(Json)
6300 .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
6301}
6302
6303#[derive(serde::Deserialize)]
6304#[serde(tag = "operation", rename_all = "snake_case")]
6305enum WorkCommand {
6306 Transition {
6307 item_id: String,
6308 to: vak_session::types::WorkItemStatus,
6309 #[serde(default)]
6310 reason: String,
6311 },
6312 Cancel {
6313 #[serde(default)]
6314 reason: String,
6315 },
6316 Resume {
6317 #[serde(default)]
6318 reason: String,
6319 },
6320 Retry {
6321 item_id: String,
6322 #[serde(default)]
6323 reason: String,
6324 },
6325 ResolveAssumption {
6326 assumption_id: String,
6327 resolution: String,
6328 },
6329 AttachEvidence {
6330 item_id: String,
6331 evidence: vak_session::types::EvidenceRef,
6332 },
6333 Assign {
6334 item_id: String,
6335 owner: vak_session::types::WorkOwner,
6336 child_session_id: Option<String>,
6337 },
6338 Revise {
6339 contract: vak_session::types::WorkContract,
6340 #[serde(default)]
6341 reason: String,
6342 },
6343}
6344
6345#[derive(serde::Deserialize)]
6346struct WorkReason {
6347 #[serde(default)]
6348 reason: String,
6349}
6350
6351#[derive(serde::Deserialize)]
6352struct WorkReassign {
6353 owner: vak_session::types::WorkOwner,
6354 #[serde(default)]
6355 child_session_id: Option<String>,
6356}
6357
6358async fn session_work_confirm(
6359 state: State<AppState>,
6360 path: Path<String>,
6361 body: Option<Json<WorkReason>>,
6362) -> axum::response::Response {
6363 session_work_command(
6364 state,
6365 path,
6366 Json(WorkCommand::Resume {
6367 reason: body
6368 .map(|body| body.0.reason)
6369 .unwrap_or_else(|| "confirmed by operator".into()),
6370 }),
6371 )
6372 .await
6373}
6374
6375async fn session_work_revise(
6376 state: State<AppState>,
6377 path: Path<String>,
6378 Json(body): Json<WorkCommand>,
6379) -> axum::response::Response {
6380 session_work_command(state, path, Json(body)).await
6381}
6382
6383async fn session_work_retry(
6384 state: State<AppState>,
6385 Path((id, item_id)): Path<(String, String)>,
6386 body: Option<Json<WorkReason>>,
6387) -> axum::response::Response {
6388 session_work_command(
6389 state,
6390 Path(id),
6391 Json(WorkCommand::Retry {
6392 item_id,
6393 reason: body
6394 .map(|body| body.0.reason)
6395 .unwrap_or_else(|| "retry requested by operator".into()),
6396 }),
6397 )
6398 .await
6399}
6400
6401async fn session_work_cancel_item(
6402 state: State<AppState>,
6403 Path((id, item_id)): Path<(String, String)>,
6404 body: Option<Json<WorkReason>>,
6405) -> axum::response::Response {
6406 session_work_command(
6407 state,
6408 Path(id),
6409 Json(WorkCommand::Transition {
6410 item_id,
6411 to: vak_session::types::WorkItemStatus::Cancelled,
6412 reason: body
6413 .map(|body| body.0.reason)
6414 .unwrap_or_else(|| "cancelled by operator".into()),
6415 }),
6416 )
6417 .await
6418}
6419
6420async fn session_work_reassign(
6421 state: State<AppState>,
6422 Path((id, item_id)): Path<(String, String)>,
6423 Json(body): Json<WorkReassign>,
6424) -> axum::response::Response {
6425 session_work_command(
6426 state,
6427 Path(id),
6428 Json(WorkCommand::Assign {
6429 item_id,
6430 owner: body.owner,
6431 child_session_id: body.child_session_id,
6432 }),
6433 )
6434 .await
6435}
6436
6437/// Apply an operator/user work action to the same append-only ledger used by
6438/// the agent. The current projection supplies both the expected state and
6439/// revision, so stale clients receive a conflict instead of silently
6440/// overwriting a newer action.
6441async fn session_work_command(
6442 State(state): State<AppState>,
6443 Path(id): Path<String>,
6444 Json(command): Json<WorkCommand>,
6445) -> axum::response::Response {
6446 use axum::response::IntoResponse;
6447 let Some(handle) = state.get(&id) else {
6448 return StatusCode::NOT_FOUND.into_response();
6449 };
6450 let mut guard = handle
6451 .session
6452 .lock()
6453 .unwrap_or_else(std::sync::PoisonError::into_inner);
6454 let Some(session) = guard.as_mut() else {
6455 return (
6456 StatusCode::CONFLICT,

5457 StatusCode::ACCEPTED,
5458 Json(serde_json::json!({
5459 "request_id": evaluation.request.request_id,
5460 "decision": evaluation.decision.as_str(),
5461 "state": "status_requested",
5462 })),
5463 )
5464 .into_response();
5465 }
5466 vak_intent::InterventionKind::Pause => {
5467 handle
5468 .admissions
5469 .lock()
5470 .unwrap_or_else(std::sync::PoisonError::into_inner)
5471 .remove(&request_id);
5472 handle.steering.pause();
5473 record_control_activity(&handle, "Run paused", "pause");
5474 return (
5475 StatusCode::ACCEPTED,
5476 Json(serde_json::json!({
5477 "request_id": evaluation.request.request_id,
5478 "decision": evaluation.decision.as_str(),
5479 "state": "paused",
5480 "reason": evaluation.reason,
5481 })),
5482 )
5483 .into_response();
5484 }
5485 vak_intent::InterventionKind::Resume => {
5486 handle
5487 .admissions
5488 .lock()
5489 .unwrap_or_else(std::sync::PoisonError::into_inner)
5490 .remove(&request_id);
5491 handle.steering.resume();
5492 record_control_activity(&handle, "Run resumed", "resume");
5493 return (
5494 StatusCode::ACCEPTED,
5495 Json(serde_json::json!({
5496 "request_id": evaluation.request.request_id,
5497 "decision": evaluation.decision.as_str(),
5498 "state": "resumed",
5499 "reason": evaluation.reason,
5500 })),
5501 )
5502 .into_response();
5503 }
5504 vak_intent::InterventionKind::Cancel => {
5505 handle
5506 .admissions
5507 .lock()
5508 .unwrap_or_else(std::sync::PoisonError::into_inner)
5509 .remove(&request_id);
5510 handle
5511 .cancel
5512 .lock()
5513 .unwrap_or_else(std::sync::PoisonError::into_inner)
5514 .cancel();
5515 record_control_activity(&handle, "Run cancelled", "cancel");
5516 return (
5517 StatusCode::ACCEPTED,
5518 Json(serde_json::json!({
5519 "request_id": evaluation.request.request_id,
5520 "decision": evaluation.decision.as_str(),
5521 "state": "cancelled",
5522 "reason": evaluation.reason,
5523 })),
5524 )
5525 .into_response();
5526 }
5527 _ => {}
5528 }
5529 let usable: Vec<&RunAttachment> = body
5530 .attachments
5531 .iter()
5532 .filter(|a| !a.data.trim().is_empty())
5533 .collect();
5534 let message = if usable.is_empty() {
5535 vak_llm::Message::user_text(body.text.clone())
5536 } else {
5537 let mut blocks = vec![vak_llm::ContentBlock::text(body.text.clone())];
5538 for a in usable {
5539 blocks.push(vak_llm::ContentBlock::image_base64(
5540 a.mime.clone(),
5541 a.data.trim().to_string(),
5542 ));
5543 }
5544 vak_llm::Message {
5545 role: vak_llm::Role::User,
5546 content: blocks,
5547 }
5548 };
5549
5550 // Admit at the same busy boundary `/run` uses (docs/design/64, "Request
5551 // durability and delivery"): a steer that lands on an IDLE session is a
5552 // fresh admission and starts its own chain, rather than sitting in a
5553 // queue nothing is left to drain (finding 1b). A steer is never a
5554 // restricted (goal/managed/auto) request, so this never rejects.
5555 match admit_or_queue(&handle, false, message.clone()) {
5556 Admission::RejectedBusy => {
5557 unreachable!("send_steering never admits a restricted request kind")
5558 }
5559 Admission::Queued => {
5560 // The ledger was busy: the durable "Intervention queued" /
5561 // "accepted" activity recorded above already covers this
5562 // request. If it landed in the buffer rather than the ledger
5563 // (session was busy at that check too), the busy leg's own
5564 // `http_settle` flush removes this admission guard when it
5565 // settles; nothing here needs to.
5566 (
5567 StatusCode::ACCEPTED,
5568 Json(serde_json::json!({
5569 "request_id": evaluation.request.request_id,
5570 "decision": evaluation.decision.as_str(),
5571 "state": "steering_queued",
5572 })),
5573 )
5574 .into_response()
5575 }
5576 Admission::Started(taken) => {
5577 // The activity recorded above is durable now (whether it was
5578 // written straight into the ledger or is about to be, via this
5579 // very chain) — the in-memory admission guard can be released.
5580 handle
5581 .admissions
5582 .lock()
5583 .unwrap_or_else(std::sync::PoisonError::into_inner)
5584 .remove(&request_id);
5585 let core = handle.core.clone();
5586 let preview_intent = core.resolve_turn_intent(&taken, &message);
5587 let mut preview_outcome =
5588 vak_intent::OutcomeSpec::from_intent(&body.text, &preview_intent);
5589 preview_outcome.evidence_max_age_secs = Some(core.effective_evidence_max_age_secs());
5590 *handle
5591 .intent
5592 .lock()
5593 .unwrap_or_else(std::sync::PoisonError::into_inner) =
5594 Some(vak_session::types::IntentRecord {
5595 reading: preview_intent.reading,
5596 strands: preview_intent.strands,
5597 engagement: preview_intent.engagement,
5598 provenance: preview_intent.provenance,
5599 outcome: Some(preview_outcome),
5600 model_visible: None,
5601 commitment_id: None,
5602 strand_commitments: Default::default(),
5603 });
5604 wait_for_external_subscriber(&handle).await;
5605 spawn_http_turn_chain(
5606 &state,
5607 handle,
5608 core,
5609 taken,
5610 TurnStart::message(message),
5611 None,
5612 );
5613 (
5614 StatusCode::ACCEPTED,
5615 Json(serde_json::json!({
5616 "request_id": evaluation.request.request_id,
5617 "decision": evaluation.decision.as_str(),
5618 "state": "started",
5619 })),
5620 )
5621 .into_response()
5622 }
5623 }
5624}
5625
5626async fn cancel_run(State(state): State<AppState>, Path(id): Path<String>) -> StatusCode {
5627 let Some(handle) = state.get(&id) else {
5628 return StatusCode::NOT_FOUND;
5629 };
5630 handle
5631 .cancel
5632 .lock()
5633 .unwrap_or_else(std::sync::PoisonError::into_inner)
5634 .cancel();
5635 record_control_activity(&handle, "Run cancelled", "cancel");
5636 deny_pending_approvals(&handle);
5637 // A stop means stop: whatever was queued for a continuation leg is
5638 // discarded rather than silently running as the "next" turn once the
5639 // cancelled run unwinds (finding 1c). Input that arrives AFTER this
5640 // drain but before the run actually unwinds is a fresh push into the
5641 // same queue and is unaffected — it becomes the next leg of the chain,
5642 // same as any other steering.
5643 let discarded = handle.steering.drain(vak_agent::DrainMode::All);
5644 if !discarded.is_empty() {
5645 record_activity_or_buffer(
5646 &handle,
5647 vak_session::ActivityRecord {
5648 activity_id: format!("cancel-discard-{}", uuid::Uuid::now_v7()),
5649 turn: None,
5650 kind: vak_session::ActivityKind::Diagnostic,
5651 status: vak_session::ActivityStatus::Cancelled,
5652 label: "Queued input discarded by stop".into(),
5653 detail: Some(format!(
5654 "{} queued message{} discarded",
5655 discarded.len(),
5656 if discarded.len() == 1 { "" } else { "s" }
5657 )),
5658 data: std::collections::BTreeMap::new(),
5659 },
5660 );
5661 }
5662 // No synthesized `RunFinished` here: the run's own settle path
5663 // (`run_turn_chain`) emits the terminal event once it actually stops.
5664 // Broadcasting one from here raced the real one — cancel_run's
5665 // synthetic event could arrive, mark the chat finished on the client,
5666 // and then the run's genuine `RunFinished` landed afterward and
5667 // un-terminated it.
5668 StatusCode::ACCEPTED
5669}
5670
5671async fn pause_run(State(state): State<AppState>, Path(id): Path<String>) -> StatusCode {
5672 let Some(handle) = state.get(&id) else {
5673 return StatusCode::NOT_FOUND;
5674 };
5675 handle.steering.pause();
5676 record_control_activity(&handle, "Run paused", "pause");
5677 StatusCode::ACCEPTED
5678}
5679
5680async fn resume_run(State(state): State<AppState>, Path(id): Path<String>) -> StatusCode {
5681 let Some(handle) = state.get(&id) else {
5682 return StatusCode::NOT_FOUND;
5683 };
5684 handle.steering.resume();
5685 record_control_activity(&handle, "Run resumed", "resume");
5686 StatusCode::ACCEPTED
5687}
5688
5689async fn control_state(
5690 State(state): State<AppState>,
5691 Path(id): Path<String>,
5692) -> axum::response::Response {
5693 let Some(handle) = state.get(&id) else {
5694 return StatusCode::NOT_FOUND.into_response();
5695 };
5696 let running = handle
5697 .session
5698 .lock()
5699 .map(|session| session.is_none())
5700 .unwrap_or(false);
5701 let revision = handle
5702 .session
5703 .lock()
5704 .ok()
5705 .and_then(|guard| {
5706 guard.as_ref().and_then(|session| {
5707 session.chain_to_root().iter().rev().find_map(|entry| {
5708 if let vak_session::EntryPayload::Intent(record) = &entry.payload {
5709 record.outcome.as_ref().map(|outcome| outcome.revision)
5710 } else {
5711 None
5712 }
5713 })
5714 })
5715 })
5716 .or_else(|| {
5717 handle.intent.lock().ok().and_then(|guard| {
5718 guard
5719 .as_ref()?
5720 .outcome
5721 .as_ref()
5722 .map(|outcome| outcome.revision)
5723 })
5724 })
5725 .unwrap_or(0);
5726 Json(serde_json::json!({
5727 "running": running,
5728 "paused": handle.steering.is_paused(),
5729 "revision": revision,
5730 }))
5731 .into_response()
5732}
5733
5734#[derive(serde::Deserialize)]
5735struct PlanChangeBody {
5736 text: String,
5737 #[serde(default = "default_intervention_source")]
5738 source: String,
5739 #[serde(default)]
5740 target_revision: Option<u64>,
5741}
5742
5743fn apply_plan_change(
5744 mut outcome: vak_intent::OutcomeSpec,
5745 request: &vak_intent::InterventionRequest,
5746) -> (vak_intent::OutcomeSpec, (String, String)) {
5747 let before = outcome
5748 .requirements
5749 .iter()
5750 .map(|requirement| requirement.id.clone())
5751 .collect::<Vec<_>>();
5752 match request.kind {
5753 vak_intent::InterventionKind::AddRequirement => {
5754 outcome.requirements.push(vak_intent::OutcomeRequirement {
5755 id: format!("intervention-{}", request.request_id),
5756 kind: vak_intent::RequirementKind::Deliverable,
5757 description: request.text.clone(),
5758 origin: vak_intent::RequirementOrigin::Explicit,
5759 importance: vak_intent::RequirementImportance::Must,
5760 target: None,
5761 })
5762 }
5763 vak_intent::InterventionKind::RemoveRequirement => {
5764 if let Some(target) = request.text.split_whitespace().last() {
5765 outcome
5766 .requirements
5767 .retain(|requirement| requirement.id != target);
5768 }
5769 }
5770 vak_intent::InterventionKind::Replan | vak_intent::InterventionKind::Reprioritize => {
5771 outcome.assumptions.push(request.text.clone());
5772 }
5773 _ => {}
5774 }
5775 outcome.revision = outcome.revision.saturating_add(1);
5776 let after = outcome
5777 .requirements
5778 .iter()
5779 .map(|requirement| requirement.id.clone())
5780 .collect::<Vec<_>>();
5781 (outcome, (before.join(","), after.join(",")))
5782}
5783
5784async fn plan_change(
5785 State(state): State<AppState>,
5786 Path(id): Path<String>,
5787 Json(body): Json<PlanChangeBody>,
5788) -> axum::response::Response {
5789 let Some(handle) = state.get(&id) else {
5790 return StatusCode::NOT_FOUND.into_response();
5791 };
5792 let (kind, command_text) = intervention_of(&body.text);
5793 if !matches!(
5794 kind,
5795 vak_intent::InterventionKind::Replan
5796 | vak_intent::InterventionKind::Reprioritize
5797 | vak_intent::InterventionKind::AddRequirement
5798 | vak_intent::InterventionKind::RemoveRequirement
5799 ) {
5800 return (StatusCode::BAD_REQUEST, "request is not a plan change").into_response();
5801 }
5802 let revision = handle
5803 .session
5804 .lock()
5805 .ok()
5806 .and_then(|guard| {
5807 guard.as_ref().and_then(|session| {
5808 session.chain_to_root().iter().rev().find_map(|entry| {
5809 if let vak_session::EntryPayload::Intent(record) = &entry.payload {
5810 record.outcome.as_ref().map(|outcome| outcome.revision)
5811 } else {
5812 None
5813 }
5814 })
5815 })
5816 })
5817 .or_else(|| {
5818 handle.intent.lock().ok().and_then(|guard| {
5819 guard
5820 .as_ref()
5821 .and_then(|record| record.outcome.as_ref().map(|outcome| outcome.revision))
5822 })
5823 })
5824 .unwrap_or(0);
5825 if body
5826 .target_revision
5827 .is_some_and(|target| target != revision)
5828 {
5829 return (
5830 StatusCode::CONFLICT,
5831 Json(serde_json::json!({
5832 "decision": "rejected",
5833 "reason": "plan revision is stale",
5834 "revision": revision,
5835 })),
5836 )
5837 .into_response();
5838 }
5839 let evaluation = vak_intent::evaluate_intervention(vak_intent::InterventionRequest {
5840 request_id: format!("plan-change-{}", uuid::Uuid::now_v7()),
5841 kind,
5842 text: command_text,
5843 source: control_source_for(&body.source, state.core.surface().slug(), &id),
5844 target_revision: Some(revision),
5845 target_session_id: Some(id.clone()),
5846 target_parent_session_id: target_parent_of(&handle),
5847 });
5848 let mut admitted_revision = None;
5849 let mut requirement_diff = None;
5850 if evaluation.decision == vak_intent::InterventionDecision::Queued
5851 && let Ok(mut guard) = handle.session.lock()
5852 && let Some(session) = guard.as_mut()
5853 && let Some(record) = session.chain_to_root().iter().rev().find_map(|entry| {
5854 if let vak_session::EntryPayload::Intent(record) = &entry.payload {
5855 Some(record.as_ref())
5856 } else {
5857 None
5858 }
5859 })
5860 && let Some(outcome) = record.outcome.clone()
5861 {
5862 let (outcome, diff) = apply_plan_change(outcome, &evaluation.request);
5863 requirement_diff = Some(diff);
5864 let update = vak_session::types::IntentRecord {
5865 reading: record.reading.clone(),
5866 strands: record.strands.clone(),
5867 engagement: record.engagement.clone(),
5868 provenance: record.provenance.clone(),
5869 outcome: Some(outcome),
5870 model_visible: None,
5871 commitment_id: record.commitment_id.clone(),
5872 strand_commitments: record.strand_commitments.clone(),
5873 };
5874 if session.append_intent(update.clone()).is_ok() {
5875 *handle
5876 .intent
5877 .lock()
5878 .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(update);
5879 admitted_revision = Some(revision + 1);
5880 }
5881 }
5882 if evaluation.decision == vak_intent::InterventionDecision::Queued
5883 && admitted_revision.is_none()
5884 && let Some(mut record) = handle.intent.lock().ok().and_then(|guard| guard.clone())
5885 && let Some(outcome) = record.outcome.take()
5886 {
5887 let (outcome, diff) = apply_plan_change(outcome, &evaluation.request);
5888 requirement_diff = Some(diff);
5889 let update = vak_session::types::IntentRecord {
5890 reading: record.reading,
5891 strands: record.strands,
5892 engagement: record.engagement,
5893 provenance: record.provenance,
5894 outcome: Some(outcome.clone()),
5895 model_visible: None,
5896 commitment_id: record.commitment_id,
5897 strand_commitments: record.strand_commitments,
5898 };
5899 *handle
5900 .intent
5901 .lock()
5902 .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(update.clone());
5903 handle.steering.push_outcome_update(update);
5904 admitted_revision = Some(outcome.revision);
5905 }
5906 record_activity_or_buffer(
5907 &handle,
5908 vak_session::ActivityRecord {
5909 activity_id: evaluation.request.request_id.clone(),
5910 turn: None,
5911 kind: vak_session::ActivityKind::Diagnostic,
5912 status: if evaluation.decision == vak_intent::InterventionDecision::RequiresHuman {
5913 vak_session::ActivityStatus::Denied
5914 } else {
5915 vak_session::ActivityStatus::Pending
5916 },
5917 label: "Plan change evaluated".into(),
5918 detail: Some(evaluation.reason.clone()),
5919 data: std::collections::BTreeMap::from([
5920 ("decision".into(), evaluation.decision.as_str().into()),
5921 ("source".into(), body.source),
5922 (
5923 "revision".into(),
5924 admitted_revision.unwrap_or(revision).to_string(),
5925 ),
5926 (
5927 "requirements_before".into(),
5928 requirement_diff
5929 .as_ref()
5930 .map(|diff| diff.0.clone())
5931 .unwrap_or_else(String::new),
5932 ),
5933 (
5934 "requirements_after".into(),
5935 requirement_diff
5936 .as_ref()
5937 .map(|diff| diff.1.clone())
5938 .unwrap_or_else(String::new),
5939 ),
5940 ("request".into(), body.text),
5941 ]),
5942 },
5943 );
5944 let status = if evaluation.decision == vak_intent::InterventionDecision::RequiresHuman {
5945 StatusCode::CONFLICT
5946 } else {
5947 handle
5948 .steering
5949 .push_steering(evaluation.request.text.clone());
5950 StatusCode::ACCEPTED
5951 };
5952 (
5953 status,
5954 Json(serde_json::json!({
5955 "request_id": evaluation.request.request_id,
5956 "decision": evaluation.decision.as_str(),
5957 "revision": admitted_revision.unwrap_or(revision),
5958 "reason": evaluation.reason,
5959 })),
5960 )
5961 .into_response()
5962}
5963
5964pub(crate) fn record_control_activity(handle: &SessionHandle, label: &str, control: &str) {
5965 record_activity_or_buffer(
5966 handle,
5967 vak_session::ActivityRecord {
5968 activity_id: format!("control-{}", uuid::Uuid::now_v7()),
5969 turn: None,
5970 kind: vak_session::ActivityKind::Diagnostic,
5971 status: vak_session::ActivityStatus::Succeeded,
5972 label: label.into(),
5973 detail: Some("operator control-plane request".into()),
5974 data: std::collections::BTreeMap::from([
5975 ("control".into(), control.into()),
5976 ("source".into(), "human".into()),
5977 ]),
5978 },
5979 );
5980}
5981
5982fn record_activity_or_buffer(handle: &SessionHandle, activity: vak_session::ActivityRecord) {
5983 if let Ok(mut session) = handle.session.lock()
5984 && let Some(session) = session.as_mut()
5985 {
5986 let _ = session.append_activity(activity);
5987 } else if let Ok(mut activities) = handle.activity_buffer.lock() {
5988 activities.push(activity);
5989 }
5990}
5991
5992fn record_routing_data(
5993 data: &mut std::collections::BTreeMap<String, String>,
5994 routing: &RoutingEnvelope,
5995) {
5996 if let Some(value) = routing.message_id.as_deref() {
5997 data.insert("message_id".into(), value.into());
5998 }
5999 if let Some(value) = routing.conversation_id.as_deref() {
6000 data.insert("conversation_id".into(), value.into());
6001 }
6002 if let Some(value) = routing.target_work_id.as_deref() {
6003 data.insert("target_work_id".into(), value.into());
6004 }
6005 if let Some(value) = routing.target_result_id.as_deref() {
6006 data.insert("target_result_id".into(), value.into());
6007 }
6008 if let Some(value) = routing.relation.as_deref() {
6009 data.insert("relation".into(), value.into());
6010 }
6011 if let Some(value) = routing.outcome_revision {
6012 data.insert("outcome_revision".into(), value.to_string());
6013 }
6014 if let Some(value) = routing.provenance.as_deref() {
6015 data.insert("routing_provenance".into(), value.into());
6016 }
6017}
6018
6019/// The session that dispatched this one, from its own header: what "an agent
6020/// may control only its own children" is checked against. Unknown while the
6021/// ledger is out on a running turn, and unknown refuses an agent's control.
6022fn target_parent_of(handle: &SessionHandle) -> Option<String> {
6023 handle.session.lock().ok().and_then(|guard| {
6024 guard.as_ref().and_then(|log| {
6025 log.header()
6026 .and_then(|header| header.parent_session_id.clone())
6027 })
6028 })
6029}
6030
6031fn routing_revision_is_current(handle: &SessionHandle, expected: u64) -> bool {
6032 handle.intent.lock().ok().is_some_and(|record| {
6033 record
6034 .as_ref()
6035 .and_then(|intent| intent.outcome.as_ref())
6036 .is_some_and(|outcome| outcome.revision == expected)
6037 })
6038}
6039
6040fn routing_summary(routing: Option<&RoutingEnvelope>) -> String {
6041 let Some(routing) = routing else {
6042 return String::new();
6043 };
6044 let mut data = std::collections::BTreeMap::new();
6045 record_routing_data(&mut data, routing);
6046 serde_json::to_string(&data).unwrap_or_default()
6047}
6048
6049fn deny_pending_approvals(handle: &SessionHandle) {
6050 let requests: Vec<ApprovalRequest> = handle
6051 .pending
6052 .lock()
6053 .unwrap_or_else(std::sync::PoisonError::into_inner)
6054 .drain()
6055 .map(|(_, request)| request)
6056 .collect();
6057 for request in requests {
6058 request.respond(false);
6059 }
6060}
6061
6062// ---- Worker control plane -------------------------------------------------
6063//
6064// Children already stream lifecycle/tool events into the parent session's
6065// SSE channel; these endpoints add the missing half: listing, steering, and
6066// stopping from a remote surface. Scope-checked against the parent so one
6067// session can never touch another's child.
6068
6069async fn list_workers(
6070 State(state): State<AppState>,
6071 Path(id): Path<String>,
6072) -> Json<serde_json::Value> {
6073 let children = state.core.workers().active_for(&id);
6074 Json(serde_json::json!({ "workers": children }))
6075}
6076
6077#[derive(serde::Deserialize)]
6078struct WorkerSteerBody {
6079 text: String,
6080}
6081
6082async fn steer_worker(
6083 State(state): State<AppState>,
6084 Path((id, child)): Path<(String, String)>,
6085 Json(body): Json<WorkerSteerBody>,
6086) -> StatusCode {
6087 if state.core.workers().parent_of(&child).as_deref() != Some(id.as_str()) {
6088 return StatusCode::NOT_FOUND;
6089 }
6090 if state.core.workers().steer(&child, &body.text) {
6091 StatusCode::ACCEPTED
6092 } else {
6093 StatusCode::CONFLICT
6094 }
6095}
6096
6097async fn stop_worker(
6098 State(state): State<AppState>,
6099 Path((id, child)): Path<(String, String)>,
6100) -> StatusCode {
6101 if state.core.workers().parent_of(&child).as_deref() != Some(id.as_str()) {
6102 return StatusCode::NOT_FOUND;
6103 }
6104 if state.core.workers().stop(&child) {
6105 StatusCode::ACCEPTED
6106 } else {
6107 StatusCode::CONFLICT
6108 }
6109}
6110
6111#[derive(serde::Deserialize)]
6112struct ApprovalBody {
6113 approve: bool,
6114 /// "…and don't ask again for calls like this one". Derives the narrowest
6115 /// rule that covers this call and persists it to
6116 /// `.vak/permissions.local.toml`. Only meaningful alongside
6117 /// `approve: true` — remembering a refusal would be a deny rule, which
6118 /// is a different and much heavier decision than answering one gate.
6119 #[serde(default)]
6120 remember: bool,
6121}
6122
6123#[derive(serde::Deserialize)]
6124struct OutcomeReviewBody {
6125 verdict: String,
6126 #[serde(default)]
6127 turn: Option<usize>,
6128 #[serde(default)]
6129 note: Option<String>,
6130}
6131
6132/// Record an operator's review of a computed outcome. This is deliberately
6133/// separate from approval: reviewing a result never authorizes a tool call.
6134async fn record_outcome_review(
6135 State(state): State<AppState>,
6136 Path(id): Path<String>,
6137 Json(body): Json<OutcomeReviewBody>,
6138) -> axum::response::Response {
6139 use axum::response::IntoResponse;
6140 if !matches!(
6141 body.verdict.as_str(),
6142 "accepted" | "needs_work" | "rejected"
6143 ) {
6144 return (StatusCode::BAD_REQUEST, "invalid outcome review verdict").into_response();
6145 }
6146 let Some(handle) = state.get(&id) else {
6147 return StatusCode::NOT_FOUND.into_response();
6148 };
6149 let Ok(mut guard) = handle.session.lock() else {
6150 return StatusCode::CONFLICT.into_response();
6151 };
6152 let Some(session) = guard.as_mut() else {
6153 return StatusCode::NOT_FOUND.into_response();
6154 };
6155 let reviewed_turn = body.turn.or_else(|| {
6156 session
6157 .chain_to_root()
6158 .iter()
6159 .filter_map(|entry| match &entry.payload {
6160 vak_session::EntryPayload::Activity(activity)
6161 if activity.label == "Outcome evaluation" =>
6162 {
6163 activity.turn
6164 }
6165 _ => None,
6166 })
6167 .next_back()
6168 });
6169 if body.turn.is_some() && reviewed_turn.is_none() {
6170 return (StatusCode::NOT_FOUND, "outcome evaluation turn not found").into_response();
6171 }
6172 let activity = vak_session::ActivityRecord {
6173 activity_id: format!("outcome-review-{}", uuid::Uuid::now_v7()),
6174 turn: reviewed_turn,
6175 kind: vak_session::ActivityKind::Diagnostic,
6176 status: vak_session::ActivityStatus::Succeeded,
6177 label: "Outcome review".into(),
6178 detail: body.note,
6179 data: std::collections::BTreeMap::from([("verdict".into(), body.verdict)]),
6180 };
6181 if let Err(error) = session.append_activity(activity) {
6182 return (StatusCode::INTERNAL_SERVER_ERROR, error.to_string()).into_response();
6183 }
6184 Json(serde_json::json!({ "recorded": true })).into_response()
6185}
6186
6187/// `POST /sessions/{id}/approvals/{req_id}` — answer one gate.
6188///
6189/// `remember` is the mechanism `docs/design/08-permissions.md` has described
6190/// since the engine shipped and nothing ever called: `Core::learn_allow_rule`
6191/// existed, was tested, wrote a validated file, and had no caller on any
6192/// surface, because the interactive approver its comment referred to was
6193/// never built. This is that caller.
6194///
6195/// Remembering never blocks the answer. The gate is resolved first; a
6196/// failure to derive or persist a rule is reported alongside a successful
6197/// approval, because the run is already waiting and a bookkeeping problem
6198/// must not become a denial.
6199async fn answer_approval(
6200 State(state): State<AppState>,
6201 Path((id, req_id)): Path<(String, String)>,
6202 Json(body): Json<ApprovalBody>,
6203) -> axum::response::Response {
6204 use axum::response::IntoResponse;
6205 // Look the request up in THIS session's pending map only: approvals
6206 // are never resolvable across sessions.
6207 let Some(handle) = state.get(&id) else {
6208 return StatusCode::NOT_FOUND.into_response();
6209 };
6210 let Some(req) = handle
6211 .pending
6212 .lock()
6213 .unwrap_or_else(std::sync::PoisonError::into_inner)
6214 .remove(&req_id)
6215 else {
6216 return StatusCode::NOT_FOUND.into_response();
6217 };
6218 let tool = req.tool.clone();
6219 let args_json = req.args_json.clone();
6220 *req.answered_by
6221 .lock()
6222 .unwrap_or_else(std::sync::PoisonError::into_inner) =
6223 Some(("operator".into(), "You".into()));
6224 req.respond(body.approve);
6225
6226 let mut learned: Option<String> = None;
6227 let mut learn_error: Option<String> = None;
6228 if body.remember && body.approve {
6229 match serde_json::from_str::<serde_json::Value>(&args_json) {
6230 // The gate's own core, not the server's: a learned rule belongs
6231 // to the workspace whose call raised it.
6232 Ok(args) => match handle.core.learn_from_call(&tool, &args) {
6233 Ok(spec) => {
6234 vak_core::security_events::record(
6235 &state.core.sessions_home(),
6236 vak_core::security_events::EventKind::ConfigChange,
6237 "permission_rule_learned",
6238 &format!("session={id} rule={spec}"),
6239 None,
6240 );
6241 learned = Some(spec);
6242 }
6243 Err(error) => learn_error = Some(error.to_string()),
6244 },
6245 Err(error) => learn_error = Some(format!("unreadable tool arguments: {error}")),
6246 }
6247 }
6248 (
6249 StatusCode::OK,
6250 Json(serde_json::json!({
6251 "approved": body.approve,
6252 "learned_rule": learned,
6253 "learn_error": learn_error,
6254 })),
6255 )
6256 .into_response()
6257}
6258
6259/// Dispatch forensics (docs/design/42-managed-work-contracts.md): the session's work
6260/// receipts, newest last.
6261async fn session_receipts(
6262 State(state): State<AppState>,
6263 Path(id): Path<String>,
6264) -> Result<Json<Vec<vak_llm::WorkReceipt>>, StatusCode> {
6265 let Some(handle) = state.get(&id) else {
6266 return Err(StatusCode::NOT_FOUND);
6267 };
6268 let Ok(session) = handle.session.lock() else {
6269 return Err(StatusCode::NOT_FOUND);
6270 };
6271 match session.as_ref() {
6272 Some(log) => Ok(Json(log.receipts().into_iter().cloned().collect())),
6273 None => Err(StatusCode::NOT_FOUND),
6274 }
6275}
6276
6277async fn session_work(
6278 State(state): State<AppState>,
6279 Path(id): Path<String>,
6280) -> Result<Json<Option<vak_session::work::WorkProjection>>, StatusCode> {
6281 if let Some(handle) = state.get(&id) {
6282 let guard = handle
6283 .session
6284 .lock()
6285 .unwrap_or_else(std::sync::PoisonError::into_inner);
6286 let Some(session) = guard.as_ref() else {
6287 return Err(StatusCode::NOT_FOUND);
6288 };
6289 return session
6290 .work_projection()
6291 .map(Json)
6292 .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR);
6293 }
6294 let Some(session) = open_historical_session(&state, &id) else {
6295 return Err(StatusCode::NOT_FOUND);
6296 };
6297 session
6298 .work_projection()
6299 .map(Json)
6300 .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
6301}
6302
6303#[derive(serde::Deserialize)]
6304#[serde(tag = "operation", rename_all = "snake_case")]
6305enum WorkCommand {
6306 Transition {
6307 item_id: String,
6308 to: vak_session::types::WorkItemStatus,
6309 #[serde(default)]
6310 reason: String,
6311 },
6312 Cancel {
6313 #[serde(default)]
6314 reason: String,
6315 },
6316 Resume {
6317 #[serde(default)]
6318 reason: String,
6319 },
6320 Retry {
6321 item_id: String,
6322 #[serde(default)]
6323 reason: String,
6324 },
6325 ResolveAssumption {
6326 assumption_id: String,
6327 resolution: String,
6328 },
6329 AttachEvidence {
6330 item_id: String,
6331 evidence: vak_session::types::EvidenceRef,
6332 },
6333 Assign {
6334 item_id: String,
6335 owner: vak_session::types::WorkOwner,
6336 child_session_id: Option<String>,
6337 },
6338 Revise {
6339 contract: vak_session::types::WorkContract,
6340 #[serde(default)]
6341 reason: String,
6342 },
6343}
6344
6345#[derive(serde::Deserialize)]
6346struct WorkReason {
6347 #[serde(default)]
6348 reason: String,
6349}
6350
6351#[derive(serde::Deserialize)]
6352struct WorkReassign {
6353 owner: vak_session::types::WorkOwner,
6354 #[serde(default)]
6355 child_session_id: Option<String>,
6356}
6357
6358async fn session_work_confirm(
6359 state: State<AppState>,
6360 path: Path<String>,
6361 body: Option<Json<WorkReason>>,
6362) -> axum::response::Response {
6363 session_work_command(
6364 state,
6365 path,
6366 Json(WorkCommand::Resume {
6367 reason: body
6368 .map(|body| body.0.reason)
6369 .unwrap_or_else(|| "confirmed by operator".into()),
6370 }),
6371 )
6372 .await
6373}
6374
6375async fn session_work_revise(
6376 state: State<AppState>,
6377 path: Path<String>,
6378 Json(body): Json<WorkCommand>,
6379) -> axum::response::Response {
6380 session_work_command(state, path, Json(body)).await
6381}
6382
6383async fn session_work_retry(
6384 state: State<AppState>,
6385 Path((id, item_id)): Path<(String, String)>,
6386 body: Option<Json<WorkReason>>,
6387) -> axum::response::Response {
6388 session_work_command(
6389 state,
6390 Path(id),
6391 Json(WorkCommand::Retry {
6392 item_id,
6393 reason: body
6394 .map(|body| body.0.reason)
6395 .unwrap_or_else(|| "retry requested by operator".into()),
6396 }),
6397 )
6398 .await
6399}
6400
6401async fn session_work_cancel_item(
6402 state: State<AppState>,
6403 Path((id, item_id)): Path<(String, String)>,
6404 body: Option<Json<WorkReason>>,
6405) -> axum::response::Response {
6406 session_work_command(
6407 state,
6408 Path(id),
6409 Json(WorkCommand::Transition {
6410 item_id,
6411 to: vak_session::types::WorkItemStatus::Cancelled,
6412 reason: body
6413 .map(|body| body.0.reason)
6414 .unwrap_or_else(|| "cancelled by operator".into()),
6415 }),
6416 )
6417 .await
6418}
6419
6420async fn session_work_reassign(
6421 state: State<AppState>,
6422 Path((id, item_id)): Path<(String, String)>,
6423 Json(body): Json<WorkReassign>,
6424) -> axum::response::Response {
6425 session_work_command(
6426 state,
6427 Path(id),
6428 Json(WorkCommand::Assign {
6429 item_id,
6430 owner: body.owner,
6431 child_session_id: body.child_session_id,
6432 }),
6433 )
6434 .await
6435}
6436
6437/// Apply an operator/user work action to the same append-only ledger used by
6438/// the agent. The current projection supplies both the expected state and
6439/// revision, so stale clients receive a conflict instead of silently
6440/// overwriting a newer action.
6441async fn session_work_command(
6442 State(state): State<AppState>,
6443 Path(id): Path<String>,
6444 Json(command): Json<WorkCommand>,
6445) -> axum::response::Response {
6446 use axum::response::IntoResponse;
6447 let Some(handle) = state.get(&id) else {
6448 return StatusCode::NOT_FOUND.into_response();
6449 };
6450 let mut guard = handle
6451 .session
6452 .lock()
6453 .unwrap_or_else(std::sync::PoisonError::into_inner);
6454 let Some(session) = guard.as_mut() else {
6455 return (
6456 StatusCode::CONFLICT,

Lib

Indexing the workspace…

Lib