Something went wrong. Try again.
Personal fork of prime agent. Mirror of: https://mnemosyne.sigilzero.dev/sigilzero/prime-agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651//! Queued input: the item model, the lanes, admission, delivery batching,//! and queue recovery.use super::{ emit_worker_event_with, json, oneshot, Arc, Duration, EventPump, Mutex, Notify, Result, SessionCore, Value, VecDeque, WorkerRecoveryJournal, AGENT_MESSAGE_SOURCE, AUTONOMOUS_QUEUE_KEY,};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]pub enum Lane { Steering, FollowUp,}
impl Lane { pub(crate) fn as_str(self) -> &'static str { match self { Lane::Steering => "steering", Lane::FollowUp => "follow_up", } }}
/// How long the close paths (`shutdown`, `kill`) wait for aborted side/// question runs to queue their terminal cancelled events: headroom, not a gate.pub(crate) const SIDE_QUESTION_SETTLE_TIMEOUT: Duration = Duration::from_secs(3);
pub(crate) const QUEUED_INPUT_SUSPENDED: &str = "Cannot admit a session action while queued session input is suspended.";
/// Whether a queued-input suspension still has work it is meant to hold. The/// gate parks the rows that existed when it was set (user work); an agent-message/// delivery that arrives during the hold rides it but never extends it, so agent/// traffic cannot keep a session deaf on its own (issue #33).pub(crate) fn hold_is_protected(core: &SessionCore) -> bool { core.steering .iter() .chain(core.follow_up.iter()) .any(|item| item.agent_message.is_none())}
/// The item's turn-execution class: items co-deliver as one batched turn only/// within the same class; the direct-prompt hand-off never joins a batch.#[derive(Debug, Clone, Copy, PartialEq, Eq)]pub(crate) enum TurnPolicy { Queued, Injected, Direct,}
impl TurnPolicy { /// The journal record's string form (`worker.recovery` queue /// snapshots); the restore maps it back through the same names. pub(crate) fn journal_value(self) -> &'static str { match self { TurnPolicy::Queued => "queued", TurnPolicy::Injected => "injected", TurnPolicy::Direct => "direct", } }}
/// The turn-execution class restored from a wire `restore_actions` payload;/// an absent or unknown policy restores as the queued class.pub(crate) fn restored_turn_policy(payload: &Value) -> TurnPolicy { let timing = payload .get("executionPolicy") .and_then(|policy| policy.get("nextTurnContextTiming")) .and_then(Value::as_str); match timing { Some("preparation") => TurnPolicy::Direct, _ => TurnPolicy::Queued, }}
/// The wire text of an aborted turn's settle (the `turn_end` error frame and/// the waiting prompt's failure): aborted before an assistant message.pub(crate) const ABORTED_TURN_SETTLE_ERROR: &str = "No response produced.";
/// The wire text of a prompt cancelled before delivery (the/// queue-invisible abort path).pub(crate) const PROMPT_ABORTED_BEFORE_DELIVERY: &str = "Prompt aborted before delivery.";
/// The wire text of a queued prompt deleted through a queue mutation (TS/// `QueuedMessageError` verbatim).pub(crate) const QUEUED_PROMPT_DELETED: &str = "Queued prompt was deleted before delivery.";
/// The typed settle of one queued prompt: the variants classify the/// settle without reading the (provider-controllable) error text.#[derive(Debug, Clone, PartialEq, Eq)]pub(crate) enum TurnSettle { Completed, Aborted, /// The prompt was withdrawn before delivery; the text is the /// wire-facing reason. Withdrawn(String), /// The turn settled with an error; the text surfaces to the waiting /// caller. Failed(String),}
impl TurnSettle { /// The wire-facing failure text of the settle (`None` when the /// settle is a success). pub(crate) fn wire_error(&self) -> Option<String> { match self { TurnSettle::Completed => None, TurnSettle::Aborted => Some(ABORTED_TURN_SETTLE_ERROR.to_string()), TurnSettle::Withdrawn(text) | TurnSettle::Failed(text) => Some(text.clone()), } }}
/// Priority is applied only at admission. Existing lane positions (including user/// moves and restored snapshots) remain authoritative until another item arrives.#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]#[serde(rename_all = "snake_case")]pub enum QueuePriority { Human, Pinned, #[serde(other)] Background,}
impl QueuePriority { fn rank(self) -> u8 { match self { Self::Background => 0, Self::Human => 1, Self::Pinned => 2, } }}
/// Priority insertion: walk back only across a lower-priority suffix. Never/// re-sort a lane (moves and restored order can cross; equal priorities stay FIFO).pub(crate) fn enqueue_priority(lane: &mut VecDeque<QueuedItem>, item: QueuedItem) { let mut index = lane.len(); while index > 0 && lane[index - 1].priority.rank() < item.priority.rank() { index -= 1; } lane.insert(index, item);}
#[derive(Debug)]pub(crate) struct QueuedItem { pub(crate) message: String, pub(crate) priority: QueuePriority, /// The labeled queue-strip row (the snapshot and the active-action /// label read it); the turn's prompt text stays `message`. pub(crate) preview: Option<String>, /// An injected custom row that replaces this turn's user message (the /// RLM child terminal notices ride the follow-up lane this way). pub(crate) custom_message: Option<Value>, /// The original agent-message text when this item came from an /// `agent_message` delivery (the markers remove queued items by). pub(crate) agent_message: Option<String>, /// The scheduler's queue key: a later fire replaces the queued /// item with the same key instead of stacking. pub(crate) queue_key: Option<String>, /// The prompt-admission id (the `cancel_prompt_admission` /// bookkeeping); `None` when it carried none. pub(crate) admission_id: Option<String>, /// Images attached to the prompt (wire `images`: base64 plus mime /// type). pub(crate) images: Vec<pa_agent::types::ImageContent>, pub(crate) done: Option<oneshot::Sender<TurnSettle>>, /// The item shows in the queue projection and projects the active-action /// phases; injected continuations and a direct prompt admission stay invisible. pub(crate) queue_visible: bool, /// The item's turn-execution class (see [`TurnPolicy`]): the batch /// gathering's compatibility gate. pub(crate) policy: TurnPolicy, /// Membership of the one-shot forced steering batch: armed items co-deliver as /// one batched turn even under "one-at-a-time". Never journaled. pub(crate) forced_batch: bool,}
/// Parse the wire `images` array: entries without payload data or a/// mime type are dropped, not failed — the text still admits.pub(crate) fn parse_prompt_images(payload: &Value) -> Vec<pa_agent::types::ImageContent> { let Some(images) = payload.get("images").and_then(Value::as_array) else { return Vec::new(); }; images .iter() .filter_map(|image| { if image.get("type").and_then(Value::as_str) != Some("image") { return None; } let data = image.get("data").and_then(Value::as_str)?; let mime_type = image.get("mimeType").and_then(Value::as_str)?; Some(pa_agent::types::ImageContent { data: data.to_string(), mime_type: mime_type.to_string(), }) }) .collect()}
/// The pending queue lanes of a session (journal persistence payload): the/// full parked rows, so recovery restores a queued heartbeat's component.pub(crate) struct QueueLanes { pub(crate) steering: Vec<crate::journal::WorkerQueueItemRecord>, pub(crate) follow_up: Vec<crate::journal::WorkerQueueItemRecord>,}
/// The wire `customMessage` of a prompt/follow-up command: an injected custom/// row that replaces the turn's user row. `Err` rejects loudly — a malformed/// notice must not degrade into a plain prompt.pub(crate) fn parse_custom_message(value: Option<&Value>) -> Result<Option<Value>, String> { let Some(value) = value else { return Ok(None); }; if value.is_null() { return Ok(None); } let invalid = "Invalid customMessage: expected a custom message object with a customType"; let Some(object) = value.as_object() else { return Err(invalid.to_string()); }; if object.get("role").and_then(Value::as_str) != Some("custom") { return Err(invalid.to_string()); } let custom_type = object .get("customType") .and_then(Value::as_str) .filter(|kind| !kind.is_empty()); if custom_type.is_none() { return Err(invalid.to_string()); } Ok(Some(value.clone()))}
/// One queue-lane recovery checkpoint: the verdict and the persisted/// snapshot come from one locked read (no stale verdict).#[derive(Clone, Copy)]pub(crate) enum QueueCheckpoint { /// The lanes hold admitted live work: `busy = true`. Admitted { operation: &'static str }, /// The verdict follows the lanes (`busy = lanes remain queued`); mutations /// with no TS record use it too, so the journal keeps no stale verdict. Settle { operation: &'static str },}
/// Clear a withdrawn run's note before publishing an idle, empty queue./// Call while holding the core lock; summary readers take core then note too.pub(crate) fn clear_withdrawn_progress_note( core: &SessionCore, engine: &dyn crate::engine::SessionEngine,) { if !core.busy && core.steering.is_empty() && core.follow_up.is_empty() { engine.clear_progress_note(); }}
/// Write one queue-lane recovery checkpoint: under the recovery lock (then/// the core lock) the lanes and the busy verdict record from the same read.pub(crate) fn checkpoint_queue_recovery( recovery: &std::sync::Mutex<Option<WorkerRecoveryJournal>>, core_lock: &std::sync::Mutex<SessionCore>, checkpoint: QueueCheckpoint,) { let mut guard = recovery.lock().unwrap(); let Some(journal) = guard.as_mut() else { return; }; // The lanes are read under the recovery lock (a microsecond core hold — // never across the journal's fsyncs): no persist interleaves this read. let (active_session_id, session_id, session_file, lanes, turn_in_flight) = { let core = core_lock.lock().unwrap(); ( core.active_session_id.clone(), core.store .as_ref() .map(|s| s.session_id().to_string()) .unwrap_or_default(), core.store .as_ref() .map(|s| s.path.to_string_lossy().to_string()), queue_lanes(&core), core.busy, ) }; let (busy, operation) = match checkpoint { QueueCheckpoint::Admitted { operation } => (true, operation), // A settled verdict never comes from the lanes alone: a withdrawal landing // mid-turn must not flip the journal to idle while the turn still streams, // or a crash in that window parks live work. QueueCheckpoint::Settle { operation } => ( turn_in_flight || !lanes.steering.is_empty() || !lanes.follow_up.is_empty(), operation, ), }; // The verdict never publishes over a snapshot that did not // persist: the pair rides ONE durable append. let _ = journal.record_queue_checkpoint( &active_session_id, &session_id, session_file.as_deref(), busy, operation, &lanes.steering, &lanes.follow_up, );}
pub(crate) fn queue_lanes(core: &SessionCore) -> QueueLanes { fn items(lane: &VecDeque<QueuedItem>) -> Vec<crate::journal::WorkerQueueItemRecord> { lane.iter() .map(|item| crate::journal::WorkerQueueItemRecord { message: item.message.clone(), priority: Some(item.priority), preview: item.preview.clone(), custom_message: item.custom_message.clone(), queue_key: item.queue_key.clone(), queue_visible: item.queue_visible, policy: item.policy.journal_value().to_string(), }) .collect() } QueueLanes { steering: items(&core.steering), follow_up: items(&core.follow_up), }}
/// The lane's front item anchors the delivery; under queue mode "all" — or/// the forced steering batch — the same-class prefix behind it joins as/// co-delivered rows of ONE turn. A non-batchable front delivers solo.pub(crate) fn gather_delivery_batch(core: &mut SessionCore, lane: Lane) -> Vec<QueuedItem> { let (items, mode) = match lane { Lane::Steering => (&mut core.steering, core.steering_mode.as_str()), Lane::FollowUp => (&mut core.follow_up, core.follow_up_mode.as_str()), }; let Some(first) = items.front() else { return Vec::new(); }; // The armed set forces "all" only when the front item is armed; an un-armed // front disarms the batch once no armed item remains queued. let forced = lane == Lane::Steering && core.forced_all_steering && first.forced_batch; let mut batch = Vec::new(); if lane == Lane::Steering && core.forced_all_steering && !forced && !items.iter().any(|item| item.forced_batch) { core.forced_all_steering = false; } if first.custom_message.is_some() || first.policy == TurnPolicy::Direct || crate::session_commands::parse_prompt_session_command(&first.message).is_some() { batch.push(items.pop_front().expect("front checked")); return batch; } let first_policy = first.policy; batch.push(items.pop_front().expect("front checked")); if forced || mode == "all" { while let Some(next) = items.front() { if next.policy != first_policy || next.custom_message.is_some() || (forced && !next.forced_batch) || crate::session_commands::parse_prompt_session_command(&next.message).is_some() { break; } batch.push(items.pop_front().expect("front checked")); } } batch}
/// Queue snapshot restore from the worker recovery journal (crash/respawn/// recovery): the latest persisted lanes for this session.pub(crate) fn restore_queue_snapshot( journal: &WorkerRecoveryJournal, active_session_id: &str,) -> (VecDeque<QueuedItem>, VecDeque<QueuedItem>) { fn pending(lanes: Vec<crate::journal::WorkerQueueItemRecord>) -> VecDeque<QueuedItem> { // Images on a queued prompt do not survive the restart; the // delivery rows ride the item record without them. lanes .into_iter() .map(|record| { let policy = record.policy(); // The journal retains the custom row, including the original // body. Reconstruct the marker used by clear/pause/status. let agent_message = record.custom_message.as_ref().and_then(|row| { (row.get("customType").and_then(Value::as_str) == Some(AGENT_MESSAGE_SOURCE)) .then(|| { row.get("details")? .get("message")? .as_str() .map(str::to_string) }) .flatten() }); QueuedItem { preview: record.preview, message: record.message, priority: record.priority.unwrap_or_else(|| { if record.custom_message.is_some() { QueuePriority::Background } else { QueuePriority::Human } }), custom_message: record.custom_message, agent_message, queue_key: record.queue_key, admission_id: None, images: Vec::new(), done: None, queue_visible: record.queue_visible, policy, forced_batch: false, } }) .collect() }
let mut steering = VecDeque::new(); let mut follow_up = VecDeque::new(); if let Some((steering_lanes, follow_up_lanes)) = journal.latest_queue_snapshot(active_session_id) { steering = pending(steering_lanes); follow_up = pending(follow_up_lanes); } (steering, follow_up)}
/// Admit one held autonomous continuation through the follow-up lane:/// the item runs as its own queue item after the current run settles.pub(crate) fn admit_autonomous_follow_up( recovery: &std::sync::Mutex<Option<WorkerRecoveryJournal>>, core: &Arc<Mutex<SessionCore>>, work_notify: &Arc<Notify>, text: String,) { { let mut core = core.lock().unwrap(); core.follow_up.push_back(QueuedItem { priority: QueuePriority::Background, preview: None, message: text, custom_message: None, agent_message: None, queue_key: Some(AUTONOMOUS_QUEUE_KEY.to_string()), admission_id: None, images: Vec::new(), done: None, queue_visible: false, policy: TurnPolicy::Injected, forced_batch: false, }); } // The admission checkpoint: a continuation admitted while idle is undelivered // live work the journal must prove (a kill before the settle would park it). checkpoint_queue_recovery( recovery, core, QueueCheckpoint::Admitted { operation: "follow_up_queued", }, ); work_notify.notify_waiters();}
pub(crate) fn admit_goal_follow_up( recovery: &std::sync::Mutex<Option<WorkerRecoveryJournal>>, core: &Arc<Mutex<SessionCore>>, events: &Arc<EventPump>, work_notify: &Arc<Notify>, work: crate::engine::GoalTurnEndWork,) { let (lane, follow_up) = match work { crate::engine::GoalTurnEndWork::BudgetLimitSteer(follow_up) => (Lane::Steering, follow_up), crate::engine::GoalTurnEndWork::Continuation(follow_up) => (Lane::FollowUp, follow_up), }; if let Some(goal) = &follow_up.goal_update { emit_worker_event_with(core, events, json!({ "type": "goal_update", "goal": goal })); } { let mut core = core.lock().unwrap(); let item = QueuedItem { priority: QueuePriority::Background, preview: None, message: follow_up.request.message, custom_message: follow_up.request.custom_message, agent_message: None, queue_key: None, admission_id: None, images: follow_up.request.images, done: None, queue_visible: false, policy: TurnPolicy::Injected, forced_batch: false, }; match lane { Lane::Steering => enqueue_priority(&mut core.steering, item), Lane::FollowUp => enqueue_priority(&mut core.follow_up, item), } } // The admission checkpoint: same idle-admission gap as the // autonomous continuation above. checkpoint_queue_recovery( recovery, core, QueueCheckpoint::Admitted { operation: match lane { Lane::Steering => "steer_queued", Lane::FollowUp => "follow_up_queued", }, }, ); // The runner re-checks the queue at its loop head, so the minted // turn runs as the next admitted turn. work_notify.notify_one();}
/// Admit one detached kernel bash completion notice: the row queues on the/// steering lane (busy sessions keep a visible steer row, idle sessions wake/// into it); a crash before the delivery revives with the row replaying.pub(crate) fn admit_bash_completion_notice( recovery: &std::sync::Mutex<Option<WorkerRecoveryJournal>>, core: &Arc<Mutex<SessionCore>>, work_notify: &Arc<Notify>, notice: &crate::engine::BashCompletionNotice, session_is_closed: impl Fn() -> bool,) { let row = pa_core::session_engine::messages::create_async_bash_completion_message( notice.pid, ¬ice.command, notice.exit_code, crate::util::now_ms(), ); let content = match &row.content { pa_types::ai::UserContent::Text(text) => text.clone(), pa_types::ai::UserContent::Blocks(_) => String::new(), }; // The busy sample and the push share ONE critical section: a turn starting // between them would queue an invisible row for a busy session. let mut core_guard = core.lock().unwrap(); // A notice that raced past the sink's first check is refused here or wiped // by the close's clear — never a completion turn for a closed session. if session_is_closed() { return; } let (policy, queue_visible) = if core_guard.busy { (TurnPolicy::Queued, true) } else { (TurnPolicy::Injected, false) }; { core_guard.steering.push_back(QueuedItem { priority: QueuePriority::Background, // The queue strip's labeled preview (TS `previewLabel`). preview: Some(format!( "{}: {content}", pa_core::session_engine::messages::ASYNC_BASH_COMPLETION_PREVIEW_LABEL )), message: content, custom_message: Some(crate::session_commands::custom_message_value(&row)), agent_message: None, queue_key: None, admission_id: None, images: Vec::new(), done: None, queue_visible, policy, forced_batch: false, }); } // The checkpoint re-locks the core (documented order: recovery lock // first), so the admission's guard must release first. drop(core_guard); // The busy-evidence checkpoint: the notice is undelivered live // work until its turn settles. checkpoint_queue_recovery( recovery, core, QueueCheckpoint::Admitted { operation: "steer_queued", }, ); work_notify.notify_one();}
/// Withdraw one queued bash completion notice: the kernel read the result/// before the notice delivered (pids are reused, so the command disambiguates).pub(crate) fn withdraw_bash_completion_notice( recovery: &std::sync::Mutex<Option<WorkerRecoveryJournal>>, core: &Arc<Mutex<SessionCore>>, notice: &crate::engine::BashConsumedNotice, engine: Option<&dyn crate::engine::SessionEngine>,) { let removed = { let mut core_guard = core.lock().unwrap(); let before = core_guard.steering.len() + core_guard.follow_up.len(); // The front-most match across the two lanes, never the // whole set. let mut withdrawn = false; let mut withdraw_one = |item: &QueuedItem| { if !withdrawn && is_bash_completion_notice_for(item, notice) { withdrawn = true; false } else { true } }; core_guard.steering.retain(&mut withdraw_one); core_guard.follow_up.retain(withdraw_one); if let Some(engine) = engine { clear_withdrawn_progress_note(&core_guard, engine); } before != core_guard.steering.len() + core_guard.follow_up.len() }; if removed { // The withdrawal refreshes the verdict so a consumed notice cannot keep // busy=true promising a revive the withdrawn row would replay. checkpoint_queue_recovery( recovery, core, QueueCheckpoint::Settle { operation: "queue_purged", }, ); }}
/// Whether one queued item is the completion notice for this pid+command/// (the details carry both — pids alone are reused).fn is_bash_completion_notice_for( item: &QueuedItem, notice: &crate::engine::BashConsumedNotice,) -> bool { let Some(row) = item.custom_message.as_ref() else { return false; }; if row.get("customType").and_then(Value::as_str) != Some(pa_core::session_engine::messages::ASYNC_BASH_COMPLETION_CUSTOM_TYPE) { return false; } let details = row.get("details").unwrap_or(&Value::Null); details.get("pid").and_then(Value::as_u64) == Some(u64::from(notice.pid)) && details.get("command").and_then(Value::as_str) == Some(notice.command.as_str())}