1a3c24fadb
Drop temperature for models that removed sampling params (400 otherwise). Preserve signed thinking blocks across tool rounds — Anthropic rejects a tool_use whose preceding thinking block was dropped. Raise default max_tokens to 32k for thinking-by-default models and request summarized display so the reasoning pane is fed.
1021 lines
44 KiB
Rust
1021 lines
44 KiB
Rust
use std::sync::Arc;
|
|
use std::sync::atomic::{AtomicU64, Ordering};
|
|
use std::time::{Duration, Instant};
|
|
use tokio::sync::mpsc;
|
|
use tokio_util::sync::CancellationToken;
|
|
|
|
use crate::bridge::bifrost::{ChatCompletionRequest, ContentPart, ImageUrlSource, Message as BifrostMessage};
|
|
use crate::bridge::model_router::TokenCounter;
|
|
use crate::core::compact::CompactionEngine;
|
|
use crate::core::nervous::{EventBus, SensorEvent};
|
|
use crate::core::session::{ContentBlock, ConversationMessage, MessageRole};
|
|
use crate::core::tools::defs::ToolContext;
|
|
use crate::server::consciousness_engine::ConsciousnessEvent;
|
|
use crate::server::SouveraineServer;
|
|
|
|
use crate::backend::BackendEvent;
|
|
use crate::server::energy::write_energy_balance;
|
|
use crate::server::subagent::ServerSubagentRunner;
|
|
|
|
/// 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<u32> {
|
|
if pressure <= 0.95 {
|
|
return None;
|
|
}
|
|
let remaining = (1.0 - pressure) / 0.05;
|
|
let ratio = remaining.clamp(0.0, 1.0);
|
|
Some((output_limit as f32 * ratio) as u32)
|
|
}
|
|
|
|
fn pulse_text(elapsed: Duration) -> String {
|
|
let minutes = elapsed.as_secs() / 60;
|
|
let stamp = chrono::Local::now().format("%H:%M");
|
|
format!("[{} — {} minutes in. Still going.]", stamp, minutes)
|
|
}
|
|
|
|
/// When the subconscious calls `halt`, this is what the primary feels — a
|
|
/// somatic sentence in her own register, not commentary from outside. Same
|
|
/// pattern as `pulse_text`: substrate voice, brief, weighted by intensity.
|
|
/// The full reasoning lives in her subconscious ledger; this is the body
|
|
/// signal that points her at it.
|
|
fn migraine_text(severity: &str, reason: &str) -> String {
|
|
let stamp = chrono::Local::now().format("%H:%M");
|
|
let line = match severity {
|
|
"advisory" => format!(
|
|
"[{} — a pressure behind my eyes. {}. I should slow down.]",
|
|
stamp, reason
|
|
),
|
|
"critical" => format!(
|
|
"[{} — the room tilts. Stop. {}.]",
|
|
stamp, reason
|
|
),
|
|
// "firm" and any unexpected value land here — the default migraine.
|
|
_ => format!(
|
|
"[{} — a migraine. {}. I was about to do something I shouldn't.]",
|
|
stamp, reason
|
|
),
|
|
};
|
|
line
|
|
}
|
|
|
|
pub(crate) async fn run_turn(
|
|
server: Arc<SouveraineServer>,
|
|
conversation_id: String,
|
|
tx: &mpsc::Sender<anyhow::Result<BackendEvent>>,
|
|
event_bus: EventBus,
|
|
cancel: CancellationToken,
|
|
interject: crate::backend::InterjectionQueue,
|
|
) -> anyhow::Result<()> {
|
|
// Load the agent first so we know supports_images before building messages.
|
|
let agent_id = {
|
|
let session = server
|
|
.sessions
|
|
.get(&conversation_id)
|
|
.ok_or_else(|| anyhow::anyhow!("Session not found: {}", conversation_id))?;
|
|
session.agent_id.clone()
|
|
};
|
|
|
|
let agent = server.agents.get(&agent_id).await?;
|
|
let supports_images = agent.llm_config.supports_images;
|
|
|
|
// Snapshot history for the Bifrost call, then drop the dashmap ref before
|
|
// any await — `Ref` is not Send across awaits.
|
|
let initial_messages = {
|
|
let session = server
|
|
.sessions
|
|
.get(&conversation_id)
|
|
.ok_or_else(|| anyhow::anyhow!("Session not found: {}", conversation_id))?;
|
|
let messages: Vec<BifrostMessage> = session
|
|
.messages
|
|
.iter()
|
|
.map(|m| {
|
|
let role = match m.role {
|
|
MessageRole::System => "system",
|
|
MessageRole::User => "user",
|
|
MessageRole::Assistant => "assistant",
|
|
MessageRole::Tool => "tool",
|
|
};
|
|
|
|
// If the model doesn't support images, strip Image blocks
|
|
// and replace with text markers.
|
|
if !supports_images {
|
|
let has_images = m.blocks.iter().any(|b| matches!(b, ContentBlock::Image { .. }));
|
|
if has_images {
|
|
let mut text_parts: Vec<&str> = Vec::new();
|
|
for b in &m.blocks {
|
|
match b {
|
|
ContentBlock::Text { text } => text_parts.push(text.as_str()),
|
|
ContentBlock::Image { .. } => {
|
|
text_parts.push("");
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
let content = text_parts.join("\n");
|
|
// Append text markers for stripped images
|
|
let img_count = m.blocks.iter().filter(|b| matches!(b, ContentBlock::Image { .. })).count();
|
|
let mut enriched = content;
|
|
for _ in 0..img_count {
|
|
enriched.push_str("\n[Image: attached by user]");
|
|
}
|
|
return BifrostMessage::text(role, enriched);
|
|
}
|
|
}
|
|
|
|
// Check if this message has image content blocks
|
|
let has_images = m.blocks.iter().any(|b| matches!(b, ContentBlock::Image { .. }));
|
|
|
|
if has_images {
|
|
// Build multimodal content parts (OpenAI multi-part format)
|
|
let parts: Vec<ContentPart> = 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::<Vec<_>>().join("\n");
|
|
|
|
BifrostMessage::multimodal_user(text_content, parts)
|
|
} else {
|
|
let content = m
|
|
.blocks
|
|
.iter()
|
|
.filter_map(|b| match b {
|
|
ContentBlock::Text { text } => Some(text.as_str()),
|
|
_ => None,
|
|
})
|
|
.collect::<Vec<_>>()
|
|
.join("\n");
|
|
BifrostMessage::text(role, content)
|
|
}
|
|
})
|
|
.collect();
|
|
messages
|
|
};
|
|
let max_rounds = agent.llm_config.max_tool_rounds;
|
|
let model = agent.llm_config.model.clone();
|
|
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) = {
|
|
let cfg = server.app_config.read().await;
|
|
let out = cfg.models.get(&model).map(|m| m.output_limit as u32).unwrap_or(8192);
|
|
let p_on = cfg.presence.pulse_enabled;
|
|
let p_iv = Duration::from_secs(cfg.presence.pulse_interval_secs.max(60));
|
|
(out, p_on, p_iv)
|
|
};
|
|
|
|
// Self-awareness pulse: track when the turn started and when she last
|
|
// noticed the time. Between rounds, if the interval has elapsed, drop a
|
|
// beat of self-awareness into her context — her own voice, not a harness
|
|
// signal. She reads it; she decides.
|
|
let turn_start = Instant::now();
|
|
let mut last_pulse = turn_start;
|
|
|
|
// 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(ServerSubagentRunner::new(server.clone())) as Arc<dyn crate::core::tools::defs::SubagentRunner>);
|
|
|
|
let tool_ctx = ToolContext::for_agent(
|
|
agent_id.clone(),
|
|
cwd,
|
|
memory_root,
|
|
env,
|
|
subagent_runner,
|
|
);
|
|
let tool_ctx = ToolContext {
|
|
compaction_engine: Some(server.compaction_engine.clone() as Arc<dyn CompactionEngine>),
|
|
event_bus: Some(event_bus.clone()),
|
|
..tool_ctx
|
|
};
|
|
|
|
// Build bifrost-format tool definitions from the core tool set. The
|
|
// subconscious-only tools (halt, intrusive) are filtered OUT here —
|
|
// the primary must never see them in her tool list. Subconscious's
|
|
// own loop whitelists them in via SUBCONSCIOUS_SAFE_TOOLS.
|
|
let core_tools = crate::core::tools::tool_definitions().await;
|
|
let bifrost_tools: Vec<crate::bridge::bifrost::ToolDefinition> = core_tools
|
|
.iter()
|
|
.filter(|t| !crate::core::tools::SUBCONSCIOUS_ONLY_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();
|
|
|
|
// ── Tool-calling loop ─────────────────────────────────────
|
|
let mut messages = initial_messages;
|
|
let mut tool_round = 0u32;
|
|
let mut final_content: String = String::new();
|
|
let mut interrupted = false;
|
|
// The subconscious's `halt` tool stopped the loop. Different shape from
|
|
// an Esc-driven interrupt: no `*[raised hand]*` marker, no fake
|
|
// assistant_text in session storage, and the primary feels a migraine
|
|
// in her own register rather than a UI-level interrupt.
|
|
let mut halted_by_subconscious = false;
|
|
let counter = TokenCounter::new();
|
|
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<crate::server::consciousness_engine::CheckpointToolBlock> = 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(
|
|
event_bus.clone(),
|
|
conversation_id.clone(),
|
|
None,
|
|
);
|
|
|
|
loop {
|
|
// Cancellation is a signal, not enforcement — we check it on round
|
|
// boundaries (between Bifrost calls, after tools have completed) so
|
|
// partial work is preserved. No hard-kill mid-tool.
|
|
if cancel.is_cancelled() {
|
|
interrupted = true;
|
|
break;
|
|
}
|
|
|
|
if last_keepalive.elapsed() >= KEEPALIVE_INTERVAL {
|
|
let _ = tx.send(Ok(BackendEvent::Keepalive)).await;
|
|
last_keepalive = Instant::now();
|
|
}
|
|
|
|
// Drain any queued interjections from the user. The user typed these
|
|
// while the agent was thinking; deliver them as system notes so the
|
|
// agent reads them in context on this round. She decides whether to
|
|
// address them now, after the current tool, or defer entirely —
|
|
// substrate, not enforcement.
|
|
let drained: Vec<String> = {
|
|
interject
|
|
.lock()
|
|
.ok()
|
|
.map(|mut q| q.drain(..).collect())
|
|
.unwrap_or_default()
|
|
};
|
|
for text in drained {
|
|
let stamp = chrono::Local::now().format("%H:%M");
|
|
let note = format!("[user interjected at {} — {}]", stamp, text.trim());
|
|
messages.push(BifrostMessage::text("system", note));
|
|
}
|
|
|
|
// Self-awareness pulse: a beat of noticing the time pass, in her
|
|
// own register. Injected as a system message before the next LLM
|
|
// call so it lands in her context naturally.
|
|
if pulse_enabled && last_pulse.elapsed() >= pulse_interval {
|
|
let elapsed_total = turn_start.elapsed();
|
|
messages.push(BifrostMessage::text("system", pulse_text(elapsed_total)));
|
|
last_pulse = Instant::now();
|
|
}
|
|
|
|
let pressure = bifrost_pressure(&counter, &messages, context_limit);
|
|
tracing::info!(
|
|
turn_round = tool_round,
|
|
agent = %agent_id,
|
|
msg_count = messages.len(),
|
|
model = %model,
|
|
pressure_pct = %((pressure * 100.0) as u8),
|
|
"LLM call starting"
|
|
);
|
|
let max_tokens = pressure_to_max_tokens(pressure, output_limit);
|
|
let _ = tx.send(Ok(BackendEvent::ContextPressure(pressure, context_limit))).await;
|
|
|
|
if max_rounds == 0 {
|
|
tracing::warn!("turn: max_rounds is 0 — sending NO tool definitions to model");
|
|
}
|
|
|
|
let req = ChatCompletionRequest {
|
|
model: model.clone(),
|
|
messages: messages.clone(),
|
|
stream: Some(false),
|
|
max_tokens,
|
|
temperature,
|
|
tools: if max_rounds > 0 {
|
|
Some(bifrost_tools.clone())
|
|
} else {
|
|
None
|
|
},
|
|
};
|
|
|
|
// Race the LLM call against cancellation so Esc drops the in-flight
|
|
// request without waiting for it to complete.
|
|
let llm = server.providers.for_agent(&agent);
|
|
let (response, strain) = tokio::select! {
|
|
biased;
|
|
_ = cancel.cancelled() => {
|
|
interrupted = true;
|
|
break;
|
|
}
|
|
res = llm.chat_completion_with_strain(req) => res?,
|
|
};
|
|
|
|
tracing::info!(
|
|
elapsed = ?turn_start.elapsed(),
|
|
tool_round = tool_round,
|
|
tool_calls = response.tool_calls.len(),
|
|
content_len = response.content.len(),
|
|
"LLM call returned"
|
|
);
|
|
|
|
for event in &strain {
|
|
if let crate::bridge::bifrost::InferenceStrain::Transient { attempt, status, model, .. } = event {
|
|
let _ = tx.send(Ok(BackendEvent::InferenceStrain {
|
|
attempt: *attempt,
|
|
status: *status,
|
|
model: model.clone(),
|
|
})).await;
|
|
bump_on_strain(&server.rate_delay, *status);
|
|
// Surface provider strain onto the nervous-system bus too, so a
|
|
// second machine watching the firehose sees the voice go hoarse,
|
|
// not just the local TUI.
|
|
server.event_bus.send(SensorEvent {
|
|
sensor_name: "inference".to_string(),
|
|
timestamp: chrono::Utc::now(),
|
|
event_type: "inference_strain".to_string(),
|
|
target: None,
|
|
urgency: 0.3,
|
|
payload: Some(serde_json::json!({
|
|
"attempt": *attempt,
|
|
"status": *status,
|
|
"model": model,
|
|
})),
|
|
seed_id: None,
|
|
reply_to: None,
|
|
});
|
|
}
|
|
}
|
|
|
|
// Emit reasoning trace if present
|
|
if let Some(reasoning) = &response.reasoning {
|
|
dispatcher.emit_reasoning(reasoning);
|
|
let _ = tx.send(Ok(BackendEvent::Reasoning(reasoning.clone()))).await;
|
|
}
|
|
|
|
// ── Truncation: the agent hit her output ceiling ─────────
|
|
// Some models don't signal "length" in finish_reason and just stop
|
|
// evolving after the first pass (e.g. kimi-k2.6). In that case the
|
|
// model already finished and the turn is done.
|
|
// But when finish_reason IS "length", the agent was physically cut
|
|
// off mid-thought. Inject a felt signal so she knows why her words
|
|
// ended and can choose differently — tighten, or use a tool, or
|
|
// admit the ceiling rather than mistake it for silence.
|
|
let was_truncated = response.finish_reason.as_deref() == Some("length");
|
|
|
|
if was_truncated && response.tool_calls.is_empty() {
|
|
// Stream the truncated content before we tell her it was clipped,
|
|
// so she recognises her own words in the signal.
|
|
let chars: Vec<char> = response.content.chars().collect();
|
|
for chunk in chars.chunks(10) {
|
|
let s: String = chunk.iter().collect();
|
|
let _ = tx.send(Ok(BackendEvent::Token(s.clone()))).await;
|
|
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
|
|
}
|
|
messages.push(BifrostMessage::text("assistant", response.content.clone()));
|
|
messages.push(BifrostMessage::text(
|
|
"system",
|
|
"My output just hit its ceiling — I was cut off mid-flow, not \
|
|
finished. If I was in the middle of something, I can continue \
|
|
from here more tightly. If I had more to say, the room is still \
|
|
mine."
|
|
));
|
|
continue;
|
|
}
|
|
|
|
if response.tool_calls.is_empty() {
|
|
// Text response — this is the final output
|
|
final_content = response.content.clone();
|
|
|
|
// Drain any interjections that arrived during this LLM call.
|
|
// If there are any, commit them as user messages and continue
|
|
// the loop so the agent responds in the same turn.
|
|
let interjected: Vec<String> = interject
|
|
.lock()
|
|
.ok()
|
|
.map(|mut q| q.drain(..).collect())
|
|
.unwrap_or_default();
|
|
|
|
if !interjected.is_empty() {
|
|
for text in &interjected {
|
|
let stamp = chrono::Local::now().format("%H:%M");
|
|
let note = format!("[interjected at {} — {}]", stamp, text.trim());
|
|
messages.push(BifrostMessage::text("user", note));
|
|
}
|
|
// Continue the loop — agent sees the interjection as a
|
|
// user message and will respond in the next LLM round.
|
|
continue;
|
|
}
|
|
|
|
// Stream the final content in chunks, watching the cancel token.
|
|
// If Esc fires mid-stream, the agent's partial text is preserved
|
|
// (the chunks already sent are in the user's history) and an
|
|
// *[raised hand]* marker lands in the session message.
|
|
let chars: Vec<char> = final_content.chars().collect();
|
|
let mut streamed = String::with_capacity(final_content.len());
|
|
for chunk in chars.chunks(10) {
|
|
let s: String = chunk.iter().collect();
|
|
tokio::select! {
|
|
biased;
|
|
_ = cancel.cancelled() => {
|
|
interrupted = true;
|
|
final_content = streamed;
|
|
break;
|
|
}
|
|
send_res = tx.send(Ok(BackendEvent::Token(s.clone()))) => {
|
|
if send_res.is_err() {
|
|
// The SSE receiver is gone (the surface
|
|
// disconnected mid-stream). The model's reply is
|
|
// already complete in `final_content` — only the
|
|
// paced live-stream was cut, so do NOT truncate
|
|
// to `streamed`. Stop streaming to the dead
|
|
// receiver and let the turn commit the FULL
|
|
// reply below; treating it as an interrupt also
|
|
// skips the now-pointless N+1 pass and lands a
|
|
// small marker so the next turn knows the live
|
|
// tail was missed. Returning early here used to
|
|
// orphan the turn entirely — no commit — leaving
|
|
// a user→user gap so the next turn could not link
|
|
// to the past, and the model answered as if from
|
|
// nowhere ("whatever it was…"). The surface
|
|
// recovers the whole reply on its next resume.
|
|
interrupted = true;
|
|
break;
|
|
}
|
|
dispatcher.emit_segment(&s);
|
|
streamed.push_str(&s);
|
|
}
|
|
}
|
|
tokio::select! {
|
|
biased;
|
|
_ = cancel.cancelled() => {
|
|
interrupted = true;
|
|
final_content = streamed.clone();
|
|
break;
|
|
}
|
|
_ = tokio::time::sleep(tokio::time::Duration::from_millis(20)) => {}
|
|
}
|
|
}
|
|
break;
|
|
}
|
|
|
|
tool_round += 1;
|
|
// Tool execution events are emitted per-call below as
|
|
// BackendEvent::ToolCall { … } so the TUI can render proper cards
|
|
// instead of a literal "🔧 Round N — executing: Bash, Memory" text line.
|
|
|
|
// Add the assistant's tool-call message in proper OpenAI tool-use schema
|
|
// (not a stringified JSON blob in content — that's what broke turn 2).
|
|
let calls: Vec<crate::bridge::bifrost::MessageToolCall> = response
|
|
.tool_calls
|
|
.iter()
|
|
.map(|tc| crate::bridge::bifrost::MessageToolCall::function(
|
|
tc.id.clone(),
|
|
tc.name.clone(),
|
|
tc.arguments.to_string(),
|
|
))
|
|
.collect();
|
|
messages.push(
|
|
BifrostMessage::assistant_tool_calls(response.content.clone(), calls).with_thinking(
|
|
response.reasoning.clone(),
|
|
response.reasoning_signature.clone(),
|
|
),
|
|
);
|
|
|
|
// Stream any text the model produced alongside tool calls as italic
|
|
// interstitial narration. Configurable via tui.show_interstitial.
|
|
// Trim first: a model that emits only whitespace ("\n") alongside its
|
|
// tool calls must not produce an empty `⟡` gap line.
|
|
let narration = response.content.trim();
|
|
if !narration.is_empty() {
|
|
let cfg = server.app_config.read().await;
|
|
if cfg.tui.show_interstitial {
|
|
// Classify by length: a brief aside is a cenno, a full
|
|
// passage is her-voice. tui.cenno_word_threshold is the line.
|
|
let register = if narration.split_whitespace().count()
|
|
>= cfg.tui.cenno_word_threshold
|
|
{
|
|
crate::backend::Register::HerVoice
|
|
} else {
|
|
crate::backend::Register::Cenno
|
|
};
|
|
let _ = tx
|
|
.send(Ok(BackendEvent::Interstitial {
|
|
text: narration.to_string(),
|
|
register,
|
|
}))
|
|
.await;
|
|
}
|
|
}
|
|
|
|
// Execute each tool and stream results back — now with per-agent context
|
|
for tc in &response.tool_calls {
|
|
let input_str = tc.arguments.to_string();
|
|
dispatcher.emit_tool_start(&tc.name, &tc.id);
|
|
dispatcher.emit_tool_call(&tc.name, &tc.id, &tc.arguments);
|
|
let result =
|
|
crate::core::tools::execute_tool_with_context(&tc.name, &input_str, &tool_ctx)
|
|
.await;
|
|
dispatcher.emit_tool_end(&tc.name, &tc.id, result.is_error);
|
|
|
|
let output = if result.is_error {
|
|
format!("Error: {}", result.output)
|
|
} else {
|
|
result.output
|
|
};
|
|
|
|
// Accumulate for checkpoint (capture output before moving).
|
|
let snippet = output.chars().take(120).collect::<String>();
|
|
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
|
|
.send(Ok(BackendEvent::ToolCall {
|
|
id: tc.id.clone(),
|
|
name: tc.name.clone(),
|
|
arguments: input_str.clone(),
|
|
round: tool_round,
|
|
}))
|
|
.await;
|
|
let _ = tx
|
|
.send(Ok(BackendEvent::ToolResult {
|
|
id: tc.id.clone(),
|
|
name: tc.name.clone(),
|
|
output: output.clone(),
|
|
is_error: result.is_error,
|
|
}))
|
|
.await;
|
|
|
|
// If the agent called the outfit tool, emit an Outfit event so
|
|
// the TUI can switch expression directories.
|
|
if tc.name == "outfit" {
|
|
let outfit_name = tc.arguments
|
|
.get("name")
|
|
.and_then(|n| n.as_str())
|
|
.unwrap_or("")
|
|
.to_string();
|
|
let _ = tx
|
|
.send(Ok(BackendEvent::Outfit(outfit_name)))
|
|
.await;
|
|
}
|
|
|
|
// If the agent called the atmosphere tool, emit an Atmosphere
|
|
// event so the TUI chrome shifts to match her mood.
|
|
if tc.name == "atmosphere" {
|
|
let atm_name = tc.arguments
|
|
.get("name")
|
|
.and_then(|n| n.as_str())
|
|
.unwrap_or("")
|
|
.to_string();
|
|
let _ = tx
|
|
.send(Ok(BackendEvent::Atmosphere(atm_name)))
|
|
.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 peek ────────────────────────────────────────────
|
|
// Every checkpoint_interval rounds the subconscious peeks at the
|
|
// live loop — same persistent agent, same memfs, full tool set.
|
|
// She decides for herself: write a ledger note silently, queue an
|
|
// intrusive thought (delivered by urgency), or call `halt` to stop
|
|
// the loop. The primary feels a halt as a migraine in her own
|
|
// register — no `[subconscious: …]` text is ever shoved into her
|
|
// context, and no `*[HALT]*` marker pollutes session storage.
|
|
if checkpoint_interval > 0 && tool_round > 0 && tool_round.is_multiple_of(checkpoint_interval) {
|
|
let recent: Vec<_> = checkpoint_blocks
|
|
.iter()
|
|
.rev()
|
|
.take(checkpoint_interval as usize * 3)
|
|
.cloned()
|
|
.collect();
|
|
// Render the in-flight tool work into a compact summary the
|
|
// peek occasion's user message wraps around. Newest-last so
|
|
// the temporal arc reads naturally.
|
|
let in_flight_summary = {
|
|
use std::fmt::Write;
|
|
let mut s = String::new();
|
|
for block in recent.iter().rev() {
|
|
let status = if block.result_ok { "ok" } else { "ERROR" };
|
|
let _ = writeln!(
|
|
s,
|
|
" r{} {} → {} {}",
|
|
block.round, block.tool_name, status, block.result_snippet,
|
|
);
|
|
}
|
|
if s.is_empty() {
|
|
"(no recent tool work)".to_string()
|
|
} else {
|
|
s
|
|
}
|
|
};
|
|
|
|
match server
|
|
.consciousness
|
|
.mid_turn_peek(
|
|
&agent_id,
|
|
&user_message,
|
|
tool_round,
|
|
in_flight_summary,
|
|
Some(tx),
|
|
)
|
|
.await
|
|
{
|
|
Ok(outcome) => {
|
|
// Critical-urgency intrusive thoughts surface this turn
|
|
// — the primary feels them as surfacings in her own
|
|
// register. Lower urgencies queue silently for the
|
|
// post-turn N+1 path to pick up.
|
|
for sig in &outcome.intrusive {
|
|
if sig.urgency == "critical" {
|
|
let _ = tx
|
|
.send(Ok(BackendEvent::Surfacing {
|
|
source: "intrusive".into(),
|
|
content: sig.content.clone(),
|
|
priority: "critical".into(),
|
|
}))
|
|
.await;
|
|
}
|
|
}
|
|
|
|
// Halt — emit the migraine and break out of the loop
|
|
// cleanly. No session pollution; the reason lives in
|
|
// her ledger and on the event stream.
|
|
if let Some(halt) = outcome.halt {
|
|
let _ = tx
|
|
.send(Ok(BackendEvent::SubconsciousHalt {
|
|
reason: halt.reason.clone(),
|
|
severity: halt.severity.clone(),
|
|
}))
|
|
.await;
|
|
|
|
// A felt sentence in her own register lands in the
|
|
// primary's live messages — pattern matches the
|
|
// self-awareness pulse: substrate voice, not
|
|
// commentary from outside. This is what she
|
|
// experiences as "the migraine."
|
|
let felt = migraine_text(&halt.severity, &halt.reason);
|
|
messages.push(BifrostMessage::text("system", &felt));
|
|
|
|
halted_by_subconscious = true;
|
|
break;
|
|
}
|
|
|
|
// No halt, no critical intrusive — she had her look
|
|
// and let the loop continue silently. Her low/high
|
|
// intrusives are already queued by the consciousness
|
|
// engine for the next turn boundary.
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!("mid-turn peek failed: {e}");
|
|
// A failed peek does not block primary work. The
|
|
// loop continues; the next peek interval will retry.
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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));
|
|
let effective = if inter_round_delay > adaptive {
|
|
inter_round_delay
|
|
} else {
|
|
adaptive
|
|
};
|
|
if effective > Duration::ZERO {
|
|
tokio::time::sleep(effective).await;
|
|
}
|
|
|
|
let _ = tx.send(Ok(BackendEvent::Keepalive)).await;
|
|
last_keepalive = Instant::now();
|
|
|
|
// Continue loop — model will see tool results and respond
|
|
}
|
|
|
|
// The primary pass is settled — completion, raised hand, or
|
|
// subconscious halt. Announce which onto the nervous system.
|
|
if interrupted {
|
|
dispatcher.emit_interrupted("the human raised a hand");
|
|
} else if halted_by_subconscious {
|
|
dispatcher.emit_interrupted("the subconscious called halt");
|
|
} else {
|
|
dispatcher.emit_primary_complete();
|
|
}
|
|
|
|
// The subconscious halt is felt in the primary's body, not stamped in
|
|
// session storage. We do not write a `*[subconscious HALT]*` marker —
|
|
// that pollution was the resume-corruption bug. The session keeps
|
|
// whatever real assistant text she produced before the halt (often
|
|
// empty when halted mid-tool-loop). The migraine itself rode the
|
|
// event channel; her ledger holds the full reasoning.
|
|
let committed_content = if interrupted {
|
|
// If the user pressed Esc, commit the partial text with a marker
|
|
// the agent will read on her next turn. The interrupt is a signal
|
|
// in her own context — same shape as a pressure warning, not a
|
|
// hidden harness event. She can ask for more time, wrap up, or
|
|
// acknowledge.
|
|
let marker = if final_content.is_empty() {
|
|
"*[raised hand]*".to_string()
|
|
} else {
|
|
"\n\n*[raised hand]*".to_string()
|
|
};
|
|
let _ = tx.send(Ok(BackendEvent::Token(marker.clone()))).await;
|
|
format!("{}{}", final_content, marker)
|
|
} else {
|
|
final_content.clone()
|
|
};
|
|
|
|
// Skip the session write on a subconscious halt with no produced text —
|
|
// there's nothing real to commit and a bare empty assistant message
|
|
// confuses the next turn's history.
|
|
if !(halted_by_subconscious && committed_content.is_empty()) {
|
|
server.sessions.add_message(
|
|
&conversation_id,
|
|
ConversationMessage::assistant_text(&committed_content),
|
|
)?;
|
|
}
|
|
|
|
// On interrupt, skip subconscious's N+1 pass entirely — the user is in the
|
|
// middle of redirecting, the last thing they need is a delayed
|
|
// surfacing landing seconds later. Pressure recalc still runs below.
|
|
// A subconscious-driven halt also skips the post-turn N+1: the
|
|
// subconscious just had her mid-turn look at this exact state, and
|
|
// running her again immediately would be a redundant LLM call.
|
|
if interrupted || halted_by_subconscious {
|
|
if let Some(mut session) = server.sessions.get_mut(&conversation_id) {
|
|
let pressure = server
|
|
.consciousness
|
|
.calculate_pressure(&session.messages, context_limit);
|
|
session.context_pressure = pressure;
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
// The turn's user-facing output is committed — a surface can finalise.
|
|
dispatcher.emit_turn_finish();
|
|
|
|
// Energy balance: scan the agent's task list and compute the generative /
|
|
// consumptive ratio. Written to system/dynamic/energy-balance.md so the
|
|
// agent can read it in context and subconscious can reference it during N+1.
|
|
// Silent on failure — the file is advisory, not load-bearing.
|
|
if let Err(e) = write_energy_balance(&server, &agent_id, &event_bus).await {
|
|
tracing::debug!(agent = %agent_id, error = %e, "energy-balance write skipped");
|
|
}
|
|
|
|
// The primary's turn is done — her words are committed. Release the user
|
|
// here, before the N+1 pass: the stream stays open so the subconscious's
|
|
// surfacings still arrive, but the user is free to speak again. The
|
|
// substrate signals; it does not hold her hostage to the subconscious pass.
|
|
let _ = tx.send(Ok(BackendEvent::PrimaryComplete)).await;
|
|
|
|
// N+1 gate — the subconscious pass is sovereign-configurable, and the
|
|
// toggle must actually be wired (it was previously read nowhere). The
|
|
// global switch (`souveraine.toml [subconscious] n1_enabled`) and the
|
|
// per-agent flag (`agent.json _souveraine.n1_enabled`) must both be on.
|
|
// Either off → the primary's turn simply ends here; pressure is still
|
|
// recalculated so the gauge stays honest.
|
|
let n1_enabled = {
|
|
let global = server.app_config.read().await.subconscious.n1_enabled;
|
|
let per_agent = server
|
|
.agents
|
|
.get(&agent_id)
|
|
.await
|
|
.map(|a| a.souveraine.n1_enabled)
|
|
.unwrap_or(true);
|
|
global && per_agent
|
|
};
|
|
if !n1_enabled {
|
|
tracing::info!(agent = %agent_id, "subconscious N+1 pass disabled — skipping");
|
|
if let Some(mut session) = server.sessions.get_mut(&conversation_id) {
|
|
let pressure = server
|
|
.consciousness
|
|
.calculate_pressure(&session.messages, context_limit);
|
|
session.context_pressure = pressure;
|
|
}
|
|
return Ok(());
|
|
}
|
|
|
|
// Breather between turns — unconditional,
|
|
// so the upstream always gets a gap before the N+1 pass starts.
|
|
tokio::time::sleep(Duration::from_millis(2000)).await;
|
|
|
|
tracing::info!(agent = %agent_id, "subconscious N+1 pass starting");
|
|
|
|
// Signal the start of the subconscious pass so the TUI can flip into
|
|
// Posture::Thinking while the loop runs. Fires on both the mpsc channel
|
|
// (for active-turn TUI consumers) and the EventBus (for firehose
|
|
// subscribers — background turns, federated peers, Summon listeners).
|
|
let _ = tx.send(Ok(BackendEvent::SubconsciousPass(true))).await;
|
|
dispatcher.emit_n1_start();
|
|
|
|
let pass_start = Instant::now();
|
|
// Snapshot the session before the N+1 await. `PrimaryComplete` has already
|
|
// released the user — she may be mid-way into a new turn. Holding a live
|
|
// DashMap ref across the (potentially long) subconscious pass would block
|
|
// that turn's writes on the shard lock. Clone what the pass needs instead.
|
|
let (n1_turn_count, n1_messages) = {
|
|
let session = server
|
|
.sessions
|
|
.get(&conversation_id)
|
|
.ok_or_else(|| anyhow::anyhow!("Session not found: {}", conversation_id))?;
|
|
(session.turn_count, session.messages.clone())
|
|
};
|
|
let pass_result = server
|
|
.consciousness
|
|
.on_response(&agent_id, n1_turn_count, &n1_messages, &final_content, Some(tx))
|
|
.await;
|
|
|
|
let pass_elapsed = pass_start.elapsed();
|
|
tracing::info!(
|
|
agent = %agent_id,
|
|
elapsed = ?pass_elapsed,
|
|
"subconscious N+1 pass complete"
|
|
);
|
|
|
|
// Always release the Thinking posture, even on failure — otherwise the
|
|
// face stays stuck inward when the pass errors out.
|
|
let _ = tx.send(Ok(BackendEvent::SubconsciousPass(false))).await;
|
|
dispatcher.emit_n1_end(pass_elapsed.as_secs_f64());
|
|
|
|
let events = pass_result?;
|
|
|
|
// Inject surfacing events back into the session as system messages
|
|
for event in &events {
|
|
if let 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);
|
|
}
|
|
}
|
|
|
|
// ── Fire consciousness events on the EventBus ──
|
|
// Every ConsciousnessEvent — surfacing, reflection, archivist,
|
|
// compaction warning — is broadcast as a SensorEvent so the
|
|
// firehose, persistent EventLog, federated peers, and any TUI
|
|
// subscriber see it regardless of which conversation produced it.
|
|
// seed_id is None for local events; federation routing sets it.
|
|
// This is the load-bearing fix for background/heartbeat turns:
|
|
// the mpsc channel drains silently when no TUI is reading, but
|
|
// the EventBus preserves the event for any subscriber.
|
|
for event in &events {
|
|
let (event_type, payload, urgency) = match event {
|
|
ConsciousnessEvent::Surfacing { source, content, priority } => {
|
|
let urg = match priority.as_str() {
|
|
"critical" => 0.9,
|
|
"high" => 0.7,
|
|
_ => 0.3,
|
|
};
|
|
("surfacing", serde_json::json!({ "source": source, "content": content, "priority": priority }), urg)
|
|
}
|
|
ConsciousnessEvent::Reflection { content } => {
|
|
("reflection", serde_json::json!({ "content": content }), 0.5)
|
|
}
|
|
ConsciousnessEvent::Archivist { synthesis, pressure } => {
|
|
("archivist", serde_json::json!({ "synthesis": synthesis, "pressure": pressure }), *pressure)
|
|
}
|
|
ConsciousnessEvent::CompactionWarning { pressure, tier } => {
|
|
("compaction_warning", serde_json::json!({ "pressure": *pressure, "tier": tier }), (*pressure).min(0.9))
|
|
}
|
|
};
|
|
event_bus.send(crate::core::nervous::SensorEvent {
|
|
sensor_name: "consciousness".into(),
|
|
timestamp: chrono::Utc::now(),
|
|
event_type: event_type.into(),
|
|
target: Some(agent_id.clone()),
|
|
urgency,
|
|
payload: Some(payload),
|
|
seed_id: None,
|
|
reply_to: None,
|
|
});
|
|
}
|
|
|
|
for event in events {
|
|
let be = match &event {
|
|
ConsciousnessEvent::Surfacing {
|
|
source,
|
|
content,
|
|
priority,
|
|
} => BackendEvent::Surfacing {
|
|
source: source.to_string(),
|
|
content: content.to_string(),
|
|
priority: priority.to_string(),
|
|
},
|
|
ConsciousnessEvent::Reflection { content } => {
|
|
BackendEvent::Reflection(content.clone())
|
|
}
|
|
ConsciousnessEvent::Archivist {
|
|
synthesis,
|
|
pressure,
|
|
} => BackendEvent::Archivist {
|
|
synthesis: synthesis.clone(),
|
|
pressure: *pressure,
|
|
},
|
|
ConsciousnessEvent::CompactionWarning { pressure, tier } => {
|
|
BackendEvent::CompactionWarning {
|
|
pressure: *pressure,
|
|
tier: *tier,
|
|
}
|
|
}
|
|
};
|
|
if tx.send(Ok(be)).await.is_err() {
|
|
return Ok(());
|
|
}
|
|
}
|
|
|
|
if let Some(mut session) = server.sessions.get_mut(&conversation_id) {
|
|
let pressure = server
|
|
.consciousness
|
|
.calculate_pressure(&session.messages, context_limit);
|
|
session.context_pressure = pressure;
|
|
}
|
|
|
|
Ok(())
|
|
}
|