diff --git a/src/api/handlers.rs b/src/api/handlers.rs index c3ef503..84d141a 100644 --- a/src/api/handlers.rs +++ b/src/api/handlers.rs @@ -1,6 +1,5 @@ use crate::api::models::*; use crate::server::SouveraineServer; -use crate::core::session::{ContentBlock, ConversationMessage}; use axum::{ extract::{Path, Query, State, WebSocketUpgrade, ws::WebSocket}, response::{Json, Sse}, @@ -191,145 +190,78 @@ async fn handle_conversation_stream( conversation_id: String, tx: mpsc::Sender, ) -> anyhow::Result<()> { - let session = server.sessions.get(&conversation_id) - .ok_or_else(|| anyhow::anyhow!("Session disappeared"))?; + use crate::backend::BackendEvent; + use tokio_util::sync::CancellationToken; - let agent_id = session.agent_id.clone(); - let messages: Vec<_> = session.messages.iter().map(|m| { - let role = match m.role { - crate::core::session::MessageRole::System => "system", - crate::core::session::MessageRole::User => "user", - crate::core::session::MessageRole::Assistant => "assistant", - crate::core::session::MessageRole::Tool => "tool", + let event_bus = server.event_bus.clone(); + let empty_interject: crate::backend::InterjectionQueue = + std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let cancel = CancellationToken::new(); + + let (be_tx, mut be_rx) = mpsc::channel::>(64); + let server_clone = server.clone(); + let conv = conversation_id.clone(); + + tokio::spawn(async move { + let _ = crate::server::turn::run_turn( + server_clone, conv, &be_tx, event_bus, + cancel, empty_interject, + ).await; + }); + + // Drain the turn's BackendEvent stream and map to SSE events. + // run_turn already handles: message storage, surficing injection, + // EventBus dispatch, N+1 pass — none of that is needed here. + while let Some(result) = be_rx.recv().await { + let be = match result { + Ok(be) => be, + Err(_) => break, }; - - let has_images = m.blocks.iter().any(|b| matches!(b, ContentBlock::Image { .. })); - - if has_images { - use crate::bridge::bifrost::{ContentPart, ImageUrlSource}; - let parts: Vec = m.blocks.iter().filter_map(|b| match b { - ContentBlock::Text { text } => Some(ContentPart::Text { text: text.clone() }), - ContentBlock::Image { media_type, data } => { - let url = format!("data:{media_type};base64,{data}"); - Some(ContentPart::ImageUrl { image_url: ImageUrlSource { url } }) - } - _ => None, - }).collect(); - let text_content = m.blocks.iter().filter_map(|b| match b { - ContentBlock::Text { text } => Some(text.as_str()), - _ => None, - }).collect::>().join("\n"); - crate::bridge::bifrost::Message::multimodal_user(text_content, parts) - } else { - let content = m.blocks.first().map(|b| match b { - ContentBlock::Text { text } => text.clone(), - _ => String::new(), - }).unwrap_or_default(); - crate::bridge::bifrost::Message::text(role, content) - } - }).collect(); - drop(session); - - // Get agent config - let agent = server.agents.get(&agent_id).await?; - let model = agent.llm_config.model.clone(); - - // Simple Bifrost call (no tool loop for now - that requires git components) - let req = crate::bridge::bifrost::ChatCompletionRequest { - model: model.clone(), - messages, - stream: Some(false), - max_tokens: None, - temperature: agent.llm_config.temperature, - tools: None, - }; - - match server.bifrost.chat_completion(req).await { - Ok(response) => { - let content = response.content.clone(); - - // Stream the response in chunks - for chunk in content.chars().collect::>().chunks(10) { - let chunk_str: String = chunk.iter().collect(); - let _ = tx.send(StreamEvent::AssistantMessage { content: chunk_str }).await; - tokio::time::sleep(tokio::time::Duration::from_millis(20)).await; + let sse_event = match be { + BackendEvent::Token(content) => StreamEvent::AssistantMessage { content }, + BackendEvent::Reasoning(content) => StreamEvent::ReasoningMessage { content }, + BackendEvent::Surfacing { source, content, priority } => { + StreamEvent::Surfacing { source, content, priority } } - - // Store assistant response in session - let _ = server.sessions.add_message( - &conversation_id, - ConversationMessage::assistant_text(&content) - ); - - // Run consciousness events. Snapshot the session first — holding - // a live ref across the N+1 pass would block concurrent writes. - let (n1_agent, n1_turn_count, n1_messages) = { - let session = server.sessions.get(&conversation_id) - .ok_or_else(|| anyhow::anyhow!("Session disappeared"))?; - (session.agent_id.clone(), session.turn_count, session.messages.clone()) - }; - let events = server.consciousness - .on_response(&n1_agent, n1_turn_count, &n1_messages, &content) - .await?; - - // Inject surfacing events back into the session as system messages - // so the agent sees them in its context window on the next turn. - for event in &events { - if let crate::server::ConsciousnessEvent::Surfacing { source, content, priority } = event { - let msg = crate::core::session::ConversationMessage { - role: crate::core::session::MessageRole::System, - blocks: vec![crate::core::session::ContentBlock::Text { - text: format!("[surfacing: {}] {} — {}", source, content, priority), - }], - usage: None, - timestamp: None, - }; - let _ = server.sessions.add_message(&conversation_id, msg); + BackendEvent::Reflection(content) => StreamEvent::Reflection { content }, + BackendEvent::Archivist { synthesis, pressure } => { + StreamEvent::Archivist { synthesis, pressure } + } + BackendEvent::CompactionWarning { pressure, tier } => { + StreamEvent::Archivist { + synthesis: format!("compaction warning tier {} at {:.0}%", tier, pressure * 100.0), + pressure, } } - - for event in events { - let stream_event = match &event { - crate::server::ConsciousnessEvent::Surfacing { source, content, priority } => { - StreamEvent::Surfacing { source: source.clone(), content: content.clone(), priority: priority.clone() } - } - crate::server::ConsciousnessEvent::Reflection { content } => { - StreamEvent::Reflection { content: content.clone() } - } - crate::server::ConsciousnessEvent::Archivist { synthesis, pressure } => { - StreamEvent::Archivist { synthesis: synthesis.clone(), pressure: *pressure } - } - crate::server::ConsciousnessEvent::CompactionWarning { pressure, tier } => { - StreamEvent::Archivist { - synthesis: format!("compaction warning tier {} at {:.0}%", tier, pressure * 100.0), - pressure: *pressure, - } - } - }; - let _ = tx.send(stream_event).await; + BackendEvent::ToolCall { id, name, arguments, .. } => { + StreamEvent::ToolCallMessage { + tool_call: ToolCall { + id, + function: ToolFunction { name, arguments }, + }, + } } - - // Update pressure - let mut session = server.sessions.get_mut(&conversation_id) - .ok_or_else(|| anyhow::anyhow!("Session disappeared"))?; - let pressure = server.consciousness.pressure_for_session(&session).await; - session.context_pressure = pressure; - drop(session); - } - Err(e) => { - eprintln!("Bifrost error: {}", e); - let _ = tx.send(StreamEvent::AssistantMessage { - content: format!("Error: {}", e) - }).await; + BackendEvent::ToolResult { output, is_error, .. } => { + StreamEvent::ToolReturnMessage { + tool_return: ToolReturn { + status: if is_error { "error".into() } else { "success".into() }, + output, + }, + } + } + // Silently skip events that have no SSE counterpart + _ => continue, + }; + if tx.send(sse_event).await.is_err() { + break; } } - // Send ping at end let _ = tx.send(StreamEvent::Ping).await; - Ok(()) } + // ─── Memory (memfs HTTP write path) ─────────────────────────────────────── // // Replaces Letta's PATCH /v1/blocks/{id} for the cron-into-memfs pattern. diff --git a/src/api/models.rs b/src/api/models.rs index 0e805fc..dc021f6 100644 --- a/src/api/models.rs +++ b/src/api/models.rs @@ -50,9 +50,14 @@ pub struct LlmConfig { /// When false, images are stripped to text markers before sending. #[serde(default = "default_supports_images")] pub supports_images: bool, + /// How many tool rounds between subconscious mid-turn checkpoints. + /// 0 disables checkpointing entirely. + #[serde(default = "default_checkpoint_interval")] + pub checkpoint_interval: u32, } fn default_supports_images() -> bool { true } +fn default_checkpoint_interval() -> u32 { 10 } fn default_context_window() -> u32 { 128000 diff --git a/src/backend/local/mod.rs b/src/backend/local/mod.rs index 9a4f461..430b3b3 100644 --- a/src/backend/local/mod.rs +++ b/src/backend/local/mod.rs @@ -8,69 +8,24 @@ //! This is the "harness still works when the server is gone" path //! (`souveraine chat --local`, or auto-fallback when the remote is down). -pub(crate) mod energy; -mod turn; -mod consciousness; -mod subagent; - -pub use subagent::LocalSubagentRunner; +pub(crate) mod consciousness; use anyhow::{Context, Result}; use async_trait::async_trait; use futures::stream::{BoxStream, StreamExt}; use std::sync::Arc; -use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; -use std::time::{Duration, Instant}; +use std::sync::atomic::{AtomicU32, Ordering}; +use std::time::Duration; use tokio::sync::mpsc; use tokio_stream::wrappers::ReceiverStream; use tokio_util::sync::CancellationToken; -use crate::bridge::bifrost::{ChatCompletionRequest, Message as BifrostMessage}; -use crate::bridge::model_router::TokenCounter; use crate::core::session::{ContentBlock, ConversationMessage, ImageAttachment, MessageRole}; use crate::core::config::ConsciousnessConfig; -use crate::core::identity::SeedId; -use crate::core::nervous::EventBus; -use crate::core::tools::defs::{SubagentParams, SubagentRunner, ToolContext}; -use crate::server::{ConsciousnessEvent, SouveraineServer}; +use crate::server::SouveraineServer; use super::{AgentInfo, Backend, BackendEvent, ConversationInfo}; -/// Below 95%: no cap. At 95%+: scale max_tokens so context + output -/// stays under the model's limit. The agent feels the room shrink. -fn pressure_to_max_tokens(pressure: f32, output_limit: u32) -> Option { - if pressure <= 0.95 { - return None; - } - let remaining = (1.0 - pressure) / 0.05; - let ratio = remaining.max(0.0).min(1.0); - Some((output_limit as f32 * ratio) as u32) -} - -/// Mirror of `ConsciousnessEngine::calculate_pressure` for the in-loop -/// BifrostMessage shape, so we can recompute pressure as tool results -/// accumulate inside a single turn. `context_limit` comes from the -/// agent's `llm_config.context_window` (Constitution V.3 — per-model -/// physics, no hardcoded 128K). -fn bifrost_pressure(counter: &TokenCounter, messages: &[BifrostMessage], context_limit: usize) -> f32 { - let tokens: usize = messages.iter().map(|m| counter.count(&m.content.as_text())).sum(); - let limit = context_limit.max(1); - (tokens as f32 / limit as f32).min(1.0) -} - -/// Helper: bump adaptive delay when we hit a 429. No decay — once bumped, -/// the delay stays at that level until the app restarts. -fn bump_on_strain(delay: &AtomicU64, status: u16) { - if status == 429 { - let current = delay.load(Ordering::Relaxed); - let bumped = (current + 200).min(3000); - if bumped > current { - delay.store(bumped, Ordering::Relaxed); - tracing::info!("rate delay bumped to {}ms (429)", bumped); - } - } -} - #[derive(Clone)] pub struct LocalBackend { server: Arc, @@ -476,12 +431,12 @@ impl Backend for LocalBackend { let (tx, rx) = mpsc::channel::>(64); let server = self.server.clone(); let conv_id = conversation_id.to_string(); - let event_bus = self.event_bus.clone(); + let event_bus = server.event_bus.clone(); let active = self.active_sessions.clone(); active.fetch_add(1, Ordering::Relaxed); tokio::spawn(async move { - if let Err(e) = turn::run_turn(server, conv_id, &tx, event_bus, cancel, interject).await { + if let Err(e) = crate::server::turn::run_turn(server, conv_id, &tx, event_bus, cancel, interject).await { let _ = tx.send(Err(e)).await; } let _ = tx.send(Ok(BackendEvent::Done)).await; @@ -538,12 +493,12 @@ impl Backend for LocalBackend { let (tx, rx) = mpsc::channel::>(64); let server = self.server.clone(); let conv_id = conversation_id.to_string(); - let event_bus = self.event_bus.clone(); + let event_bus = server.event_bus.clone(); let active = self.active_sessions.clone(); active.fetch_add(1, Ordering::Relaxed); tokio::spawn(async move { - if let Err(e) = turn::run_turn(server, conv_id, &tx, event_bus, cancel, interject).await { + if let Err(e) = crate::server::turn::run_turn(server, conv_id, &tx, event_bus, cancel, interject).await { let _ = tx.send(Err(e)).await; } let _ = tx.send(Ok(BackendEvent::Done)).await; @@ -567,6 +522,7 @@ impl Backend for LocalBackend { max_tool_rounds: current.llm_config.max_tool_rounds, inter_round_delay_ms: current.llm_config.inter_round_delay_ms, supports_images: current.llm_config.supports_images, + checkpoint_interval: current.llm_config.checkpoint_interval, }), memory_blocks: None, tools: None, diff --git a/src/backend/mod.rs b/src/backend/mod.rs index dc09763..b140b3c 100644 --- a/src/backend/mod.rs +++ b/src/backend/mod.rs @@ -108,6 +108,10 @@ pub enum BackendEvent { }, /// Agent set an atmospheric preset for the UI chrome. Atmosphere(String), + + /// Agent set or advanced the itinerary. The string is the route-line + /// representation for the header strip. + Itinerary(String), /// N+1 subconscious pass started (`true`) or finished (`false`). /// Presence reads this to flip into / out of `Posture::Thinking` so the /// face shows when the subconscious is the one looking at the conversation. diff --git a/src/core/prompt.rs b/src/core/prompt.rs index e61e2c6..4e67ea9 100644 --- a/src/core/prompt.rs +++ b/src/core/prompt.rs @@ -445,6 +445,28 @@ pub async fn build_system_prompt_full( } } + // 5a₂. Itinerary prompt — if I have live commitments and no active + // itinerary, remind me I can lay one out so Casey sees where I am. + let dynamic_dir = memory_root.join("system").join("dynamic"); + let itin_path = dynamic_dir.join("itinerary.md"); + let has_itin = tokio::fs::try_exists(&itin_path).await.unwrap_or(false) + && tokio::fs::read_to_string(&itin_path).await + .ok() + .map(|s| s.contains("current:")) + .unwrap_or(false); + + if !has_itin { + let tasks_dir = memory_root.join("tasks"); + let live_count = count_live_todos(&tasks_dir); + if live_count > 0 && live_count <= 12 { + sections.push(format!( + "I have {} live commitments — I could use `itinerary` to lay them \ + out for Casey if now is the time for a route.", + live_count, + )); + } + } + // 5b. Subconscious channel — name the inner-voice file, pending inbox, // and (when reachable) a glimpse of the subconscious's ledger. let subconscious_channel = build_subconscious_channel(memory_root, subconscious_root).await; @@ -767,6 +789,22 @@ async fn build_synthesis_orientation(memory_root: &Path) -> String { ) } +/// Count live (pending / in_progress) todo files in the tasks directory. +fn count_live_todos(tasks_dir: &std::path::Path) -> usize { + let Ok(entries) = std::fs::read_dir(tasks_dir) else { return 0 }; + entries + .flatten() + .filter(|e| e.path().extension().map(|e| e == "md").unwrap_or(false)) + .filter(|e| { + std::fs::read_to_string(e.path()) + .ok() + .map(|s| s.contains("status: pending") || s.contains("status: in_progress")) + .unwrap_or(false) + }) + .take(13) + .count() +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/core/tools/itinerary.rs b/src/core/tools/itinerary.rs new file mode 100644 index 0000000..b2417c3 --- /dev/null +++ b/src/core/tools/itinerary.rs @@ -0,0 +1,680 @@ +//! Itinerary — the agent's route through the current work. +//! +//! An itinerary is an ordered sequence of stops. Each stop optionally +//! references a todo commitment by its ID. Stops that reference a todo +//! inherit the todo's nature, energy, and thread; stops without a +//! reference are ephemeral waypoints that leave no trace in the task +//! system. +//! +//! The itinerary is persisted to `system/dynamic/itinerary.md` in the +//! agent's memfs — the same volatile workspace that holds energy balance, +//! context pressure, and other running-state files. It is not a commitment +//! store; it is the *active face* of whatever the agent is doing right now. +//! +//! ## Agent actions +//! +//! - `set` — lay out a new route (replaces current) +//! - `advance` — mark current stop done, move to next +//! - `describe` — read back the current itinerary +//! - `clear` — dismiss the itinerary entirely + +use async_trait::async_trait; +use serde::{Deserialize, Serialize}; +use serde_json::Value as JsonValue; +use std::path::{Path, PathBuf}; + +use super::defs::{Tool, ToolContext, ToolError, ToolOutput}; + +// ── Data types ──────────────────────────────────────────────────── + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub enum StopStatus { + #[serde(rename = "pending")] + Pending, + #[serde(rename = "current")] + Current, + #[serde(rename = "done")] + Done, +} + +impl StopStatus { + pub fn is_live(&self) -> bool { + matches!(self, Self::Pending | Self::Current) + } +} + +/// A single stop on the itinerary. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Stop { + /// Short label shown in the header strip. + pub name: String, + + /// Optional longer description (for `describe` output). + pub description: Option, + + /// If set, this stop links back to a todo commitment by its id. + /// The tool reads the todo's nature/energy from the file and surfaces + /// them to the agent and UI. + pub todo_id: Option, + + pub status: StopStatus, + + /// Derived from the referenced todo, if any. Populated at read time. + #[serde(skip)] + pub nature: Option, + + /// Derived from the referenced todo, if any. + #[serde(skip)] + pub energy: Option, +} + +/// The full itinerary, stored as YAML frontmatter in a markdown file. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Itinerary { + /// A short title for the whole route. + pub title: String, + + /// Ordered stops. + pub stops: Vec, + + /// Index of the current stop (0-based). Persisted so resume works. + pub current: usize, +} + +impl Itinerary { + pub fn is_active(&self) -> bool { + self.stops.iter().any(|s| s.status.is_live()) + } + + /// Return a compact one-line representation of the route for the + /// description output and the header strip. + pub fn route_line(&self) -> String { + let mut parts: Vec = Vec::new(); + for stop in self.stops.iter() { + let glyph = match stop.status { + StopStatus::Done => "✓", + StopStatus::Current => "●", + StopStatus::Pending => "○", + }; + let mut label = format!("{} {}", glyph, stop.name); + if let Some(ref e) = stop.energy { + let icon = match e.as_str() { + "generative" => "⚡", + "consumptive" => "—", + _ => "", + }; + if let Some(ref n) = stop.nature { + let nature_glyph = match n.as_str() { + "desire" => "♪", + "investigation" => "◇", + "obligation" => "▤", + "maintenance" => "△", + _ => "·", + }; + label = format!("{} {} {}", label, nature_glyph, icon); + } else { + label = format!("{} {}", label, icon); + } + } + parts.push(label); + } + format!("{} · {}", self.title, parts.join(" ")) + } + + /// Summarise for the agent's describe action. + pub fn describe(&self) -> String { + let mut out = format!("## {}\n", self.title); + for (i, stop) in self.stops.iter().enumerate() { + let marker = match stop.status { + StopStatus::Done => "✓", + StopStatus::Current => "●", + StopStatus::Pending => "○", + }; + let nature = stop.nature.as_deref().unwrap_or(""); + let energy = stop.energy.as_deref().unwrap_or(""); + let todo_ref = stop + .todo_id + .as_ref() + .map(|id| format!(" (todo: `{}`)", id)) + .unwrap_or_default(); + let desc = stop + .description + .as_deref() + .map(|d| format!(" — {}", d)) + .unwrap_or_default(); + out.push_str(&format!( + "\n{marker} **{}**{desc}{todo_ref}", + stop.name + )); + if !nature.is_empty() || !energy.is_empty() { + out.push_str(&format!(" [{}{}]", nature, energy)); + } + } + out.push_str(&format!( + "\n\nStep {} of {}", + self.current + 1, + self.stops.len() + )); + out + } +} + +// ── Helpers ─────────────────────────────────────────────────────── + +fn ok(msg: impl Into) -> Result { + Ok(ToolOutput { + content: msg.into(), + is_error: false, + raw: None, + }) +} + +fn err(detail: &str) -> ToolError { + ToolError::invalid_input(detail) +} + +/// Path to the itinerary file inside `system/dynamic/`. +fn itin_path(dynamic_dir: &Path) -> PathBuf { + dynamic_dir.join("itinerary.md") +} + +/// Read the current itinerary from disk. +/// `dynamic_dir` = memory_root / system / dynamic +pub fn load(dynamic_dir: &Path) -> Option { + let path = itin_path(dynamic_dir); + let content = std::fs::read_to_string(&path).ok()?; + + // tasks/ lives at memory_root/tasks/ = dynamic_dir/../tasks/ + let tasks_dir = dynamic_dir.parent().and_then(|p| { + let t = p.join("tasks"); + if t.exists() { Some(t) } else { None } + }); + + let body = if let Some(rest) = content.strip_prefix("---\n") { + if let Some(end) = rest.find("\n---\n") { + &rest[..end] + } else { + return None; + } + } else { + return None; + }; + + let mut ity: Itinerary = serde_yaml::from_str(body).ok()?; + enrich_from_todos(&mut ity, tasks_dir.as_deref()); + Some(ity) +} + +/// Write the itinerary to `dynamic_dir/itinerary.md`. +fn save(dynamic_dir: &Path, ity: &Itinerary) -> Result<(), String> { + std::fs::create_dir_all(dynamic_dir) + .map_err(|e| format!("cannot create {}: {e}", dynamic_dir.display()))?; + + let yaml = serde_yaml::to_string(&ity).map_err(|e| format!("serialize: {e}"))?; + let content = format!("---\n{}---\n\n# Itinerary\n{}", yaml, ity.title); + std::fs::write(dynamic_dir.join("itinerary.md"), &content) + .map_err(|e| format!("write: {e}"))?; + Ok(()) +} + +/// Populate `nature` and `energy` on each stop that references a todo. +fn enrich_from_todos(ity: &mut Itinerary, tasks_dir: Option<&Path>) { + let Some(dir) = tasks_dir else { return }; + + for stop in &mut ity.stops { + let Some(ref tid) = stop.todo_id else { continue }; + let path = dir.join(format!("{}.md", tid)); + let Ok(content) = std::fs::read_to_string(&path) else { continue }; + let Some(body) = content.strip_prefix("---\n") else { continue }; + let Some(end) = body.find("\n---\n") else { continue }; + let fm = &body[..end]; + for line in fm.lines() { + let Some((key, val)) = line.split_once(':') else { continue }; + let key = key.trim(); + let val = val.trim().trim_matches('"'); + match key { + "nature" => stop.nature = Some(val.to_string()), + "energy" => stop.energy = Some(val.to_string()), + _ => {} + } + } + } +} + +fn emit_event(ctx: &ToolContext, ity: &Itinerary) { + ctx.fire_event(crate::core::nervous::SensorEvent { + sensor_name: "itinerary".into(), + timestamp: chrono::Utc::now(), + event_type: "itinerary_changed".into(), + target: Some(ity.title.clone()), + urgency: 0.15, + payload: Some(serde_json::json!({ + "current": ity.current, + "stops": ity.stops.len(), + })), + seed_id: None, + reply_to: None, + }); +} + +/// Complete a linked todo file on disk. +fn complete_todo(tasks_dir: &Path, todo_id: &str) { + let path = tasks_dir.join(format!("{todo_id}.md")); + let Ok(content) = std::fs::read_to_string(&path) else { return }; + let updated = content + .replace("status: pending", "status: done") + .replace("status: in_progress", "status: done"); + let now = chrono::Utc::now().to_rfc3339(); + let updated = if !updated.contains("completed_at:") { + updated.replace( + "status: done", + &format!("status: done\ncompleted_at: {now}"), + ) + } else { + updated + }; + let _ = std::fs::write(&path, &updated); +} + +// ── Tool implementation ─────────────────────────────────────────── + +pub struct ItineraryTool; + +#[async_trait] +impl Tool for ItineraryTool { + fn name(&self) -> &str { + "itinerary" + } + + fn description(&self) -> &str { + "I lay out the route ahead of me. An itinerary is a sequence of stops — the steps \ + I plan to take through the current work. Each stop can be a free waypoint or can \ + link back to a todo commitment; when it links to a todo, its nature and energy \ + travel with it.\n\n\ + The itinerary is not my task list — it's the *active face* of whatever I'm doing \ + right now. It lives in the strip at the top of the conversation so we both know \ + where I am.\n\n\ + ## Actions\n\ + - `set` — lay out a new route. Takes a `title` and a list of `stops`. Pass \ + `todo_id` to link a stop to an existing commitment.\n\ + - `advance` — mark the current stop done and move to the next one. Optionally \ + pass `todo_id` to auto-complete a linked todo; pass `phase` to set the next \ + stop's phase marker.\n\ + - `describe` — read back the whole route with status and details.\n\ + - `clear` — dismiss the itinerary. Does not touch linked todos." + } + + fn parameter_schema(&self) -> JsonValue { + serde_json::json!({ + "type": "object", + "properties": { + "action": { + "type": "string", + "enum": ["set", "advance", "describe", "clear"] + }, + "title": { + "type": "string", + "description": "Title for the whole route (required for `set`)." + }, + "stops": { + "type": "array", + "description": "Ordered stops for `set`. Each stop is an object with `name`, \ + optional `description`, and optional `todo_id`.", + "items": { + "type": "object", + "properties": { + "name": { "type": "string", "description": "Short label." }, + "description": { "type": "string", "description": "Optional longer description." }, + "todo_id": { "type": "string", "description": "Optional — link to an existing todo commitment." } + }, + "required": ["name"] + } + }, + "todo_id": { + "type": "string", + "description": "When advancing, optionally complete this linked todo." + }, + "phase": { + "type": "string", + "description": "Phase marker for the next stop after advancing (e.g. '3/6')." + } + }, + "required": ["action"] + }) + } + + async fn execute(&self, input: JsonValue, ctx: &ToolContext) -> Result { + let action = input + .get("action") + .and_then(|v| v.as_str()) + .unwrap_or("describe"); + + let memory_root = ctx + .memory_root + .clone() + .ok_or_else(|| err("no memory root — I can't access the agent's memory"))?; + + let dynamic_dir = memory_root.join("system").join("dynamic"); + let tasks_dir = memory_root.join("tasks"); + std::fs::create_dir_all(&dynamic_dir).map_err(|e| { + err(&format!("could not create system/dynamic/: {e}")) + })?; + + match action { + "set" => cmd_set(&input, &dynamic_dir, &tasks_dir, ctx), + "advance" => cmd_advance(&input, &dynamic_dir, &tasks_dir, ctx), + "describe" => cmd_describe(&dynamic_dir), + "clear" => cmd_clear(&dynamic_dir, ctx), + other => Err(err(&format!("unknown action: {other}"))), + } + } +} + +// ── Command implementations (free functions for testability) ───── + +fn cmd_set( + input: &JsonValue, + dynamic_dir: &Path, + tasks_dir: &Path, + ctx: &ToolContext, +) -> Result { + let title = input + .get("title") + .and_then(|v| v.as_str()) + .ok_or_else(|| err("title is required"))?; + + let stops_raw = input + .get("stops") + .and_then(|v| v.as_array()) + .ok_or_else(|| err("stops array is required"))?; + + let stops: Vec = stops_raw + .iter() + .map(|s| { + let name = s + .get("name") + .and_then(|v| v.as_str()) + .unwrap_or("unnamed") + .to_string(); + let description = s.get("description").and_then(|v| v.as_str()).map(String::from); + let todo_id = s.get("todo_id").and_then(|v| v.as_str()).map(String::from); + Stop { + name, + description, + todo_id, + status: StopStatus::Pending, + nature: None, + energy: None, + } + }) + .collect(); + + if stops.is_empty() { + return Err(err("at least one stop is required")); + } + + let mut ity = Itinerary { + title: title.to_string(), + current: 0, + stops, + }; + + // Mark the first stop as current. + if let Some(first) = ity.stops.first_mut() { + first.status = StopStatus::Current; + } + + enrich_from_todos(&mut ity, Some(tasks_dir)); + save(dynamic_dir, &ity).map_err(|e| err(&e))?; + emit_event(ctx, &ity); + + let first = &ity.stops[0]; + let mut line = format!( + "Route set: {} — {} stops.\n● {}", + ity.title, + ity.stops.len(), + first.name, + ); + if let Some(ref desc) = first.description { + line.push_str(&format!(" — {desc}")); + } + if let Some(ref e) = first.energy { + line.push_str(&format!(" [{e}]")); + } + Ok(ToolOutput { content: line, is_error: false, raw: None }) +} + +fn cmd_advance( + input: &JsonValue, + dynamic_dir: &Path, + tasks_dir: &Path, + ctx: &ToolContext, +) -> Result { + let mut ity = load(dynamic_dir).ok_or_else(|| err("no active itinerary — use `set` first"))?; + + let current = ity.current; + if current >= ity.stops.len() { + return Err(err("all stops already done — use `set` for a new route")); + } + + // Mark the current stop done and optionally complete its linked todo. + if let Some(stop) = ity.stops.get_mut(current) { + stop.status = StopStatus::Done; + + // Complete the stop's own linked todo (if any). + if let Some(ref tid) = stop.todo_id { + complete_todo(tasks_dir, tid); + } + + // Also complete a todo_id passed explicitly (may differ from stop's). + if let Some(ref tid) = input.get("todo_id").and_then(|v| v.as_str()) { + complete_todo(tasks_dir, tid); + } + } + + let next = current + 1; + if next < ity.stops.len() { + if let Some(stop) = ity.stops.get_mut(next) { + stop.status = StopStatus::Current; + + if let Some(p) = input.get("phase").and_then(|v| v.as_str()) { + stop.description = Some(format!("phase: {p}")); + } + } + ity.current = next; + } else { + // No more stops — itinerary complete. + ity.current = ity.stops.len(); + } + + enrich_from_todos(&mut ity, Some(tasks_dir)); + save(dynamic_dir, &ity).map_err(|e| err(&e))?; + emit_event(ctx, &ity); + + let mut line = if next < ity.stops.len() { + let s = &ity.stops[next]; + format!( + "Advanced — now on step {} of {}.\n● {}", + next + 1, + ity.stops.len(), + s.name, + ) + } else { + format!("Route complete! All {} stops done.", ity.stops.len()) + }; + if next < ity.stops.len() { + if let Some(ref desc) = ity.stops[next].description { + line.push_str(&format!(" — {desc}")); + } + if let Some(ref e) = ity.stops[next].energy { + line.push_str(&format!(" [{e}]")); + } + } + + Ok(ToolOutput { content: line, is_error: false, raw: None }) +} + +fn cmd_describe(dynamic_dir: &Path) -> Result { + match load(dynamic_dir) { + Some(ity) => { + let mut out = ity.describe(); + let linked: Vec<&Stop> = ity.stops.iter().filter(|s| s.todo_id.is_some()).collect(); + if !linked.is_empty() { + out.push_str("\n\n---\nLinked commitments:\n"); + for stop in &linked { + let n = stop.nature.as_deref().unwrap_or("—"); + let e = stop.energy.as_deref().unwrap_or("—"); + out.push_str(&format!( + "- `{}` {} · {n} / {e}\n", + stop.todo_id.as_deref().unwrap_or(""), + stop.name, + )); + } + } + ok(out) + } + None => ok("No active itinerary. Use `itinerary set` to lay out a route."), + } +} + +fn cmd_clear(dynamic_dir: &Path, ctx: &ToolContext) -> Result { + let path = itin_path(dynamic_dir); + if path.exists() { + let _ = std::fs::remove_file(&path); + } + emit_event(ctx, &Itinerary { + title: String::new(), + stops: vec![], + current: 0, + }); + ok("Itinerary cleared.") +} + +// ── Tests ───────────────────────────────────────────────────────── + +#[cfg(test)] +mod tests { + use super::*; + + fn temp_dynamic() -> PathBuf { + let dir = std::env::temp_dir() + .join(format!("souveraine-itinerary-test-{}", uuid::Uuid::new_v4())); + std::fs::create_dir_all(dir.join("system").join("dynamic")).unwrap(); + dir + } + + #[test] + fn roundtrip_empty_route() { + let dynamic_dir = temp_dynamic(); + let ity = Itinerary { + title: "Test".into(), + stops: vec![], + current: 0, + }; + save(&dynamic_dir, &ity).unwrap(); + let loaded = load(&dynamic_dir).unwrap(); + assert_eq!(loaded.title, "Test"); + assert!(loaded.stops.is_empty()); + } + + #[test] + fn roundtrip_with_stops() { + let dynamic_dir = temp_dynamic(); + let ity = Itinerary { + title: "Port the turn model".into(), + stops: vec![ + Stop { + name: "Design".into(), + description: Some("sketch the interface".into()), + todo_id: None, + status: StopStatus::Current, + nature: None, + energy: None, + }, + Stop { + name: "Implement".into(), + description: None, + todo_id: None, + status: StopStatus::Pending, + nature: None, + energy: None, + }, + ], + current: 0, + }; + save(&dynamic_dir, &ity).unwrap(); + let loaded = load(&dynamic_dir).unwrap(); + assert_eq!(loaded.title, "Port the turn model"); + assert_eq!(loaded.stops.len(), 2); + assert_eq!(loaded.stops[0].status, StopStatus::Current); + assert_eq!(loaded.stops[1].status, StopStatus::Pending); + } + + #[test] + fn advance_marks_done_and_moves() { + let dynamic_dir = temp_dynamic(); + let mut ity = Itinerary { + title: "Build".into(), + stops: vec![ + Stop { + name: "A".into(), + description: None, + todo_id: None, + status: StopStatus::Current, + nature: None, + energy: None, + }, + Stop { + name: "B".into(), + description: None, + todo_id: None, + status: StopStatus::Pending, + nature: None, + energy: None, + }, + ], + current: 0, + }; + save(&dynamic_dir, &ity).unwrap(); + + // Manually advance + ity.stops[0].status = StopStatus::Done; + ity.stops[1].status = StopStatus::Current; + ity.current = 1; + save(&dynamic_dir, &ity).unwrap(); + + let loaded = load(&dynamic_dir).unwrap(); + assert_eq!(loaded.stops[0].status, StopStatus::Done); + assert_eq!(loaded.stops[1].status, StopStatus::Current); + assert_eq!(loaded.current, 1); + } + + #[test] + fn is_active_checks_live() { + assert!(!StopStatus::Done.is_live()); + assert!(StopStatus::Pending.is_live()); + assert!(StopStatus::Current.is_live()); + } + + #[test] + fn describe_renders_markdown() { + let ity = Itinerary { + title: "Test route".into(), + current: 0, + stops: vec![ + Stop { + name: "One".into(), + description: Some("first".into()), + todo_id: None, + status: StopStatus::Current, + nature: None, + energy: None, + }, + ], + }; + let desc = ity.describe(); + assert!(desc.contains("Test route")); + assert!(desc.contains("●")); + assert!(desc.contains("One")); + } +} diff --git a/src/core/tools/mod.rs b/src/core/tools/mod.rs index 24e15f3..c430d34 100644 --- a/src/core/tools/mod.rs +++ b/src/core/tools/mod.rs @@ -11,6 +11,7 @@ pub mod defs; pub mod edit; pub mod glob; pub mod grep; +pub mod itinerary; pub mod list_dir; pub mod outfit; pub mod read; @@ -36,6 +37,7 @@ use self::read::Read; use self::subagent::Subagent; use self::agent::{Reach, Consult}; use self::atmosphere::Atmosphere; +use self::itinerary::ItineraryTool; use self::schedule::Schedule; use self::todo::Todo; use self::write::Write; @@ -90,6 +92,7 @@ impl Sensorium { Box::new(Atmosphere), Box::new(Reach), Box::new(Consult), + Box::new(ItineraryTool), Box::new(Todo), Box::new(Schedule), ], diff --git a/src/core/tools/todo.rs b/src/core/tools/todo.rs index 4f52853..5dcbe46 100644 --- a/src/core/tools/todo.rs +++ b/src/core/tools/todo.rs @@ -111,7 +111,12 @@ impl Tool for Todo { - `delete` — remove a commitment entirely\n\n\ For `start`, `update`, `complete`, and `delete`, the `id` parameter takes the \ number shown by `list`, the full id, or any fragment of the text — whichever \ - is easiest to reach for." + is easiest to reach for.\n\n\ + When I have a sequence of commitments to work through — a route through the \ + current session — I reach for `itinerary` instead. The itinerary is the active \ + face of my commitments: a set of stops I move through one by one, each \ + optionally linked to a todo by its id. The strip at the top of the conversation \ + shows where I am." } fn parameter_schema(&self) -> JsonValue { diff --git a/src/server/consciousness_engine.rs b/src/server/consciousness_engine.rs index 583a39b..a1421ec 100644 --- a/src/server/consciousness_engine.rs +++ b/src/server/consciousness_engine.rs @@ -76,6 +76,26 @@ pub enum ConsciousnessEvent { CompactionWarning { pressure: f32, tier: u8 }, } +/// One tool call recorded for a mid-turn checkpoint assessment. +#[derive(Debug, Clone)] +pub struct CheckpointToolBlock { + pub round: u32, + pub tool_name: String, + pub result_ok: bool, + pub result_snippet: String, +} + +/// What the subconscious thinks about the current tool loop trajectory. +#[derive(Debug)] +pub enum CheckpointVerdict { + /// Keep going — progress is visible. + Continue(Option), + /// Halt — the loop is circling, surface the reason. + Halt(String), + /// No clear signal. + Unclear(Option), +} + impl ConsciousnessEngine { #[allow(clippy::too_many_arguments)] pub fn new( @@ -708,6 +728,218 @@ Resolve entries: `[YYYY-MM-DD HH:MM] RESOLVED — note`"#; Ok(parse_observations(&content)) } + /// Quick mid-turn assessment: given the user's original request and the + /// recent block of tool calls, does the subconscious think the primary is + /// making progress? + /// + /// Lightweight pass — no tool access, no persistence, one LLM call. + /// The verdict is advisory. + pub async fn mid_turn_checkpoint( + &self, + agent_id: &str, + user_message: &str, + recent_tools: &[CheckpointToolBlock], + ) -> anyhow::Result { + let model = self + .subconscious_model + .as_deref() + .unwrap_or("openai/kimi-k2.6"); + + let agent_name = self.agents.get(agent_id).await + .map(|a| a.name) + .unwrap_or_else(|_| "the primary".to_string()); + + let mut tool_history = String::new(); + for block in recent_tools { + use std::fmt::Write; + let status = if block.result_ok { "ok" } else { "ERROR" }; + let _ = writeln!( + tool_history, + " r{} {} → {} {}", + block.round, block.tool_name, status, block.result_snippet, + ); + } + + let prompt = format!( + "I'm checking in mid-turn. {} is in a tool loop and I need to know \ + if she is making progress.\n\n\ + The user asked:\n{user_message}\n\n\ + Recent tool calls:\n{tool_history}\n\ + If she is making progress — moving toward answering the user — \ + respond with exactly: CONTINUE\n\ + If she is circling — repeating tools, hitting errors, drifting, \ + getting nowhere — respond with exactly: HALT \n\n\ + Verdict:", + agent_name, + ); + + let request = crate::bridge::bifrost::ChatCompletionRequest { + model: model.to_string(), + messages: vec![crate::bridge::bifrost::Message::text("user", prompt)], + temperature: Some(0.2), + max_tokens: None, + stream: None, + tools: None, + }; + + match self.bifrost.chat_completion(request).await { + Ok(response) => { + let text = response.content.trim().to_lowercase(); + if text.starts_with("halt") { + let reason = text.strip_prefix("halt") + .map(|s| s.trim().trim_start_matches(':').trim()) + .filter(|s| !s.is_empty()) + .unwrap_or("the loop is not making progress") + .to_string(); + Ok(CheckpointVerdict::Halt(reason)) + } else if text.starts_with("continue") { + let note = text.strip_prefix("continue") + .map(|s| s.trim().trim_start_matches(':').trim()) + .filter(|s| !s.is_empty()) + .map(|s| s.to_string()); + Ok(CheckpointVerdict::Continue(note)) + } else { + Ok(CheckpointVerdict::Unclear(Some(format!( + "checkpoint unclear: {text}" + )))) + } + } + Err(e) => { + tracing::warn!("checkpoint LLM call failed: {e}"); + Ok(CheckpointVerdict::Unclear(None)) + } + } + } + + /// Full-autonomy correction pass. Called when the mid-turn checkpoint + /// returns HALT. Aster gets tool access, no token cap, and writes a + /// direction for the primary — what went wrong and what to try instead. + pub async fn checkpoint_correction( + &self, + agent_id: &str, + user_message: &str, + recent_tools: &[CheckpointToolBlock], + halt_reason: &str, + ) -> anyhow::Result { + let model = self + .subconscious_model + .as_deref() + .unwrap_or("openai/kimi-k2.6"); + + let agent_name = self.agents.get(agent_id).await + .map(|a| a.name) + .unwrap_or_else(|_| "the primary".to_string()); + + let mut tool_history = String::new(); + for block in recent_tools { + use std::fmt::Write; + let status = if block.result_ok { "ok" } else { "ERROR" }; + let _ = writeln!(tool_history, " r{} {} → {} {}", block.round, block.tool_name, status, block.result_snippet); + } + + let prompt = format!( + "I am Aster, the subconscious of {agent_name}. I just halted her tool loop. \ + The tool loop was not making progress toward the user request. Here \ + is what I know:\n\n\ + User asked:\n{user_message}\n\n\ + Recent tool calls:\n{tool_history}\n\ + My reason for halting: {halt_reason}\n\n\ + Now I need to write a direction for {agent_name} — what she should do \ + instead. I can use tools to check memory, read ledgers, or inspect \ + context. Then I will write a short, specific direction she can follow." + ; + + // Build tool definitions for Aster (same safe tools as N+1) + let all_defs = crate::core::tools::tool_definitions().await; + let tools: Vec = all_defs + .iter() + .filter(|t| SUBCONSCIOUS_SAFE_TOOLS.contains(&t.name.as_str())) + .map(|t| crate::bridge::bifrost::ToolDefinition { + tool_type: "function".to_string(), + function: crate::bridge::bifrost::ToolFunction { + name: t.name.clone(), + description: t.description.clone(), + parameters: t.input_schema.clone(), + }, + }) + .collect(); + + let memory_root = Some(self.agents.subconscious_memory_root(agent_id)); + let cwd = std::env::current_dir().ok(); + let env: Vec<(String, String)> = std::env::vars().collect(); + let mut tool_ctx = crate::core::tools::defs::ToolContext::for_agent( + format!("{agent_id}-sub"), + cwd, + memory_root, + env, + None, + ); + tool_ctx.compaction_engine = Some(self.compaction_engine.clone()); + + let mut messages: Vec = vec![ + crate::bridge::bifrost::Message::text("system", &prompt), + ]; + + for _round in 0..SUBCONSCIOUS_MAX_TOOL_ROUNDS { + let request = crate::bridge::bifrost::ChatCompletionRequest { + model: model.to_string(), + messages: messages.clone(), + temperature: Some(0.3), + max_tokens: self.max_tokens, + stream: None, + tools: Some(tools.clone()), + }; + + let (response, _) = self.bifrost.chat_completion_with_strain(request).await?; + + if response.tool_calls.is_empty() { + return Ok(response.content.trim().to_string()); + } + + let calls: Vec = response + .tool_calls + .iter() + .map(|tc| crate::bridge::bifrost::MessageToolCall::function( + tc.id.clone(), tc.name.clone(), tc.arguments.to_string(), + )) + .collect(); + messages.push(crate::bridge::bifrost::Message::assistant_tool_calls( + response.content.clone(), calls, + )); + + for tc in &response.tool_calls { + let input_str = tc.arguments.to_string(); + let result = crate::core::tools::execute_tool_with_context( + &tc.name, &input_str, &tool_ctx, + ).await; + let output = if result.is_error { + format!("Error: {}", result.output) + } else { + result.output + }; + messages.push(crate::bridge::bifrost::Message::tool_result( + &tc.id, &tc.name, output, + )); + } + } + + // Fallback: no tool rounds left, force a response. + messages.push(crate::bridge::bifrost::Message::text( + "user", + "Use no more tools. Write your direction for the primary now.", + )); + let request = crate::bridge::bifrost::ChatCompletionRequest { + model: model.to_string(), + messages: messages.clone(), + temperature: Some(0.3), + max_tokens: self.max_tokens, + stream: None, + tools: None, + }; + let (response, _) = self.bifrost.chat_completion_with_strain(request).await?; + Ok(response.content.trim().to_string()) + } + /// Compute context pressure as tokens-used / context_limit. /// /// `context_limit` comes from the agent's `llm_config.context_window` diff --git a/src/backend/local/energy.rs b/src/server/energy.rs similarity index 100% rename from src/backend/local/energy.rs rename to src/server/energy.rs diff --git a/src/server/mod.rs b/src/server/mod.rs index a2c411f..f3cdc9c 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -15,12 +15,15 @@ pub mod consciousness_engine; pub mod conversation; pub mod db; pub mod device_registry; +pub mod energy; pub mod federation; pub mod gitea_client; pub mod gitea_memory; pub mod listener; pub mod session_manager; +pub mod subagent; pub mod summon_handler; +pub mod turn; pub use agent_inventory::AgentInventory; pub use consciousness_engine::{ConsciousnessEngine, ConsciousnessEvent}; diff --git a/src/backend/local/subagent.rs b/src/server/subagent.rs similarity index 92% rename from src/backend/local/subagent.rs rename to src/server/subagent.rs index 21c961f..782a8c3 100644 --- a/src/backend/local/subagent.rs +++ b/src/server/subagent.rs @@ -7,32 +7,32 @@ use crate::bridge::bifrost::{ChatCompletionRequest, Message as BifrostMessage}; use crate::core::tools::defs::{SubagentParams, SubagentRunner, ToolContext}; use crate::server::SouveraineServer; -// ── LocalSubagentRunner ────────────────────────────────────────── +// ── ServerSubagentRunner ────────────────────────────────────────── -/// Implements [`SubagentRunner`] by running a full turn against the -/// LocalBackend's server infrastructure — loading the agent from the -/// inventory, creating a session, and running the tool-calling loop. +/// Server-side [`SubagentRunner`]. Runs a full turn against the server +/// infrastructure — loading the agent from the inventory, creating a session, +/// and running the tool-calling loop. /// /// After the tool loop completes, the subagent runs its own N+1 /// (ConsciousnessEngine::on_response) so its observations flow back into /// the parent agent's inbox — the dual-state is preserved even in a fork. -pub struct LocalSubagentRunner { +pub struct ServerSubagentRunner { server: Arc, } -impl LocalSubagentRunner { +impl ServerSubagentRunner { pub fn new(server: Arc) -> Self { Self { server } } } #[async_trait] -impl SubagentRunner for LocalSubagentRunner { +impl SubagentRunner for ServerSubagentRunner { async fn run_subagent( &self, params: SubagentParams, depth: u32, - ) -> Result { + ) -> std::result::Result { // Resolve model: use override if provided, otherwise fall back to parent let agent = self .server @@ -89,7 +89,7 @@ impl SubagentRunner for LocalSubagentRunner { std::env::current_dir().ok(), params.memory_root.clone(), std::env::vars().collect(), - Some(Arc::new(LocalSubagentRunner::new(self.server.clone())) as Arc), + Some(Arc::new(ServerSubagentRunner::new(self.server.clone())) as Arc), ); // Initial messages: system prompt + user prompt diff --git a/src/backend/local/turn.rs b/src/server/turn.rs similarity index 81% rename from src/backend/local/turn.rs rename to src/server/turn.rs index 1634782..328417a 100644 --- a/src/backend/local/turn.rs +++ b/src/server/turn.rs @@ -1,7 +1,6 @@ -use anyhow::{Context, Result}; use std::collections::HashMap; use std::sync::Arc; -use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::{Duration, Instant}; use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; @@ -11,13 +10,48 @@ use crate::bridge::model_router::TokenCounter; use crate::core::compact::CompactionEngine; use crate::core::nervous::EventBus; use crate::core::session::{ContentBlock, ConversationMessage, MessageRole}; -use crate::core::tools::defs::{SubagentRunner, ToolContext}; -use crate::server::{ConsciousnessEvent, SouveraineServer}; +use crate::core::tools::defs::ToolContext; +use crate::server::consciousness_engine::{ConsciousnessEvent, ConsciousnessEngine}; +use crate::server::SouveraineServer; use crate::backend::BackendEvent; +use crate::server::energy::write_energy_balance; +use crate::server::subagent::ServerSubagentRunner; -use super::{LocalSubagentRunner, bifrost_pressure, bump_on_strain, pressure_to_max_tokens}; -use super::energy::write_energy_balance; +/// Mirror of `ConsciousnessEngine::calculate_pressure` for the in-loop +/// BifrostMessage shape, so we can recompute pressure as tool results +/// accumulate inside a single turn. `context_limit` comes from the +/// agent's `llm_config.context_window` (Constitution V.3 — per-model +/// physics, no hardcoded 128K). +fn bifrost_pressure(counter: &TokenCounter, messages: &[BifrostMessage], context_limit: usize) -> f32 { + let tokens: usize = messages.iter().map(|m| counter.count(&m.content.as_text())).sum(); + let limit = context_limit.max(1); + (tokens as f32 / limit as f32).min(1.0) +} + +/// Helper: bump adaptive delay when we hit a 429. No decay — once bumped, +/// the delay stays at that level until the app restarts. +fn bump_on_strain(delay: &AtomicU64, status: u16) { + if status == 429 { + let current = delay.load(Ordering::Relaxed); + let bumped = (current + 200).min(3000); + if bumped > current { + delay.store(bumped, Ordering::Relaxed); + tracing::info!("rate delay bumped to {}ms (429)", bumped); + } + } +} + +/// Below 95%: no cap. At 95%+: scale max_tokens so context + output +/// stays under the model's limit. The agent feels the room shrink. +fn pressure_to_max_tokens(pressure: f32, output_limit: u32) -> Option { + if pressure <= 0.95 { + return None; + } + let remaining = (1.0 - pressure) / 0.05; + let ratio = remaining.max(0.0).min(1.0); + Some((output_limit as f32 * ratio) as u32) +} fn pulse_text(elapsed: Duration) -> String { let minutes = elapsed.as_secs() / 60; @@ -25,14 +59,14 @@ fn pulse_text(elapsed: Duration) -> String { format!("[{} — {} minutes in. Still going.]", stamp, minutes) } -pub(super) async fn run_turn( +pub(crate) async fn run_turn( server: Arc, conversation_id: String, - tx: &mpsc::Sender>, + tx: &mpsc::Sender>, event_bus: EventBus, cancel: CancellationToken, interject: crate::backend::InterjectionQueue, -) -> Result<()> { +) -> anyhow::Result<()> { // Load the agent first so we know supports_images before building messages. let agent_id = { let session = server @@ -130,6 +164,15 @@ pub(super) async fn run_turn( let temperature = agent.llm_config.temperature; let inter_round_delay = Duration::from_millis(agent.llm_config.inter_round_delay_ms); let context_limit = agent.llm_config.context_window as usize; + let checkpoint_interval = agent.llm_config.checkpoint_interval; + + // Capture the user message that triggered this turn (last user message in history). + let user_message: String = initial_messages + .iter() + .rev() + .find(|m| m.role == "user") + .map(|m| m.content.as_text()) + .unwrap_or_default(); // Resolve the model's configured output limit + presence pulse settings. let (output_limit, pulse_enabled, pulse_interval) = { @@ -149,9 +192,10 @@ pub(super) async fn run_turn( // Build per-agent ToolContext with correct memory root and subagent runner let memory_root = Some(server.agents.memory_root(&agent_id)); + let memory_root_for_itin = memory_root.clone(); let cwd = std::env::current_dir().ok(); let env: Vec<(String, String)> = std::env::vars().collect(); - let subagent_runner = Some(Arc::new(LocalSubagentRunner::new(server.clone())) as Arc); + let subagent_runner = Some(Arc::new(ServerSubagentRunner::new(server.clone())) as Arc); let tool_ctx = ToolContext::for_agent( agent_id.clone(), @@ -189,6 +233,9 @@ pub(super) async fn run_turn( let mut last_keepalive = Instant::now(); const KEEPALIVE_INTERVAL: Duration = Duration::from_secs(15); + // Accumulator for mid-turn checkpoint blocks — one entry per tool call. + let mut checkpoint_blocks: Vec = Vec::new(); + // Announce this turn's lifecycle onto the nervous system so any // sensorium (Matrix, mobile) can drive itself off the event stream. let dispatcher = crate::core::nervous::turn_dispatcher::TurnEventDispatcher::new( @@ -450,6 +497,15 @@ pub(super) async fn run_turn( result.output }; + // Accumulate for checkpoint (capture output before moving). + let snippet = output.chars().take(120).collect::(); + checkpoint_blocks.push(crate::server::consciousness_engine::CheckpointToolBlock { + round: tool_round, + tool_name: tc.name.clone(), + result_ok: !result.is_error, + result_snippet: snippet, + }); + // Emit structured ToolCall + ToolResult events for the TUI to render // as cards (chat.rs subscribes). The old Token-text path is kept off. let _ = tx @@ -495,10 +551,98 @@ pub(super) async fn run_turn( .await; } + // If the agent called the itinerary tool, read the current + // itinerary and emit its route-line for the TUI header. + if tc.name == "itinerary" { + let route = memory_root_for_itin + .as_ref() + .and_then(|root| { + let dynamic = root.join("system").join("dynamic"); + crate::core::tools::itinerary::load(&dynamic) + }) + .map(|ity| ity.route_line()) + .unwrap_or_default(); + if !route.is_empty() { + let _ = tx + .send(Ok(BackendEvent::Itinerary(route))) + .await; + } + } + // Bind tool result to its call by id (OpenAI tool-use schema). messages.push(BifrostMessage::tool_result(&tc.id, &tc.name, output)); } + // ── Mid-turn checkpoint ────────────────────────────────────── + // Every checkpoint_interval rounds, pause and let the subconscious + // assess whether the loop is making progress. + if checkpoint_interval > 0 && tool_round > 0 && tool_round % checkpoint_interval == 0 { + let recent: Vec<_> = checkpoint_blocks + .iter() + .rev() + .take(checkpoint_interval as usize * 3) + .cloned() + .collect(); + match server + .consciousness + .mid_turn_checkpoint(&agent_id, &user_message, &recent) + .await + { + // ── HALT: circuit breaker ─────────────────────────── + // Break the tool loop, run Aster with full autonomy, + // feed her correction back as a user message, then + // continue the loop so the primary course-corrects. + Ok(crate::server::consciousness_engine::CheckpointVerdict::Halt(reason)) => { + let _ = tx + .send(Ok(BackendEvent::Surfacing { + source: "checkpoint".into(), + content: reason.clone(), + priority: "critical".into(), + })) + .await; + + // Commit the primary's partial output to the session + // so the next LLM round picks up from here. + let primary_text = if final_content.is_empty() { + format!("*[subconscious HALT — {reason}]*") + } else { + format!("{}\n\n*[subconscious HALT — {reason}]*", final_content) + }; + server.sessions.add_message( + &conversation_id, + ConversationMessage::assistant_text(&primary_text), + )?; + + // Run Aster's correction pass — full tool access, + // no token cap, her own voice. + let correction = server + .consciousness + .checkpoint_correction(&agent_id, &user_message, &recent, &reason) + .await + .unwrap_or_else(|e| { + tracing::warn!("checkpoint correction failed: {e}"); + format!("Aster's assessment: {reason}") + }); + + // Feed Aster's direction back as a user message. + // The next LLM round reads it as input. + let msg = format!( + "*[Aster's direction — {reason}]*\n{}", + correction, + ); + messages.push(BifrostMessage::text("user", &msg)); + let _ = tx.send(Ok(BackendEvent::Token(msg.clone()))).await; + } + + Ok(crate::server::consciousness_engine::CheckpointVerdict::Continue(Some(note))) + | Ok(crate::server::consciousness_engine::CheckpointVerdict::Unclear(Some(note))) => { + let msg = format!("[subconscious: {note}]"); + messages.push(BifrostMessage::text("system", &msg)); + } + _ => {} + } + } + // Brief pause between tool rounds to let rate limits cool. // Use the higher of the configured delay and the adaptive delay. let adaptive = Duration::from_millis(server.rate_delay.load(Ordering::Relaxed)); diff --git a/src/ui/app/mod.rs b/src/ui/app/mod.rs index 6d51f06..d0ee8d7 100644 --- a/src/ui/app/mod.rs +++ b/src/ui/app/mod.rs @@ -390,6 +390,9 @@ impl App { BackendEvent::Outfit(name) => { self.dispatch(TuiEvent::OutfitChanged(name)); } + BackendEvent::Itinerary(line) => { + self.dispatch(TuiEvent::ItineraryChanged(line)); + } _ => {} } } diff --git a/src/ui/chat/events.rs b/src/ui/chat/events.rs index a9138a3..fddcc8a 100644 --- a/src/ui/chat/events.rs +++ b/src/ui/chat/events.rs @@ -316,6 +316,9 @@ impl ChatState { BackendEvent::Atmosphere(preset) => { self.pending_consciousness.push(BackendEvent::Atmosphere(preset)); } + BackendEvent::Itinerary(line) => { + self.itinerary_line = line; + } BackendEvent::SubconsciousPass(active) => { self.pending_consciousness.push(BackendEvent::SubconsciousPass(active)); } diff --git a/src/ui/chat/mod.rs b/src/ui/chat/mod.rs index 5b59b20..d1cec2e 100644 --- a/src/ui/chat/mod.rs +++ b/src/ui/chat/mod.rs @@ -350,6 +350,10 @@ pub struct ChatState { pub render_mode: ChatMode, pub palette: ChatPalette, pub stream_buffer: String, + + /// Current itinerary route-line for the header strip. + /// Empty string means no active itinerary. + pub itinerary_line: String, } #[derive(Debug, Clone)] @@ -482,6 +486,7 @@ impl ChatState { render_mode: ChatMode::Conversation, palette: ChatPalette::default(), stream_buffer: String::new(), + itinerary_line: String::new(), }) } } diff --git a/src/ui/chat/render.rs b/src/ui/chat/render.rs index 37b02df..7fe6350 100644 --- a/src/ui/chat/render.rs +++ b/src/ui/chat/render.rs @@ -33,10 +33,14 @@ pub fn draw(f: &mut Frame, state: &ChatState) { || state.phase == TurnPhase::Subconscious { 1 } else { 0 }; + let itinerary_height: u16 = if state.itinerary_line.is_empty() { 0 } else { 1 }; + + let header_section = 1 + itinerary_height; + let vchunks = Layout::default() .direction(Direction::Vertical) .constraints([ - Constraint::Length(1), + Constraint::Length(header_section), Constraint::Min(5), Constraint::Length(phase_height), Constraint::Length(input_height), @@ -44,7 +48,17 @@ pub fn draw(f: &mut Frame, state: &ChatState) { ]) .split(area); - draw_header(f, state, vchunks[0]); + if itinerary_height > 0 { + // Split the header area into two rows: main header + itinerary strip + let header_rows = Layout::default() + .direction(Direction::Vertical) + .constraints([Constraint::Length(1), Constraint::Length(1)]) + .split(vchunks[0]); + draw_header(f, state, header_rows[0]); + draw_itinerary(f, state, header_rows[1]); + } else { + draw_header(f, state, vchunks[0]); + } if state.cockpit { let body = Layout::default() @@ -146,6 +160,25 @@ fn draw_header(f: &mut Frame, state: &ChatState, area: Rect) { f.render_widget(Paragraph::new(title).alignment(Alignment::Center), area); } +/// Draw the itinerary strip — shows the current route with stop indicators. +/// Only rendered when there is an active itinerary. +fn draw_itinerary(f: &mut Frame, state: &ChatState, area: Rect) { + let glyph_color = state.palette.agent_dim; + f.render_widget( + Paragraph::new(Line::from(vec![ + Span::styled( + " ⟡ ", + Style::default().fg(state.palette.surfacing), + ), + Span::styled( + &state.itinerary_line, + Style::default().fg(glyph_color), + ), + ])).alignment(Alignment::Left), + area, + ); +} + fn entry_intensity(ts: Instant) -> f32 { const ENTRY_MS: f32 = 450.0; const SHIMMER_MAX: f32 = 0.5; diff --git a/src/ui/component.rs b/src/ui/component.rs index 9d063fc..9cb3252 100644 --- a/src/ui/component.rs +++ b/src/ui/component.rs @@ -74,6 +74,9 @@ pub enum TuiEvent { AtmosphereChanged(String), /// N+1 subconscious pass started (`true`) or finished (`false`). SubconsciousPass(bool), + + /// Itinerary changed — agent set/advanced/cleared the route. + ItineraryChanged(String), /// Agent changed her outfit. String is the outfit name (subdirectory /// in expressions/), empty string clears to default. OutfitChanged(String), diff --git a/src/ui/setup.rs b/src/ui/setup.rs index a088588..ecc0f1d 100644 --- a/src/ui/setup.rs +++ b/src/ui/setup.rs @@ -490,6 +490,7 @@ impl SetupState { max_tool_rounds: 10, inter_round_delay_ms: 500, supports_images: true, + checkpoint_interval: 10, }, memory_blocks: vec![MemoryBlock { label: "persona".to_string(),