Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591//! Print-mode session slash-command execution: `/compact`, `/goal`, `/reload`,//! and active Lua commands share the pa-core executor and stream the print-json//! shapes; a failed command rejects the prompt wait (stderr, exit 1, never the//! later prompts).
use std::sync::Arc;
use pa_core::session_engine::engine::SessionEngine;use pa_core::session_engine::session_commands::{ execute_session_command, session_command_echo_row, SessionCommandExecution, SessionCommandParams,};use pa_core::session_engine::slash_commands::SessionSlashCommand;use pa_types::ai::Model;use serde_json::{json, Value};
use crate::print_goal::PrintGoalSurface;
/// `/compact <args>`: the args are the summary-focus instructions.fn compact_custom_instructions(command: &SessionSlashCommand) -> Option<String> { let args = command.args.trim(); (!args.is_empty()).then(|| args.to_string())}
/// The successful manual `compaction_end` event: the TS `CompactionResult` wire shape.fn manual_compaction_end_success( execution: &SessionCommandExecution, custom_instructions: Option<&str>,) -> Value { let compaction = execution .compaction .as_ref() .expect("manual compaction_end success requires the compaction run"); let result = json!({ "summary": compaction.result.summary, "firstKeptEntryId": compaction.result.first_kept_entry_id, "tokensBefore": compaction.result.tokens_before, "details": compaction .entry .details .clone() .unwrap_or(json!({ "readFiles": [], "modifiedFiles": [] })), }); let mut event = json!({ "type": "compaction_end", "reason": "manual", "result": result, "aborted": false, "willRetry": false, }); if let Some(instructions) = custom_instructions { event["customInstructions"] = json!(instructions); } event}
/// The unsuccessful manual `compaction_end` event: a skip carries its message at/// `warning` severity, a failure the `Compaction failed: <message>` text.fn manual_compaction_end_unsuccessful( message: &str, severity: &str, custom_instructions: Option<&str>,) -> Value { let mut event = json!({ "type": "compaction_end", "reason": "manual", "aborted": false, "willRetry": false, "errorMessage": message, "errorSeverity": severity, }); if let Some(instructions) = custom_instructions { event["customInstructions"] = json!(instructions); } event}
/// Execute one session command and stream its surface. The driver owns the exit/// contract: the execution's `error` field carries the raw failure (stderr, exit 1).#[allow(clippy::too_many_arguments)]pub(crate) async fn execute_prompt_session_command( engine: &Arc<SessionEngine>, goal: &Arc<PrintGoalSurface>, model: &Model, api_key: Option<String>, command: &SessionSlashCommand, images: &[pa_agent::types::ImageContent],) -> SessionCommandExecution { // The `preparing`/`running` phase frames bookend the durable echo row, and // `/compact`'s `compaction_start` goes out before the summarizer runs. goal.emit_command_phase("preparing", &command.text).await; goal.emit_command_phase("running", &command.text).await; goal.emit_row_pair(&session_command_echo_row(command)); let is_compact = command.name == "compact"; let custom_instructions = compact_custom_instructions(command); if is_compact { let mut event = json!({ "type": "compaction_start", "reason": "manual" }); if let Some(instructions) = &custom_instructions { event["customInstructions"] = json!(instructions); } goal.emit_stream_event(&event); } let execution = { let mut params = SessionCommandParams { model, api_key }; execute_session_command(engine, &mut params, command, images).await }; // `/compact`: the settled `compaction_end` precedes any failure result row; a skip stays silent // beyond the event. if is_compact { if let Some(message) = execution .compaction .as_ref() .and_then(|compaction| compaction.ipython_state.as_ref()) { goal.emit_row_pair(message); } let end = if execution.compaction.is_some() { manual_compaction_end_success(&execution, custom_instructions.as_deref()) } else if let Some(skipped) = execution.compaction_skipped { manual_compaction_end_unsuccessful(skipped, "warning", custom_instructions.as_deref()) } else { let error = execution .error .as_deref() .unwrap_or("compaction did not run"); manual_compaction_end_unsuccessful( &format!("Compaction failed: {error}"), "error", custom_instructions.as_deref(), ) }; goal.emit_stream_event(&end); } // `/goal`: the state change publishes unconditionally, before the queue frame of a scheduled // continuation. if command.name == "goal" { goal.publish_goal_update_forced(engine).await; } // The queued continuation's preview frame rides while the command action is still the active // one. if let Some(continuation) = &execution.continuation_message { goal.emit_command_queue_hold(&command.text, continuation) .await; } // The executor's first row is the echo (already streamed); the result and // failure rows follow in order. for row in execution.messages.iter().skip(1) { goal.emit_row_pair(row); } if let Some(continuation) = &execution.continuation_message { goal.emit_command_queue_drain(continuation).await; } else { goal.emit_queue_idle().await; } execution}
#[cfg(test)]mod tests { use super::*; use crate::print_boundary::TurnBoundary; use crate::print_goal::PrintGoalSurface; use crate::test_support::FAUX_TEST_LOCK; use pa_core::session_engine::slash_commands::{parse_session_command, SlashCommandRegistry}; use serde_json::json;
type Frames = std::sync::Arc<std::sync::Mutex<Vec<Value>>>;
fn capture_sink() -> (Frames, crate::print_goal::EventSink) { let frames: Frames = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); let sink: crate::print_goal::EventSink = { let frames = Arc::clone(&frames); Arc::new(move |event: &Value| { frames.lock().unwrap().push(event.clone()); }) }; (frames, sink) }
/// The captured stream's compact event trace: one line per event with the /// kind and the distinguishing fields (phases, row types, goal status). fn trace(frames: &Frames) -> Vec<String> { frames .lock() .unwrap() .iter() .map(|event| match event["type"].as_str().unwrap_or_default() { "session_action_update" => { let actions = &event["actions"]; let active = &actions["active"]; if active.is_null() { format!( "sau:idle:{}", actions["queuedCount"].as_u64().unwrap_or_default() ) } else { format!( "sau:{}:{}", active["kind"].as_str().unwrap_or_default(), active["phase"].as_str().unwrap_or_default() ) } } "message_start" | "message_end" => format!( "{}:{}", event["type"].as_str().unwrap(), event["message"]["customType"].as_str().unwrap_or("message") ), "goal_update" => format!( "goal_update:{}", event["goal"]["status"].as_str().unwrap_or_default() ), "compaction_start" => "compaction_start:manual".to_string(), "compaction_end" => { if event.get("result").is_some() { "compaction_end:manual:success".to_string() } else { format!( "compaction_end:manual:{}", event["errorSeverity"].as_str().unwrap_or_default() ) } } other => other.to_string(), }) .collect() }
/// One durable transcript row as (customType, text head). async fn custom_rows(engine: &Arc<SessionEngine>) -> Vec<(String, String)> { engine .session .entries() .await .into_iter() .filter_map(|entry| match entry { pa_types::session::FileEntry::CustomMessage { payload, .. } => Some(( payload.custom_type.clone(), match &payload.content { pa_types::ai::UserContent::Text(text) => text.clone(), pa_types::ai::UserContent::Blocks(_) => String::new(), }, )), _ => None, }) .collect() }
struct Bed { engine: Arc<SessionEngine>, model: Model, goal: Arc<PrintGoalSurface>, frames: Frames, _dir: tempfile::TempDir, }
/// The faux engine bed (the `print_goal` test pattern): a persisted session /// over its own tempdir and the wired goal surface. async fn bed(script: Value) -> Bed { bed_with_settings(script, json!({})).await }
async fn bed_with_settings(script: Value, settings: Value) -> Bed { let dir = tempfile::TempDir::new().unwrap(); let agent_dir = dir.path().join("agent"); std::fs::create_dir_all(&agent_dir).unwrap(); std::fs::write(agent_dir.join("settings.json"), settings.to_string()).unwrap(); let parsed = pa_ai::faux::script::parse_faux_script(&script).unwrap(); let registration = pa_ai::faux::register_faux_provider(pa_ai::faux::RegisterFauxProviderOptions { models: Some(vec![parsed.model.clone()]), ..Default::default() }); registration.set_responses(parsed.responses); registration.set_repeat_last_response(parsed.repeat_last_response); let model = registration.get_model(); let stream_fn = pa_core::session_engine::provider_adapter::real_stream_fn(None, model.clone()); let agent_model: pa_agent::types::Model = serde_json::from_value(serde_json::to_value(&model).unwrap()).unwrap(); let session_manager = pa_core::session::manager::SessionManager::persisted( dir.path(), &dir.path().join("sessions"), ); let engine = Arc::new( pa_core::session_engine::engine::create_session( pa_core::session_engine::engine::SessionEngineConfig { on_late_sent_agent_message: None, context_startup_boundary: pa_core::session_engine::ContextStartupBoundary::Current, context_startup_admitted_input_ids: Vec::new(), on_initial_context_selection: None, ollama_cloud: None, semantic_edges: None, cron_store: None, cwd: dir.path().to_path_buf(), agent_dir, mcp_manager: None, model: Some(agent_model), thinking_level: None, stream_fn: Some(stream_fn), tools: Vec::new(), custom_system_prompt: None, prompt_guidelines: Vec::new(), generic_mcp_servers: Vec::new(), allow_recursion: None, session_manager: Some(session_manager), extra_host_handlers: None, conversation_log_path: None, additional_skill_paths: Vec::new(), additional_prompt_paths: Vec::new(), extra_builtin_skill_overrides: Vec::new(), rlm_subagent_host: None, rlm_usage_store: None, progress_note_emit: None, daemonless: true, rlm_depth: None, model_info: Some(model.clone()), prewarm_ipython_kernel: None, on_background_work_settled: None, queued_goal_context_purge: None, queued_steering_probe: None, should_stop_after_turn: None, image_model_router: None, steering_mode: None, follow_up_mode: None, }, ) .await .unwrap(), ); let (frames, sink) = capture_sink(); let goal = Arc::new(PrintGoalSurface::with_sink(true, sink)); goal.seed_publish_baseline(&engine).await; let _accounting = goal.wire_accounting(&engine, engine.session.agent()).await; crate::print_goal::wire_goal_continuation_hook( &engine, engine.session.agent(), &model, &goal, /*root_node*/ None, ); Bed { engine, model, goal, frames, _dir: dir, } }
fn command(text: &str) -> SessionSlashCommand { let registry = SlashCommandRegistry::builtin(); parse_session_command(®istry, text).expect("the test text is a session command") }
/// The driver's session-command branch (`print_runtime`'s loop body). async fn run_command(bed: &Bed, text: &str) -> Option<String> { let execution = execute_prompt_session_command( &bed.engine, &bed.goal, &bed.model, None, &command(text), &[], ) .await; if execution.error.is_some() { return execution.error; } if let Some(continuation) = execution.continuation_message { let mut boundary = TurnBoundary::new(false); bed.goal .run_session_command_continuation( &bed.engine, &mut boundary, &bed.model, None, &continuation, ) .await .unwrap(); } let mut boundary = TurnBoundary::new(false); bed.goal .drive_boundary(&bed.engine, &mut boundary, &bed.model, None) .await .unwrap(); None }
fn script(responses: &Value) -> Value { json!({ "engine": "faux", "modelId": "faux-1", "modelName": "Faux Model", "reasoning": false, "contextWindow": 128_000, "responses": responses, }) }
#[tokio::test] async fn goal_status_stream_matches_ts() { let _guard = FAUX_TEST_LOCK.lock().await; let test = bed(script(&json!([]))).await; assert_eq!(run_command(&test, "/goal status").await, None); assert_eq!( trace(&test.frames), vec![ "sau:session_command:preparing", "sau:session_command:running", "message_start:session_slash_command", "message_end:session_slash_command", "goal_update:idle", "message_start:session_slash_command_result", "message_end:session_slash_command_result", "sau:idle:0", ] ); assert_eq!( custom_rows(&test.engine).await, vec![ ( "session_slash_command".to_string(), "/goal status".to_string() ), ( "session_slash_command_result".to_string(), "No active goal.".to_string() ), ] ); // The live context carries the rows: the headless terminal selection sees the result. let state = test.engine.session.agent().state().await; let messages: Vec<pa_types::session::AgentMessage> = state .messages .iter() .filter_map(pa_core::session_engine::provider_adapter::json_round_trip) .collect(); let result = pa_core::session_engine::headless::select_headless_terminal_result(&messages); let primary = result.primary.expect("the result row is the primary"); assert_eq!(primary.stdout_text().as_deref(), Some("No active goal.")); }
#[tokio::test] async fn goal_start_admits_the_continuation() { let _guard = FAUX_TEST_LOCK.lock().await; let test = bed(script(&json!([{"text": "goal turn reply"}]))).await; assert_eq!(run_command(&test, "/goal ship it").await, None); // The queued continuation ran to the faux queue's exhaustion and the terminal error failed // the goal. assert_eq!( trace(&test.frames), vec![ "sau:session_command:preparing", "sau:session_command:running", "message_start:session_slash_command", "message_end:session_slash_command", "goal_update:active", "sau:session_command:running", "message_start:session_slash_command_result", "message_end:session_slash_command_result", "sau:idle:1", "sau:turn:preparing", "sau:turn:committing", "sau:turn:running", // The continuation turn's usage publish and the in-loop mint's bump (both // `goal_update:active`), then the terminal error fails the goal. "goal_update:active", "goal_update:active", "goal_update:error", "sau:idle:0", ] ); let rows = custom_rows(&test.engine).await; assert_eq!(rows[0].0, "session_slash_command"); assert_eq!(rows[1].1, "Goal active: ship it"); assert!(rows.iter().any(|row| row.0 == "goal_context")); }
/// The clear's reply reflects the action it took (the operator's 2026-09-25 bug /// report): a held goal record answers "Goal cleared.", never "No active goal."; an /// empty clear keeps the plain status. #[tokio::test] async fn goal_clear_answers_the_action_it_took() { let _guard = FAUX_TEST_LOCK.lock().await; // One scripted reply: the start's continuation turn consumes it, the next // mint hits the exhausted faux queue — a goal record is what a clear removes. let test = bed(script(&json!([{"text": "goal turn reply"}]))).await; assert_eq!(run_command(&test, "/goal ship it").await, None); let rows = custom_rows(&test.engine).await; assert_eq!(rows[1].1, "Goal active: ship it");
let last_result = |rows: &[(String, String)]| -> String { rows.iter() .rev() .find(|(custom_type, _)| custom_type == "session_slash_command_result") .map(|(_, text)| text.clone()) .expect("a result row") };
let trace_before_clear = trace(&test.frames).len(); assert_eq!(run_command(&test, "/goal clear").await, None); assert_eq!( last_result(&custom_rows(&test.engine).await), "Goal cleared." ); // The clear's forced publish announced the empty state. assert!( trace(&test.frames)[trace_before_clear..].contains(&"goal_update:idle".to_string()), "the clear never published the empty state: {:?}", trace(&test.frames) );
assert_eq!(run_command(&test, "/goal clear").await, None); assert_eq!( last_result(&custom_rows(&test.engine).await), "No active goal." ); assert_eq!(run_command(&test, "/goal status").await, None); assert_eq!( last_result(&custom_rows(&test.engine).await), "No active goal." ); }
#[tokio::test] async fn compact_skip_warns_without_a_row() { let _guard = FAUX_TEST_LOCK.lock().await; let test = bed(script(&json!([]))).await; assert_eq!(run_command(&test, "/compact").await, None); assert_eq!( trace(&test.frames), vec![ "sau:session_command:preparing", "sau:session_command:running", "message_start:session_slash_command", "message_end:session_slash_command", "compaction_start:manual", "compaction_end:manual:warning", "sau:idle:0", ] ); assert_eq!( custom_rows(&test.engine).await, vec![("session_slash_command".to_string(), "/compact".to_string())] ); }
#[tokio::test] async fn failed_command_carries_the_raw_error() { let _guard = FAUX_TEST_LOCK.lock().await; // Two long seed turns consume the queue; the compactable session's /compact then fails its // summarizer call. let settings = json!({ "compaction": {"enabled": true, "reserveTokens": 1, "keepRecentTokens": 10} }); let test = bed_with_settings(script(&json!([{"text": "r1"}, {"text": "r2"}])), settings).await; let long_seed = "seed turn one ".to_string() + &"x".repeat(15000); test.engine .session .prompt( &long_seed, pa_core::session_engine::PromptOptions::default(), ) .await .unwrap(); test.engine.session.agent().wait_for_idle().await; let long_seed_two = "seed turn two ".to_string() + &"x".repeat(15000); test.engine .session .prompt( &long_seed_two, pa_core::session_engine::PromptOptions::default(), ) .await .unwrap(); test.engine.session.agent().wait_for_idle().await; let error = run_command(&test, "/compact focus").await; assert_eq!( error.as_deref(), Some("compaction chain failed: Summarization failed: No more faux responses queued") ); let rows = custom_rows(&test.engine).await; assert_eq!(rows.last().unwrap().0, "session_slash_command_result"); assert!(rows.last().unwrap().1.starts_with("Command failed:")); }}