diff --git a/Cargo.lock b/Cargo.lock index 8e470168d..c7d707bcc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2254,6 +2254,7 @@ dependencies = [ "fabro-types", "fabro-util", "futures", + "pebble-coding-agent", "serde_json", "shlex", "tempfile", @@ -5856,7 +5857,7 @@ dependencies = [ [[package]] name = "pebble-agent" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=5fbb6c98e0d0c9eaea4df4930b37c3a4a98e3ba0#5fbb6c98e0d0c9eaea4df4930b37c3a4a98e3ba0" +source = "git+https://github.com/lithoscomputer/pebble?rev=3a76aeee74d6c4aef349b927b09455c5616b296e#3a76aeee74d6c4aef349b927b09455c5616b296e" dependencies = [ "async-trait", "futures-util", @@ -5873,7 +5874,7 @@ dependencies = [ [[package]] name = "pebble-coding-agent" version = "0.1.0" -source = "git+https://github.com/lithoscomputer/pebble?rev=5fbb6c98e0d0c9eaea4df4930b37c3a4a98e3ba0#5fbb6c98e0d0c9eaea4df4930b37c3a4a98e3ba0" +source = "git+https://github.com/lithoscomputer/pebble?rev=3a76aeee74d6c4aef349b927b09455c5616b296e#3a76aeee74d6c4aef349b927b09455c5616b296e" dependencies = [ "async-trait", "futures-util", diff --git a/Cargo.toml b/Cargo.toml index 7cad47863..e236c2746 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -112,8 +112,8 @@ futures-util = "0.3" # the merge commit once it lands. Pebble pins the same lithos-llm rev as # fabro, and its lockfile policy is that every shared crate resolves to the # version lithos-llm locks. -pebble-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "5fbb6c98e0d0c9eaea4df4930b37c3a4a98e3ba0" } -pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "5fbb6c98e0d0c9eaea4df4930b37c3a4a98e3ba0", features = ["mcp", "search-providers"] } +pebble-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "3a76aeee74d6c4aef349b927b09455c5616b296e" } +pebble-coding-agent = { git = "https://github.com/lithoscomputer/pebble", rev = "3a76aeee74d6c4aef349b927b09455c5616b296e", features = ["mcp", "search-providers"] } sandbox-driver = { git = "https://github.com/lithoscomputer/sandbox-driver", rev = "23062b6ad62ff4665cbbcb7dce037ec9c4c34318" } sandbox-driver-protocol = { git = "https://github.com/lithoscomputer/sandbox-driver", rev = "23062b6ad62ff4665cbbcb7dce037ec9c4c34318" } sandbox-driver-host = { git = "https://github.com/lithoscomputer/sandbox-driver", rev = "23062b6ad62ff4665cbbcb7dce037ec9c4c34318" } diff --git a/lib/components/fabro-acp/Cargo.toml b/lib/components/fabro-acp/Cargo.toml index 3e1962dbe..490b87004 100644 --- a/lib/components/fabro-acp/Cargo.toml +++ b/lib/components/fabro-acp/Cargo.toml @@ -12,6 +12,7 @@ runtime = [ "dep:fabro-sandbox", "dep:fabro-types", "dep:futures", + "dep:pebble-coding-agent", "dep:tokio", "dep:tokio-util", "dep:tracing", @@ -30,6 +31,7 @@ agent-client-protocol-tokio.workspace = true fabro-sandbox = { path = "../fabro-sandbox", optional = true } fabro-types = { path = "../../foundation/fabro-types", optional = true } fabro-util = { path = "../../foundation/fabro-util" } +pebble-coding-agent = { workspace = true, optional = true } serde_json.workspace = true shlex = "1" thiserror.workspace = true diff --git a/lib/components/fabro-acp/src/session.rs b/lib/components/fabro-acp/src/session.rs index 98a3ad9f9..91570ff9f 100644 --- a/lib/components/fabro-acp/src/session.rs +++ b/lib/components/fabro-acp/src/session.rs @@ -10,8 +10,9 @@ use agent_client_protocol::schema::{ use agent_client_protocol::util::MatchDispatch; use agent_client_protocol::{ActiveSession, Agent, Client, Error as ProtocolError, SessionMessage}; use fabro_sandbox::RunSandbox; -use fabro_types::{Principal, SteeringMessage}; use fabro_util::time::elapsed_ms; +use pebble_coding_agent::SteeringMessage; +use pebble_coding_agent::events::Actor; use tokio::sync::Notify; use tokio::sync::futures::Notified; use tokio::time::{sleep, timeout}; @@ -22,7 +23,7 @@ use crate::error::AcpError; use crate::transport::{SandboxAcpTransport, TransportState}; pub type AcpNaturalCompletionCallback = Arc bool + Send + Sync>; -pub type AcpSteerPromptCallback = Arc) + Send + Sync>; +pub type AcpSteerPromptCallback = Arc) + Send + Sync>; const CANCEL_GRACE_PERIOD: Duration = Duration::from_millis(500); @@ -49,7 +50,7 @@ impl AcpControlHandle { self.push_bounded(item, cap, false) } - pub fn interrupt(&self, _actor: Option) { + pub fn interrupt(&self) { { let mut state = self.state.lock().expect("ACP control lock poisoned"); if state.queue.is_empty() { @@ -352,9 +353,9 @@ async fn read_live_session( if !prompt_active { if let Some(message) = control_handle.pop_steer() { if let Some(on_steer_prompt) = on_steer_prompt { - on_steer_prompt(message.text.clone(), message.actor.clone()); + on_steer_prompt(message.text().to_string(), message.actor().cloned()); } - session.send_prompt(message.text)?; + session.send_prompt(message.text().to_string())?; prompt_active = true; cancel_sent = false; continue; diff --git a/lib/components/fabro-acp/tests/session.rs b/lib/components/fabro-acp/tests/session.rs index f896272b0..a98c45c76 100644 --- a/lib/components/fabro-acp/tests/session.rs +++ b/lib/components/fabro-acp/tests/session.rs @@ -11,8 +11,8 @@ use fabro_acp::{ }; use fabro_sandbox::test_support::{MockSandbox, MockStdioProcess}; use fabro_sandbox::{RunSandbox, local_sandbox, shell_quote}; -use fabro_types::SteeringMessage; use fabro_util::error::collect_chain; +use pebble_coding_agent::SteeringMessage; use tokio::fs::{read_to_string, write}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader, DuplexStream}; use tokio::process::Command; @@ -181,8 +181,7 @@ async fn steering_sends_followup_session_prompt_over_acp() { cancel_token: CancellationToken::new(), on_activity: Some(Arc::new(move || { if !queued_for_activity.swap(true, Ordering::AcqRel) { - handle_for_activity - .enqueue_bounded(SteeringMessage::new("please revise", None), 32); + handle_for_activity.enqueue_bounded(SteeringMessage::new("please revise"), 32); } })), live_control: Some(AcpLiveControl::new(control_handle)), @@ -253,10 +252,8 @@ async fn interrupt_then_steer_sends_cancel_then_followup_session_prompt_over_acp cancel_token: CancellationToken::new(), on_activity: Some(Arc::new(move || { if !queued_for_activity.swap(true, Ordering::AcqRel) { - handle_for_activity.interrupt_then_enqueue_bounded( - SteeringMessage::new("please revise", None), - 32, - ); + handle_for_activity + .interrupt_then_enqueue_bounded(SteeringMessage::new("please revise"), 32); } })), live_control: Some(AcpLiveControl::new(control_handle)), @@ -328,7 +325,7 @@ async fn inline_interrupt_terminates_agent_that_ignores_cancel() { cancel_token: CancellationToken::new(), on_activity: Some(Arc::new(move || { if !interrupted_for_activity.swap(true, Ordering::AcqRel) { - handle_for_activity.interrupt(None); + handle_for_activity.interrupt(); } })), live_control: Some(AcpLiveControl::new(control_handle)), diff --git a/lib/components/fabro-workflow/src/event.rs b/lib/components/fabro-workflow/src/event.rs index a35904ebf..ec0115925 100644 --- a/lib/components/fabro-workflow/src/event.rs +++ b/lib/components/fabro-workflow/src/event.rs @@ -23,5 +23,5 @@ pub use self::sink::{ RunEventLogger, RunEventPersistenceError, RunEventSink, StoreProgressLogger, append_event, append_event_if, append_event_to_sink, create_run, }; -pub use self::stored_fields::actor_from_principal; +pub use self::stored_fields::{actor_from_principal, principal_from_actor}; pub use crate::stage_scope::StageScope; diff --git a/lib/components/fabro-workflow/src/event/stored_fields.rs b/lib/components/fabro-workflow/src/event/stored_fields.rs index de4958cd3..05c6e83fd 100644 --- a/lib/components/fabro-workflow/src/event/stored_fields.rs +++ b/lib/components/fabro-workflow/src/event/stored_fields.rs @@ -331,7 +331,7 @@ fn agent_actor_for_event( /// The principal pebble's steering author stands for, where the mapping is /// lossless. A human author cannot be rebuilt from pebble's `Actor`; the /// durable `run.steer` event that delivered the steer carries the principal. -pub(crate) fn principal_from_actor(actor: &Actor) -> Option { +pub fn principal_from_actor(actor: &Actor) -> Option { match actor { Actor::Agent { id } => Some(Principal::Agent { session_id: id.clone(), diff --git a/lib/components/fabro-workflow/src/handler/llm/acp.rs b/lib/components/fabro-workflow/src/handler/llm/acp.rs index 8a1f16dbc..f1462ff94 100644 --- a/lib/components/fabro-workflow/src/handler/llm/acp.rs +++ b/lib/components/fabro-workflow/src/handler/llm/acp.rs @@ -15,12 +15,12 @@ use fabro_github::token_source::REFRESH_MARGIN; use fabro_graphviz::graph::Node; use fabro_sandbox::{RefreshOutcome, RunSandbox}; use fabro_static::EnvVars; -use fabro_types::{ - AgentBackend, Principal, SessionCapability, StageId, StageTiming, SteeringMessage, -}; +use fabro_types::{AgentBackend, SessionCapability, StageId, StageTiming}; use fabro_util::time::elapsed_ms; -use pebble_coding_agent::events::{CodingAgentEvent, CodingEvent}; +use pebble_coding_agent::events::{Actor, CodingAgentEvent, CodingEvent}; +use pebble_coding_agent::steering::SteerableSession; use pebble_coding_agent::tools::{StaticEnvProvider, ToolEnvProvider}; +use pebble_coding_agent::{SteeringMessage, SteeringOutcome}; use tokio::task::JoinHandle; use tokio::time::{sleep, timeout}; use tokio_util::sync::CancellationToken; @@ -29,11 +29,9 @@ use super::super::agent::{CodergenBackend, CodergenResult, CodergenRunRequest, O use super::activation_lease::{ActivationLease, ActivationLeaseOptions}; use super::changed_files; use crate::error::Error; -use crate::event::{ - Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope, actor_from_principal, -}; +use crate::event::{Emitter, Event, RunNoticeCode, RunNoticeLevel, StageScope}; use crate::handler::NodeTimeoutPolicy; -use crate::steering_hub::{ActiveControlHandle, SteeringHub, SteeringItem}; +use crate::steering_hub::SteeringHub; /// Default refresh-ahead interval — comfortably under the ~60-min GitHub App /// installation-token TTL. Used as the loop cadence when a tick reports no @@ -277,13 +275,12 @@ impl AgentAcpBackend { let lease_for_completion = Arc::new(Mutex::new(activation_lease)); let on_natural_completion = self.steering_hub.as_ref().map(|_| { let lease = Arc::clone(&lease_for_completion); - let control_handle = control_handle.clone(); Arc::new(move || { let mut lease = lease.lock().expect("ACP activation lease lock poisoned"); let Some(active_lease) = lease.as_ref() else { return true; }; - if active_lease.release_if_no_pending_control_work(&control_handle) { + if active_lease.release_if_idle() { lease.take(); true } else { @@ -296,7 +293,7 @@ impl AgentAcpBackend { let stage_scope = stage_scope.clone(); let node_id = node.id.clone(); let session_id = activation_session_id.clone(); - Arc::new(move |text: String, actor: Option| { + Arc::new(move |text: String, actor: Option| { emitter.emit_scoped( &Event::Agent { stage: node_id.clone(), @@ -306,14 +303,14 @@ impl AgentAcpBackend { CodingEvent::SteeringInjected { text, content: None, - actor: actor.as_ref().map(actor_from_principal), + actor, }, std::time::SystemTime::now(), ), }, &stage_scope, ); - }) as Arc) + Send + Sync> + }) as Arc) + Send + Sync> }); // Refresh before launch for early pushes. Schedule later refreshes from @@ -522,39 +519,41 @@ impl AgentAcpBackend { hub: Arc::clone(steering_hub), emitter: Arc::clone(emitter), }, - &(Arc::new(handle.clone()) as Arc), + Arc::new(AcpSteerable(handle.clone())), ) .map(Some) } } -impl ActiveControlHandle for AcpControlHandle { - fn enqueue_bounded(&self, item: SteeringItem, cap: usize) -> Option { - let item = match item { - SteeringItem::Steering { text, actor } => SteeringMessage::new(text, actor), - item => return Some(item), - }; - Self::enqueue_bounded(self, item, cap).map(SteeringItem::from) +/// How many steers wait on an ACP session before the oldest is dropped. +/// Pebble's own sessions bound their queue themselves; the ACP session's +/// queue is fabro's, so the bound is stated here. +const ACP_STEERING_QUEUE_CAP: usize = 32; + +/// The ACP session as a session on the steering bus. It cannot hold its +/// completion open, so a human cannot pair with it. +struct AcpSteerable(AcpControlHandle); + +impl SteerableSession for AcpSteerable { + fn steer(&self, message: SteeringMessage) -> SteeringOutcome { + self.0 + .enqueue_bounded(message, ACP_STEERING_QUEUE_CAP) + .map_or(SteeringOutcome::Accepted, SteeringOutcome::Evicted) } - fn interrupt(&self, actor: Option) { - Self::interrupt(self, actor); + fn interrupt(&self) -> bool { + self.0.interrupt(); + true } - fn interrupt_then_enqueue_bounded( - &self, - item: SteeringItem, - cap: usize, - ) -> Option { - let item = match item { - SteeringItem::Steering { text, actor } => SteeringMessage::new(text, actor), - item => return Some(item), - }; - Self::interrupt_then_enqueue_bounded(self, item, cap).map(SteeringItem::from) + fn steer_now(&self, message: SteeringMessage) -> SteeringOutcome { + self.0 + .interrupt_then_enqueue_bounded(message, ACP_STEERING_QUEUE_CAP) + .map_or(SteeringOutcome::Accepted, SteeringOutcome::Evicted) } - fn has_pending_control_work(&self) -> bool { - Self::has_pending_control_work(self) + fn has_pending_steering(&self) -> bool { + self.0.has_pending_control_work() } } diff --git a/lib/components/fabro-workflow/src/handler/llm/activation_lease.rs b/lib/components/fabro-workflow/src/handler/llm/activation_lease.rs index 7c13c4727..8cf36209f 100644 --- a/lib/components/fabro-workflow/src/handler/llm/activation_lease.rs +++ b/lib/components/fabro-workflow/src/handler/llm/activation_lease.rs @@ -1,12 +1,20 @@ +//! A stage's session on the steering bus, with fabro's lifecycle events. +//! +//! Activating attaches the session at its stage, records +//! `agent.session.activated` with the route and capabilities the run should +//! show, and then drains steers that waited for it. Releasing detaches and +//! records `agent.session.deactivated` once, however many times it is asked. + use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; use fabro_types::{PermissionLevel, SessionCapability, StageId}; use lithos_llm::types::{ReasoningEffort, Speed}; +use pebble_coding_agent::steering::SteerableSession; use crate::error::Error; use crate::event::{Emitter, Event}; -use crate::steering_hub::{ActiveControlHandle, SteeringHub}; +use crate::steering_hub::SteeringHub; pub struct ActivationLease { stage_id: StageId, @@ -33,18 +41,17 @@ pub struct ActivationLeaseOptions { impl ActivationLease { pub fn activate( options: ActivationLeaseOptions, - handle: &Arc, + session: Arc, ) -> Result, Error> { - let attached = - options - .hub - .attach_handle(&options.stage_id, &options.session_id, Arc::clone(handle)); - if !attached { - return Err(Error::Precondition(format!( - "stage {} already has a different active agent session", - options.stage_id - ))); - } + options + .hub + .attach(&options.stage_id, &options.session_id, session) + .map_err(|_| { + Error::Precondition(format!( + "stage {} already has a different active agent session", + options.stage_id + )) + })?; options.emitter.emit(&Event::AgentSessionActivated { node_id: options.stage_id.node_id().to_string(), @@ -58,9 +65,7 @@ impl ActivationLease { permission_level: options.permission_level, capabilities: options.capabilities, }); - options - .hub - .drain_pending_into(&options.stage_id, handle.as_ref()); + options.hub.drain_pending_into(&options.stage_id); Ok(Arc::new(Self { stage_id: options.stage_id, @@ -78,14 +83,13 @@ impl ActivationLease { self.hub.detach(&self.stage_id, &self.session_id); } - pub fn release_if_no_pending_control_work(&self, handle: &dyn ActiveControlHandle) -> bool { + /// The close-the-door check: release only if the session has no steering + /// waiting. Returns whether the lease is released. + pub fn release_if_idle(&self) -> bool { if self.released.load(Ordering::Acquire) { return true; } - if !self - .hub - .detach_if_no_pending_control_work(&self.stage_id, &self.session_id, handle) - { + if !self.hub.detach_if_idle(&self.stage_id, &self.session_id) { return false; } self.mark_released(); @@ -126,43 +130,37 @@ impl Drop for ActivationLease { mod tests { use std::sync::{Arc, Mutex}; - use fabro_types::{Principal, RunId}; + use fabro_types::RunId; + use pebble_coding_agent::{SteeringMessage, SteeringOutcome}; use super::*; - use crate::steering_hub::SteeringItem; - #[derive(Clone, Default)] + #[derive(Default)] struct SessionControlHandle { - queue: Arc>>, + queue: Mutex>, } impl SessionControlHandle { - fn new() -> Self { - Self::default() - } - fn queue_len(&self) -> usize { self.queue.lock().unwrap().len() } } - impl ActiveControlHandle for SessionControlHandle { - fn enqueue_bounded(&self, item: SteeringItem, _cap: usize) -> Option { - self.queue.lock().unwrap().push(item); - None + impl SteerableSession for SessionControlHandle { + fn steer(&self, message: SteeringMessage) -> SteeringOutcome { + self.queue.lock().unwrap().push(message); + SteeringOutcome::Accepted } - fn interrupt(&self, _actor: Option) {} - - fn interrupt_then_enqueue_bounded( - &self, - item: SteeringItem, - cap: usize, - ) -> Option { - self.enqueue_bounded(item, cap) + fn interrupt(&self) -> bool { + false } - fn has_pending_control_work(&self) -> bool { + fn steer_now(&self, message: SteeringMessage) -> SteeringOutcome { + self.steer(message) + } + + fn has_pending_steering(&self) -> bool { !self.queue.lock().unwrap().is_empty() } } @@ -200,8 +198,8 @@ mod tests { } } - fn control_handle(handle: &SessionControlHandle) -> Arc { - Arc::new(handle.clone()) + fn session(handle: &Arc) -> Arc { + Arc::clone(handle) as Arc } #[test] @@ -210,7 +208,7 @@ mod tests { let names = collect_event_names(&emitter); let hub = Arc::new(SteeringHub::new(Arc::clone(&emitter))); let stage_id = StageId::new("agent", 1); - let handle = SessionControlHandle::new(); + let handle = Arc::new(SessionControlHandle::default()); hub.deliver_steer("queued".to_string(), None); let _lease = ActivationLease::activate( @@ -220,7 +218,7 @@ mod tests { Arc::clone(&hub), Arc::clone(&emitter), ), - &control_handle(&handle), + session(&handle), ) .unwrap(); @@ -238,8 +236,8 @@ mod tests { let names = collect_event_names(&emitter); let hub = Arc::new(SteeringHub::new(Arc::clone(&emitter))); let stage_id = StageId::new("agent", 1); - let handle_a = SessionControlHandle::new(); - let handle_b = SessionControlHandle::new(); + let handle_a = Arc::new(SessionControlHandle::default()); + let handle_b = Arc::new(SessionControlHandle::default()); let _lease = ActivationLease::activate( options( @@ -248,7 +246,7 @@ mod tests { Arc::clone(&hub), Arc::clone(&emitter), ), - &control_handle(&handle_a), + session(&handle_a), ) .unwrap(); let result = ActivationLease::activate( @@ -258,7 +256,7 @@ mod tests { Arc::clone(&hub), Arc::clone(&emitter), ), - &control_handle(&handle_b), + session(&handle_b), ); assert!(result.is_err()); @@ -275,12 +273,12 @@ mod tests { } #[test] - fn release_is_idempotent() { + fn release_is_idempotent_and_release_if_idle_waits_for_steering() { let emitter = Arc::new(Emitter::new(RunId::new())); let names = collect_event_names(&emitter); let hub = Arc::new(SteeringHub::new(Arc::clone(&emitter))); let stage_id = StageId::new("agent", 1); - let handle = SessionControlHandle::new(); + let handle = Arc::new(SessionControlHandle::default()); let lease = ActivationLease::activate( options( @@ -289,10 +287,17 @@ mod tests { Arc::clone(&hub), Arc::clone(&emitter), ), - &control_handle(&handle), + session(&handle), ) .unwrap(); - lease.release(); + hub.deliver_steer("late".to_string(), None); + assert!( + !lease.release_if_idle(), + "a waiting steer keeps the door open" + ); + handle.queue.lock().unwrap().clear(); + assert!(lease.release_if_idle()); + assert!(lease.release_if_idle(), "released stays released"); lease.release(); assert_eq!( diff --git a/lib/components/fabro-workflow/src/handler/llm/pebble.rs b/lib/components/fabro-workflow/src/handler/llm/pebble.rs index edd7d13f3..15a6d7007 100644 --- a/lib/components/fabro-workflow/src/handler/llm/pebble.rs +++ b/lib/components/fabro-workflow/src/handler/llm/pebble.rs @@ -24,25 +24,24 @@ use fabro_mcp::pebble::pebble_servers; use fabro_sandbox::{RunSandbox, SecretRedactor}; use fabro_types::settings::run::RunModelControls; use fabro_types::{ - AgentMcpToolSummary, AgentProfileKind, ModelRef, PermissionLevel, Principal, SessionCapability, - StageId, StageTiming, UsdMicros, billing, + AgentMcpToolSummary, AgentProfileKind, ModelRef, PermissionLevel, SessionCapability, StageId, + StageTiming, UsdMicros, billing, }; use fabro_util::home::Home; use lithos_llm::catalog::{ModelId, ProviderId}; use lithos_llm::types::{Message as LlmMessage, Role, TokenCounts}; use pebble_agent::ToolMiddleware; use pebble_coding_agent::environment::Environment; -use pebble_coding_agent::events::{ - Actor, CodingAgentEvent, CodingEvent, EventSink, EventSinkError, -}; +use pebble_coding_agent::events::{CodingAgentEvent, CodingEvent, EventSink, EventSinkError}; use pebble_coding_agent::extensions::HumanInputProvider; use pebble_coding_agent::state::Message; +use pebble_coding_agent::steering::SteerableSession; use pebble_coding_agent::subagents::SubagentOptions; use pebble_coding_agent::tools::{RegisteredTool, ToolEnvProvider}; use pebble_coding_agent::{ CodingAgent, CodingAgentBuilder, CodingAgentControlHandle, CodingAgentExport, CodingAgentOptions, CodingInput, InterruptReason, MemoryDiscovery, ShutdownReason, - SkillDiscovery, SteeringLease, SteeringMessage, SteeringOutcome, + SkillDiscovery, }; use tokio_util::sync::CancellationToken; @@ -61,11 +60,11 @@ use super::routing::{self, ProviderContext}; use crate::context::WorkflowContext; use crate::context::keys::Fidelity; use crate::error::Error; -use crate::event::{Emitter, Event, StageScope, actor_from_principal}; +use crate::event::{Emitter, Event, StageScope}; use crate::model_fallback::{ModelFallbackNotice, ModelFallbackPolicy}; use crate::outcome::billed_model_usage_from_llm; use crate::services::FabroRunToolServices; -use crate::steering_hub::{ActiveControlHandle, SteeringHub, SteeringItem}; +use crate::steering_hub::SteeringHub; use crate::web_search::{self, SearchSecrets}; /// The share of the model's context window at which an agent stage compacts @@ -253,106 +252,6 @@ impl EventSink for WorkflowEventSink { } } -// --- Steering ------------------------------------------------------------- - -/// The steering hub's view of a live pebble agent. -struct PebbleControlHandle { - control: CodingAgentControlHandle, - /// Held while a human is paired, so a plain answer parks instead of - /// ending the stage under them. - pair_lease: Mutex>, -} - -impl PebbleControlHandle { - fn new(control: CodingAgentControlHandle) -> Self { - Self { - control, - pair_lease: Mutex::new(None), - } - } - - fn message(item: &SteeringItem) -> SteeringMessage { - match item { - SteeringItem::Steering { text, actor } => { - let message = SteeringMessage::new(text.clone()); - match actor { - Some(actor) => message.with_actor(actor_from_principal(actor)), - None => message, - } - } - SteeringItem::User { text } => { - SteeringMessage::new(text.clone()).with_actor(Actor::User { - id: None, - display_name: None, - }) - } - SteeringItem::System { text } => { - SteeringMessage::new(text.clone()).with_actor(Actor::System) - } - } - } - - /// The item the agent will never see, if the queue rejected or evicted - /// one. - fn rejected(item: SteeringItem, outcome: SteeringOutcome) -> Option { - match outcome { - SteeringOutcome::Accepted => None, - SteeringOutcome::Evicted(evicted) => Some(SteeringItem::Steering { - text: evicted.text().to_string(), - actor: None, - }), - // The agent is closed, or reported something this build does not - // know; either way the message was not queued. - SteeringOutcome::Closed | _ => Some(item), - } - } -} - -impl ActiveControlHandle for PebbleControlHandle { - /// Pebble bounds its own queue; `cap` is the hub's expectation of that - /// bound and is not applied twice. - fn enqueue_bounded(&self, item: SteeringItem, _cap: usize) -> Option { - let outcome = self.control.queue_steering(Self::message(&item)); - Self::rejected(item, outcome) - } - - fn interrupt(&self, _actor: Option) { - self.control.interrupt(); - } - - fn interrupt_then_enqueue_bounded( - &self, - item: SteeringItem, - _cap: usize, - ) -> Option { - let outcome = self.control.steer_now(Self::message(&item)); - Self::rejected(item, outcome) - } - - fn supports_pairing(&self) -> bool { - true - } - - fn pair_started(&self) { - let lease = self.control.hold_open_for_steering(); - *self - .pair_lease - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(lease); - } - - fn pair_ended(&self) { - self.pair_lease - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .take(); - } - - fn has_pending_control_work(&self) -> bool { - self.control.snapshot().pending_steering() > 0 - } -} - // --- Live invocation ------------------------------------------------------ /// One stage invocation's live agent and its accounting. @@ -362,7 +261,7 @@ impl ActiveControlHandle for PebbleControlHandle { /// are summed here, across whatever routes pebble moved through. struct LiveAgent { agent: CodingAgent, - handle: Arc, + handle: CodingAgentControlHandle, lease: Option>, total_usage: TokenCounts, total_cost: Option, @@ -375,7 +274,7 @@ struct LiveAgent { } impl LiveAgent { - fn new(agent: CodingAgent, handle: Arc) -> Self { + fn new(agent: CodingAgent, handle: CodingAgentControlHandle) -> Self { Self { agent, handle, @@ -737,8 +636,7 @@ impl PebbleBackend { thread_id: Option<&str>, bindings: &StageBindings<'_>, ) -> Result<(), Error> { - let handle: Arc = - Arc::clone(&live.handle) as Arc; + let session: Arc = Arc::new(live.handle.clone()); let lease = ActivationLease::activate( ActivationLeaseOptions { stage_id: stage_id.clone(), @@ -753,7 +651,7 @@ impl PebbleBackend { hub: Arc::clone(&self.steering_hub), emitter: Arc::clone(bindings.emitter), }, - &handle, + session, )?; live.lease = Some(lease); bindings.emitter.emit(&Event::AgentToolsAvailable { @@ -814,12 +712,12 @@ impl PebbleBackend { let released = live .lease .as_ref() - .is_none_or(|lease| lease.release_if_no_pending_control_work(live.handle.as_ref())); + .is_none_or(|lease| lease.release_if_idle()); if released { live.lease.take(); return Ok(response); } - let (steering, follow_ups) = live.handle.control.take_pending_input().into_parts(); + let (steering, follow_ups) = live.handle.take_pending_input().into_parts(); for message in steering.into_iter().chain(follow_ups) { response = self .prompt_live( @@ -1142,7 +1040,7 @@ impl CodergenBackend for PebbleBackend { "Agent session ready" ); - let handle = Arc::new(PebbleControlHandle::new(agent.control_handle())); + let handle = agent.control_handle(); let mut live = LiveAgent::new(agent, handle); let route = fallback_plan.current().clone(); if let Err(error) = diff --git a/lib/components/fabro-workflow/src/steering_hub.rs b/lib/components/fabro-workflow/src/steering_hub.rs index a9fb70290..df578dfd1 100644 --- a/lib/components/fabro-workflow/src/steering_hub.rs +++ b/lib/components/fabro-workflow/src/steering_hub.rs @@ -1,121 +1,38 @@ -//! Bridge between the worker's HTTP control plane and live agent -//! `Session`s. The hub owns: +//! Fabro's control plane over pebble's steering bus. //! -//! - A map of currently steerable live sessions, keyed by `StageId` → active -//! `(session_id, ActiveControlHandle)` entries. -//! - A bounded run-wide pending buffer for steers that arrive when no session -//! is registered (between stages, before the first agent stage, or after a -//! session ends but before the next registers). +//! The bus carries steers and interrupts to every live agent session, buffers +//! steers that arrive between sessions, and holds a session open while a +//! human is paired with it. What fabro adds is attribution: which run and +//! stage a session belongs to, who asked (a [`Principal`]), the pair record +//! the API serves, and the run events (`run.steer`, `run.interrupt`, +//! `agent.steer.buffered`, `agent.steer.dropped`, `agent.interrupt.injected`, +//! the pair events) that put bus activity on the run's durable stream in the +//! order fabro's consumers expect. //! -//! Lock discipline (race safety): -//! - `active` is `std::sync::RwLock`; deliver takes the read lock for the -//! entire decide-and-push step. -//! - `pending` is `std::sync::Mutex` taken under the active read lock. -//! - All methods are sync — no `.await` while holding any lock — so the -//! `CompletionCoordinator::on_natural_completion` close-the-door dance can -//! call `detach_if_no_pending_control_work(...)` synchronously from the -//! agent loop. +//! Every method is synchronous and never awaits under a lock, so the agent +//! loop's close-the-door check runs from its completion path. -use std::collections::{HashMap, VecDeque}; -use std::sync::{Arc, Mutex, RwLock}; +use std::sync::{Arc, Mutex, PoisonError}; use chrono::Utc; use fabro_types::run_event::AgentSteerDroppedReason; use fabro_types::{ PairId, PairMessageId, PairMessageRecord, PairRecord, PairStatus, PairSystemMessageKind, - PairTarget, Principal, RunId, RunPairEndedReason, StageId, SteeringMessage, + PairTarget, Principal, RunId, RunPairEndedReason, StageId, }; +use pebble_coding_agent::events::Actor; +use pebble_coding_agent::steering::{ + AttachError, Attachment, DropReason, DroppedSteer, SteerableSession, SteeringBus, TargetError, +}; +use pebble_coding_agent::{SteeringMessage, SteeringOutcome}; -use crate::event::{Emitter, Event}; - -/// One message the control plane hands a live session. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum SteeringItem { - /// Guidance from a steer: a user-role message that stays visibly - /// distinct from a paired user's message. - Steering { - text: String, - actor: Option, - }, - /// A paired human's own message. - User { text: String }, - /// A system notice, such as a human joining or leaving a pair. - System { text: String }, -} - -impl SteeringItem { - #[must_use] - pub fn actor(&self) -> Option<&Principal> { - match self { - Self::Steering { actor, .. } => actor.as_ref(), - Self::User { .. } | Self::System { .. } => None, - } - } - - #[must_use] - pub fn text(&self) -> &str { - match self { - Self::Steering { text, .. } | Self::User { text } | Self::System { text } => text, - } - } -} - -impl From for SteeringItem { - fn from(message: SteeringMessage) -> Self { - Self::Steering { - text: message.text, - actor: message.actor, - } - } -} - -/// Cap on the steering queue length kept per active session. Overflow -/// evicts the oldest entry (FIFO) and emits `agent.steer.dropped`. -pub const PER_SESSION_QUEUE_CAP: usize = 32; - -/// Cap on the run-wide pending buffer used when no session is registered. -/// Overflow evicts the oldest entry (FIFO) and emits `agent.steer.dropped`. -pub const PER_RUN_PENDING_CAP: usize = 32; - -pub trait ActiveControlHandle: Send + Sync { - /// Queue `item`, evicting and returning the oldest queued item when the - /// queue is at `cap`. - fn enqueue_bounded(&self, item: SteeringItem, cap: usize) -> Option; - fn interrupt(&self, actor: Option); - fn interrupt_then_enqueue_bounded( - &self, - item: SteeringItem, - cap: usize, - ) -> Option; - /// Queue `item` only when the queue is below `cap`, keeping every queued - /// item. Returns whether it was accepted. - fn try_enqueue_bounded(&self, item: SteeringItem, cap: usize) -> bool { - self.enqueue_bounded(item, cap).is_none() - } - /// Whether a human can pair with this session. - fn supports_pairing(&self) -> bool { - false - } - /// A pair started on this session: natural completion must wait for the - /// human until [`pair_ended`](Self::pair_ended). - fn pair_started(&self) {} - fn pair_ended(&self) {} - fn has_pending_control_work(&self) -> bool; -} - -#[derive(Clone)] -struct ActiveEntry { - handle: Arc, - session_id: String, -} +use crate::event::{Emitter, Event, actor_from_principal, principal_from_actor}; #[derive(Debug, Clone)] struct ActivePair { record: PairRecord, - /// Snapshot of the agent session id active at `start_pair` time. Used to - /// detect session replacement on subsequent pair commands and on - /// `AgentSessionDeactivated` cleanup; intentionally not exposed in the - /// public `PairRecord`. + /// The agent session active at `start_pair` time, so a later pair command + /// or a session's deactivation can tell whether the session was replaced. session_id: String, } @@ -133,9 +50,8 @@ pub enum PairControlError { reason = "external callers refer to it as SteeringHub" )] pub struct SteeringHub { - active: RwLock>, + bus: SteeringBus, active_pair: Mutex>, - pending: Mutex>, emitter: Arc, } @@ -143,9 +59,8 @@ impl SteeringHub { #[must_use] pub fn new(emitter: Arc) -> Self { Self { - active: RwLock::new(HashMap::new()), + bus: SteeringBus::new(), active_pair: Mutex::new(None), - pending: Mutex::new(VecDeque::new()), emitter, } } @@ -154,185 +69,92 @@ impl SteeringHub { #[cfg(test)] #[must_use] pub fn for_tests() -> Arc { - use fabro_types::RunId; Arc::new(Self::new(Arc::new(Emitter::new(RunId::new())))) } - /// Test-only: snapshot of pending buffer length. + /// Test-only: how many steers wait for the next session. #[cfg(test)] #[must_use] pub fn pending_len(&self) -> usize { - self.pending.lock().expect("pending lock poisoned").len() + self.bus.pending_len() } - /// Test-only: snapshot of registered stage count. + /// Test-only: how many sessions are attached. #[cfg(test)] #[must_use] pub fn active_count(&self) -> usize { - self.active.read().expect("active lock poisoned").len() + self.bus.attached_count() } - /// Attach a live backend session as steerable for this stage. Returns - /// `false` when a different session is already active for the stage. - pub fn attach_handle( + /// Attach a live session as steerable for this stage. Fails when a + /// different session is already active for the stage. + pub(crate) fn attach( &self, stage_id: &StageId, session_id: &str, - handle: Arc, - ) -> bool { - let mut active = self.active.write().expect("active lock poisoned"); - match active.get_mut(stage_id) { - Some(entry) if entry.session_id != session_id => false, - Some(entry) => { - entry.handle = handle; - true - } - None => { - active.insert(stage_id.clone(), ActiveEntry { - handle, - session_id: session_id.to_string(), - }); - true - } - } + session: Arc, + ) -> Result<(), AttachError> { + self.bus.attach(stage_id.clone(), session_id, session) } - /// Drain pending run-wide steers into `handle`. - pub fn drain_pending_into(&self, stage_id: &StageId, handle: &dyn ActiveControlHandle) { - let pending: Vec = { - let mut pending = self.pending.lock().expect("pending lock poisoned"); - pending.drain(..).collect() - }; - for item in pending { - Self::enqueue_into_session_queue(handle, item, &self.emitter, Some(stage_id)); - } + /// Move buffered run-wide steers into the stage's session. + pub(crate) fn drain_pending_into(&self, stage_id: &StageId) { + let delivery = self.bus.drain_pending_into(stage_id); + self.emit_dropped(&delivery.dropped); } /// Detach the session for this stage. Stale session ids are ignored. - pub fn detach(&self, stage_id: &StageId, session_id: &str) -> bool { - let mut active = self.active.write().expect("active lock poisoned"); - let Some(entry) = active.get(stage_id) else { - return false; - }; - if entry.session_id != session_id { + pub(crate) fn detach(&self, stage_id: &StageId, session_id: &str) -> bool { + if !self.bus.detach(stage_id, session_id) { return false; } - active.remove(stage_id); - drop(active); self.end_active_pair_for_target(stage_id, session_id, RunPairEndedReason::SessionEnded); true } - /// Atomic close-the-door check used by the agent loop's natural- - /// completion path. Under the `active` write lock: if `handle`'s queue - /// is empty and the active session id matches, remove the stage and - /// return `true`. If the queue is non-empty, leave the registration - /// intact and return `false`. - pub fn detach_if_no_pending_control_work( - &self, - stage_id: &StageId, - session_id: &str, - handle: &dyn ActiveControlHandle, - ) -> bool { - let mut active = self.active.write().expect("active lock poisoned"); - let Some(entry) = active.get(stage_id) else { - return false; - }; - if entry.session_id != session_id || handle.has_pending_control_work() { + /// The agent loop's close-the-door check: detach only when the session + /// has no steering waiting, atomically against a steer arriving. + pub(crate) fn detach_if_idle(&self, stage_id: &StageId, session_id: &str) -> bool { + if !self.bus.detach_if_idle(stage_id, session_id) { return false; } - active.remove(stage_id); - drop(active); self.end_active_pair_for_target(stage_id, session_id, RunPairEndedReason::SessionEnded); true } - /// Deliver a steer from the HTTP control plane. Broadcasts to every - /// active session if any are registered, otherwise parks the message - /// in the run-wide pending buffer. + /// Deliver a steer from the control plane: to every active session, or + /// into the run-wide buffer when none is active. pub fn deliver_steer(&self, text: String, actor: Option) { self.emitter.emit(&Event::RunSteer { text: text.clone(), actor: actor.clone(), }); - - // Hold the active read lock for the entire decide-and-dispatch - // step so register/unregister cannot race with this push. - let active = self.active.read().expect("active lock poisoned"); - if active.is_empty() { - let dropped_actor = { - let mut pending = self.pending.lock().expect("pending lock poisoned"); - let dropped_actor = if pending.len() >= PER_RUN_PENDING_CAP { - pending.pop_front().and_then(|d| d.actor().cloned()) - } else { - None - }; - pending.push_back(SteeringItem::Steering { - text, - actor: actor.clone(), - }); - dropped_actor - }; - - if let Some(dropped_actor) = dropped_actor { - self.emitter.emit(&Event::AgentSteerDropped { - reason: AgentSteerDroppedReason::QueueFull, - count: 1, - actor: Some(dropped_actor), - node_id: None, - visit: None, - }); - } + let delivery = self.bus.steer(steering_message(text, actor.as_ref())); + self.emit_dropped(&delivery.dropped); + if delivery.buffered { self.emitter.emit(&Event::AgentSteerBuffered { actor }); - drop(active); - return; - } - - // Broadcast to every active session. - for (stage_id, entry) in active.iter() { - Self::enqueue_into_session_queue( - entry.handle.as_ref(), - SteeringItem::Steering { - text: text.clone(), - actor: actor.clone(), - }, - &self.emitter, - Some(stage_id), - ); } } - /// Interrupt every active steerable session. Does not buffer when no - /// active session exists. + /// Interrupt every active session. Not buffered: with no session active + /// there is nothing to stop. pub fn interrupt(&self, actor: Option<&Principal>) { - let active = self.active.read().expect("active lock poisoned"); - if active.is_empty() { + if self.bus.attached_count() == 0 { return; } - self.emitter.emit(&Event::RunInterrupt { actor: actor.cloned(), }); - for (stage_id, entry) in active.iter() { - entry.handle.interrupt(actor.cloned()); - self.emitter.emit(&Event::AgentInterruptInjected { - node_id: stage_id.node_id().to_string(), - visit: stage_id.visit(), - session_id: entry.session_id.clone(), - actor: actor.cloned(), - }); - } + let interruption = self.bus.interrupt(); + self.emit_interrupted(&interruption.interrupted, actor); } - /// Atomically apply interrupt semantics, then deliver steering text to - /// every active steerable session. Emits persisted run events in the same - /// order. + /// Interrupt every active session and hand each the steering text as + /// what replaces its round, emitting the run events in that order. pub fn interrupt_then_steer(&self, text: &str, actor: Option<&Principal>) { - let active = self.active.read().expect("active lock poisoned"); - if active.is_empty() { + if self.bus.attached_count() == 0 { return; } - self.emitter.emit(&Event::RunInterrupt { actor: actor.cloned(), }); @@ -340,50 +162,24 @@ impl SteeringHub { text: text.to_string(), actor: actor.cloned(), }); - - for (stage_id, entry) in active.iter() { - if let Some(evicted) = entry.handle.interrupt_then_enqueue_bounded( - SteeringItem::Steering { - text: text.to_string(), - actor: actor.cloned(), - }, - PER_SESSION_QUEUE_CAP, - ) { - self.emitter.emit(&Event::AgentSteerDropped { - reason: AgentSteerDroppedReason::QueueFull, - count: 1, - actor: evicted.actor().cloned(), - node_id: Some(stage_id.node_id().to_string()), - visit: Some(stage_id.visit()), - }); - } - self.emitter.emit(&Event::AgentInterruptInjected { - node_id: stage_id.node_id().to_string(), - visit: stage_id.visit(), - session_id: entry.session_id.clone(), - actor: actor.cloned(), - }); - } + let interruption = self + .bus + .interrupt_then_steer(&steering_message(text.to_string(), actor)); + self.emit_dropped(&interruption.dropped); + self.emit_interrupted(&interruption.interrupted, actor); } - /// Drain any unconsumed pending steers and emit a single - /// `agent.steer.dropped` event with `reason: run_ended`. Called from - /// `operations::start` after the pipeline finishes (success or - /// failure) but before the emitter is flushed. + /// Drop any steer nobody read and say so once, with `reason: run_ended`. + /// Called from `operations::start` after the pipeline finishes but + /// before the emitter is flushed. pub fn drain_pending_at_run_end(&self) { - let count: u32 = { - let mut pending = self.pending.lock().expect("pending lock poisoned"); - let n = u32::try_from(pending.len()).unwrap_or(u32::MAX); - pending.clear(); - n - }; - if count > 0 { + if let Some(dropped) = self.bus.drain_pending() { self.emitter.emit(&Event::AgentSteerDropped { - reason: AgentSteerDroppedReason::RunEnded, - count, - actor: None, + reason: AgentSteerDroppedReason::RunEnded, + count: u32::try_from(dropped.count).unwrap_or(u32::MAX), + actor: None, node_id: None, - visit: None, + visit: None, }); } self.end_active_pair(RunPairEndedReason::RunEnded); @@ -396,31 +192,36 @@ impl SteeringHub { target: PairTarget, actor: Option, ) -> Result { - let active = self.active.read().expect("active lock poisoned"); - let Some(entry) = active.get(&target.stage_id) else { - return Err(PairControlError::TargetNotActive); - }; - if !entry.handle.supports_pairing() { - return Err(PairControlError::TargetNotActive); - } - let session_id = entry.session_id.clone(); - let interrupt_handle = Arc::clone(&entry.handle); - let pair_handle = Arc::clone(&entry.handle); - drop(active); + let session_id = self + .bus + .attachments() + .into_iter() + .find(|attachment| attachment.key == target.stage_id) + .map(|attachment| attachment.session_id) + .ok_or(PairControlError::TargetNotActive)?; - let mut active_pair = self.active_pair.lock().expect("active pair lock poisoned"); + let mut active_pair = self + .active_pair + .lock() + .unwrap_or_else(PoisonError::into_inner); if active_pair.is_some() { return Err(PairControlError::AlreadyPaired); } + // The hold comes first: a session that cannot be held open cannot be + // paired with, and nothing is queued on it. + match self.bus.hold_open(&target.stage_id, &session_id) { + Ok(()) => {} + Err(TargetError::AlreadyHeld) => return Err(PairControlError::AlreadyPaired), + Err(TargetError::NotAttached | TargetError::Unsupported) => { + return Err(PairControlError::TargetNotActive); + } + } let text = human_joined_text(); - if !pair_handle.try_enqueue_bounded( - SteeringItem::System { - text: text.to_string(), - }, - PER_SESSION_QUEUE_CAP, - ) { - return Err(PairControlError::MessageNotAccepted); + let notice = SteeringMessage::new(text).with_actor(Actor::System); + if let Err(error) = self.send_to_paired(&target.stage_id, &session_id, notice) { + self.bus.release_hold(&target.stage_id, &session_id); + return Err(error); } let record = PairRecord { @@ -435,11 +236,11 @@ impl SteeringHub { self.emitter.emit(&Event::RunPairStarted { pair_id, target: record.target.clone(), - actor: actor.clone(), + actor, }); - - pair_handle.pair_started(); - interrupt_handle.interrupt(actor); + // With the notice already queued the session does not park: the + // notice opens its next round. + let _ = self.bus.interrupt_at(&record.target.stage_id, &session_id); self.emitter.emit(&Event::AgentPairSystemMessage { node_id: record.target.stage_id.node_id().to_string(), visit: record.target.stage_id.visit(), @@ -463,34 +264,19 @@ impl SteeringHub { client_message_id: Option, actor: Option, ) -> Result { - let active_pair = self.active_pair.lock().expect("active pair lock poisoned"); - let Some(pair) = active_pair.as_ref() else { - return Err(PairControlError::PairNotActive); - }; - if pair.record.pair_id != pair_id { - return Err(PairControlError::PairNotCurrent); - } - if pair.record.status != PairStatus::Active { - return Err(PairControlError::PairNotActive); - } - + let active_pair = self + .active_pair + .lock() + .unwrap_or_else(PoisonError::into_inner); + let pair = current_pair(active_pair.as_ref(), pair_id)?; let target = &pair.record.target; let session_id = pair.session_id.clone(); - let active = self.active.read().expect("active lock poisoned"); - let Some(entry) = active.get(&target.stage_id) else { - return Err(PairControlError::TargetNotActive); - }; - if entry.session_id != session_id || !entry.handle.supports_pairing() { - return Err(PairControlError::TargetNotActive); - } - let pair_handle = &entry.handle; - if !pair_handle.try_enqueue_bounded( - SteeringItem::User { text: text.clone() }, - PER_SESSION_QUEUE_CAP, - ) { - return Err(PairControlError::MessageNotAccepted); - } + let message = SteeringMessage::new(text.clone()).with_actor(Actor::User { + id: None, + display_name: None, + }); + self.send_to_paired(&target.stage_id, &session_id, message)?; self.emitter.emit(&Event::AgentPairUserMessage { node_id: target.stage_id.node_id().to_string(), visit: target.stage_id.visit(), @@ -517,39 +303,18 @@ impl SteeringHub { pair_id: PairId, actor: Option, ) -> Result { - let mut active_pair = self.active_pair.lock().expect("active pair lock poisoned"); - let Some(pair) = active_pair.as_mut() else { - return Err(PairControlError::PairNotActive); - }; - if pair.record.pair_id != pair_id { - return Err(PairControlError::PairNotCurrent); - } - if pair.record.status != PairStatus::Active { - return Err(PairControlError::PairNotActive); - } - + let mut active_pair = self + .active_pair + .lock() + .unwrap_or_else(PoisonError::into_inner); + let pair = current_pair(active_pair.as_ref(), pair_id)?; let target = pair.record.target.clone(); let session_id = pair.session_id.clone(); - let text = human_left_text(); - if let Some(entry) = self - .active - .read() - .expect("active lock poisoned") - .get(&target.stage_id) - .filter(|entry| entry.session_id == session_id) - { - if !entry.handle.supports_pairing() { - return Err(PairControlError::TargetNotActive); - } - let pair_handle = &entry.handle; - if !pair_handle.try_enqueue_bounded( - SteeringItem::System { - text: text.to_string(), - }, - PER_SESSION_QUEUE_CAP, - ) { - return Err(PairControlError::MessageNotAccepted); - } + + if self.bus.is_attached(&target.stage_id, &session_id) { + let text = human_left_text(); + let notice = SteeringMessage::new(text).with_actor(Actor::System); + self.send_to_paired(&target.stage_id, &session_id, notice)?; self.emitter.emit(&Event::AgentPairSystemMessage { node_id: target.stage_id.node_id().to_string(), visit: target.stage_id.visit(), @@ -558,17 +323,17 @@ impl SteeringHub { kind: PairSystemMessageKind::HumanLeft, text: text.to_string(), }); - entry.handle.pair_ended(); + self.bus.release_hold(&target.stage_id, &session_id); } - pair.record.status = PairStatus::Ended; - pair.record.ended_at = Some(Utc::now()); + let mut record = pair.record.clone(); + record.status = PairStatus::Ended; + record.ended_at = Some(Utc::now()); self.emitter.emit(&Event::RunPairEnded { pair_id, - reason: fabro_types::RunPairEndedReason::UserRequested, + reason: RunPairEndedReason::UserRequested, actor, }); - let record = pair.record.clone(); *active_pair = None; Ok(record) } @@ -577,7 +342,7 @@ impl SteeringHub { pub fn pair_is_active_for(&self, stage_id: &StageId, session_id: &str) -> bool { self.active_pair .lock() - .expect("active pair lock poisoned") + .unwrap_or_else(PoisonError::into_inner) .as_ref() .is_some_and(|pair| { pair.record.status == PairStatus::Active @@ -586,6 +351,34 @@ impl SteeringHub { }) } + /// Queue a paired human's message or a pair notice on the target session. + /// A message the session queued by evicting an older steer is accepted, + /// and the eviction is recorded; a closed session accepts nothing. + fn send_to_paired( + &self, + stage_id: &StageId, + session_id: &str, + message: SteeringMessage, + ) -> Result<(), PairControlError> { + match self.bus.send_to(stage_id, session_id, message) { + Ok(SteeringOutcome::Accepted) => Ok(()), + Ok(SteeringOutcome::Evicted(evicted)) => { + self.emit_dropped(&[DroppedSteer { + reason: DropReason::QueueFull, + count: 1, + actor: evicted.actor().cloned(), + attachment: Some(Attachment { + key: stage_id.clone(), + session_id: session_id.to_string(), + }), + }]); + Ok(()) + } + Ok(_) => Err(PairControlError::MessageNotAccepted), + Err(_) => Err(PairControlError::TargetNotActive), + } + } + fn end_active_pair_for_target( &self, stage_id: &StageId, @@ -593,7 +386,10 @@ impl SteeringHub { reason: RunPairEndedReason, ) -> bool { let pair_id = { - let mut active_pair = self.active_pair.lock().expect("active pair lock poisoned"); + let mut active_pair = self + .active_pair + .lock() + .unwrap_or_else(PoisonError::into_inner); let Some(pair) = active_pair.as_ref() else { return false; }; @@ -607,15 +403,7 @@ impl SteeringHub { *active_pair = None; pair_id }; - if let Some(entry) = self - .active - .read() - .expect("active lock poisoned") - .get(stage_id) - .filter(|entry| entry.session_id == session_id) - { - entry.handle.pair_ended(); - } + // The bus released the session's hold when it detached. self.emitter.emit(&Event::RunPairEnded { pair_id, reason, @@ -626,7 +414,10 @@ impl SteeringHub { fn end_active_pair(&self, reason: RunPairEndedReason) -> bool { let pair_id = { - let mut active_pair = self.active_pair.lock().expect("active pair lock poisoned"); + let mut active_pair = self + .active_pair + .lock() + .unwrap_or_else(PoisonError::into_inner); let Some(mut pair) = active_pair.take() else { return false; }; @@ -646,25 +437,57 @@ impl SteeringHub { true } - /// Push an item into a session's queue, evicting the oldest entry and - /// emitting `agent.steer.dropped { queue_full }` if the cap is hit. - /// The push + eviction are atomic under the per-session queue lock. - fn enqueue_into_session_queue( - handle: &dyn ActiveControlHandle, - item: SteeringItem, - emitter: &Emitter, - stage_id: Option<&StageId>, - ) { - if let Some(evicted) = handle.enqueue_bounded(item, PER_SESSION_QUEUE_CAP) { - emitter.emit(&Event::AgentSteerDropped { - reason: AgentSteerDroppedReason::QueueFull, - count: 1, - actor: evicted.actor().cloned(), - node_id: stage_id.map(|s| s.node_id().to_string()), + /// One `agent.steer.dropped { queue_full }` per message a queue evicted, + /// naming the stage whose session dropped it when one did. + fn emit_dropped(&self, dropped: &[DroppedSteer]) { + for drop in dropped { + let stage_id = drop.attachment.as_ref().map(|attachment| &attachment.key); + self.emitter.emit(&Event::AgentSteerDropped { + reason: match drop.reason { + DropReason::Ended => AgentSteerDroppedReason::RunEnded, + DropReason::QueueFull | _ => AgentSteerDroppedReason::QueueFull, + }, + count: u32::try_from(drop.count).unwrap_or(u32::MAX), + actor: drop.actor.as_ref().and_then(principal_from_actor), + node_id: stage_id.map(|stage| stage.node_id().to_string()), visit: stage_id.map(StageId::visit), }); } } + + fn emit_interrupted(&self, interrupted: &[Attachment], actor: Option<&Principal>) { + for attachment in interrupted { + self.emitter.emit(&Event::AgentInterruptInjected { + node_id: attachment.key.node_id().to_string(), + visit: attachment.key.visit(), + session_id: attachment.session_id.clone(), + actor: actor.cloned(), + }); + } + } +} + +fn current_pair( + pair: Option<&ActivePair>, + pair_id: PairId, +) -> Result<&ActivePair, PairControlError> { + let pair = pair.ok_or(PairControlError::PairNotActive)?; + if pair.record.pair_id != pair_id { + return Err(PairControlError::PairNotCurrent); + } + if pair.record.status != PairStatus::Active { + return Err(PairControlError::PairNotActive); + } + Ok(pair) +} + +/// A steer as the session reads it, with fabro's principal as pebble's actor. +fn steering_message(text: String, actor: Option<&Principal>) -> SteeringMessage { + let message = SteeringMessage::new(text); + match actor { + Some(actor) => message.with_actor(actor_from_principal(actor)), + None => message, + } } pub fn human_joined_text() -> &'static str { @@ -677,25 +500,46 @@ pub fn human_left_text() -> &'static str { #[cfg(test)] mod tests { + use std::collections::VecDeque; + use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use fabro_types::{ PairId, PairMessageId, PairTarget, Principal, RunEvent, RunId, StageId, SystemActorKind, }; + use pebble_coding_agent::steering::SessionHold; - use super::{ActiveControlHandle, PairControlError, SteeringHub, SteeringItem}; + use super::*; + use crate::event::Emitter; - /// A steerable, pairable session with a bounded FIFO queue, standing in - /// for the pebble control handle. - #[derive(Clone, Default)] + /// A steerable, pairable session with a bounded queue, standing in for + /// pebble's control handle. struct SessionControlHandle { - queue: Arc>>, - interrupted: Arc>, + queue: Mutex>, + capacity: usize, + interrupted: AtomicUsize, + pairable: bool, } impl SessionControlHandle { - fn new() -> Self { - Self::default() + fn new() -> Arc { + Arc::new(Self { + queue: Mutex::new(VecDeque::new()), + capacity: 32, + interrupted: AtomicUsize::new(0), + pairable: true, + }) + } + + /// A session on a backend that cannot hold its completion open, as + /// the ACP adapter is. + fn unpairable() -> Arc { + Arc::new(Self { + queue: Mutex::new(VecDeque::new()), + capacity: 32, + interrupted: AtomicUsize::new(0), + pairable: false, + }) } fn queue_len(&self) -> usize { @@ -703,48 +547,38 @@ mod tests { } fn interrupt_count(&self) -> usize { - *self.interrupted.lock().unwrap() - } - - /// An interrupted session with nothing queued parks until a steer - /// arrives, as pebble's does. - fn is_waiting_for_steer(&self) -> bool { - self.interrupt_count() > 0 && self.queue_len() == 0 + self.interrupted.load(Ordering::SeqCst) } } - impl ActiveControlHandle for SessionControlHandle { - fn enqueue_bounded(&self, item: SteeringItem, cap: usize) -> Option { + impl SteerableSession for SessionControlHandle { + fn steer(&self, message: SteeringMessage) -> SteeringOutcome { let mut queue = self.queue.lock().unwrap(); - queue.push(item); - if queue.len() > cap { - return Some(queue.remove(0)); - } - None + let evicted = (queue.len() >= self.capacity) + .then(|| queue.pop_front()) + .flatten(); + queue.push_back(message); + evicted.map_or(SteeringOutcome::Accepted, SteeringOutcome::Evicted) } - fn interrupt(&self, _actor: Option) { - *self.interrupted.lock().unwrap() += 1; - } - - fn interrupt_then_enqueue_bounded( - &self, - item: SteeringItem, - cap: usize, - ) -> Option { - self.interrupt(None); - self.enqueue_bounded(item, cap) - } - - fn supports_pairing(&self) -> bool { + fn interrupt(&self) -> bool { + self.interrupted.fetch_add(1, Ordering::SeqCst); true } - fn has_pending_control_work(&self) -> bool { + fn steer_now(&self, message: SteeringMessage) -> SteeringOutcome { + self.interrupt(); + self.steer(message) + } + + fn has_pending_steering(&self) -> bool { !self.queue.lock().unwrap().is_empty() } + + fn hold_open(&self) -> Option { + self.pairable.then(|| SessionHold::new(())) + } } - use crate::event::Emitter; fn hub_with_event_names() -> (Arc, Arc>>) { let emitter = Arc::new(Emitter::new(RunId::new())); @@ -769,55 +603,25 @@ mod tests { (Arc::new(SteeringHub::new(emitter)), events) } - fn pair_target(stage_id: &StageId, _session_id: &str) -> PairTarget { + fn pair_target(stage_id: &StageId) -> PairTarget { PairTarget { stage_id: stage_id.clone(), node_label: stage_id.node_id().to_string(), } } - fn control_handle(handle: &SessionControlHandle) -> Arc { - Arc::new(handle.clone()) - } - - #[derive(Default)] - struct FakeAcpControlHandle { - queue: Mutex>, - interrupted: Mutex, - } - - impl FakeAcpControlHandle { - fn queue_len(&self) -> usize { - self.queue.lock().unwrap().len() - } - - fn interrupt_count(&self) -> usize { - *self.interrupted.lock().unwrap() - } - } - - impl ActiveControlHandle for FakeAcpControlHandle { - fn enqueue_bounded(&self, item: SteeringItem, _cap: usize) -> Option { - self.queue.lock().unwrap().push(item); - None - } - - fn interrupt(&self, _actor: Option) { - *self.interrupted.lock().unwrap() += 1; - } - - fn interrupt_then_enqueue_bounded( - &self, - item: SteeringItem, - cap: usize, - ) -> Option { - self.interrupt(None); - self.enqueue_bounded(item, cap) - } - - fn has_pending_control_work(&self) -> bool { - !self.queue.lock().unwrap().is_empty() - } + fn attach( + hub: &SteeringHub, + stage: &StageId, + session_id: &str, + handle: &Arc, + ) { + hub.attach( + stage, + session_id, + Arc::clone(handle) as Arc, + ) + .expect("attaches"); } #[test] @@ -837,43 +641,30 @@ mod tests { } #[test] - fn drain_pending_at_run_end_clears_buffer() { - let hub = SteeringHub::for_tests(); + fn drain_pending_at_run_end_reports_the_unread_steers_once() { + let (hub, events) = hub_with_events(); hub.deliver_steer("a".into(), None); hub.deliver_steer("b".into(), None); - assert_eq!(hub.pending_len(), 2); hub.drain_pending_at_run_end(); assert_eq!(hub.pending_len(), 0); + let events = events.lock().unwrap(); + let dropped = events + .iter() + .filter(|event| event.event_name() == "agent.steer.dropped") + .collect::>(); + assert_eq!(dropped.len(), 1); } #[test] - fn pending_buffer_evicts_oldest_at_cap() { - let hub = SteeringHub::for_tests(); - for i in 0..(super::PER_RUN_PENDING_CAP + 5) { - hub.deliver_steer(format!("msg{i}"), None); - } - assert_eq!(hub.pending_len(), super::PER_RUN_PENDING_CAP); - } - - #[test] - fn unregister_is_idempotent() { - let hub = SteeringHub::for_tests(); - let stage = StageId::new("agent-node", 1); - hub.detach(&stage, "session-a"); - hub.detach(&stage, "session-a"); - } - - #[test] - fn attach_and_drain_pending_into_first_session() { + fn attach_and_drain_pending_delivers_to_the_first_session() { let hub = SteeringHub::for_tests(); hub.deliver_steer("queued1".into(), None); hub.deliver_steer("queued2".into(), None); - assert_eq!(hub.pending_len(), 2); let stage = StageId::new("agent-node", 1); let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage, "session-a", control_handle(&handle))); - hub.drain_pending_into(&stage, &handle); + attach(&hub, &stage, "session-a", &handle); + hub.drain_pending_into(&stage); assert_eq!(handle.queue_len(), 2); assert_eq!(hub.pending_len(), 0); @@ -881,35 +672,14 @@ mod tests { } #[test] - fn deliver_broadcasts_to_active_sessions() { - let hub = SteeringHub::for_tests(); - let stage_a = StageId::new("a", 1); - let stage_b = StageId::new("b", 1); - let handle_a = SessionControlHandle::new(); - let handle_b = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage_a, "session-a", control_handle(&handle_a))); - assert!(hub.attach_handle(&stage_b, "session-b", control_handle(&handle_b))); - - hub.deliver_steer("hello".into(), None); - - assert_eq!(handle_a.queue_len(), 1); - assert_eq!(handle_b.queue_len(), 1); - assert_eq!(hub.pending_len(), 0); - } - - #[test] - fn deliver_broadcasts_to_api_and_acp_control_handles() { + fn deliver_broadcasts_to_pebble_and_acp_sessions_alike() { let hub = SteeringHub::for_tests(); let api_stage = StageId::new("api", 1); let acp_stage = StageId::new("acp", 1); let api_handle = SessionControlHandle::new(); - let acp_handle = Arc::new(FakeAcpControlHandle::default()); - assert!(hub.attach_handle(&api_stage, "session-api", control_handle(&api_handle))); - assert!(hub.attach_handle( - &acp_stage, - "session-acp", - Arc::clone(&acp_handle) as Arc, - )); + let acp_handle = SessionControlHandle::unpairable(); + attach(&hub, &api_stage, "session-api", &api_handle); + attach(&hub, &acp_stage, "session-acp", &acp_handle); hub.deliver_steer("hello".into(), None); hub.interrupt(None); @@ -917,57 +687,58 @@ mod tests { assert_eq!(api_handle.queue_len(), 1); assert_eq!(acp_handle.queue_len(), 1); assert_eq!(acp_handle.interrupt_count(), 1); + assert_eq!(hub.pending_len(), 0); } #[test] - fn attach_rejects_different_session_for_same_stage() { - let hub = SteeringHub::for_tests(); + fn a_steer_a_session_evicted_is_recorded_against_its_stage() { + let (hub, events) = hub_with_events(); let stage = StageId::new("a", 1); - let handle1 = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage, "session-a", control_handle(&handle1))); - hub.deliver_steer("x".into(), None); - assert_eq!(handle1.queue_len(), 1); + let handle = Arc::new(SessionControlHandle { + queue: Mutex::new(VecDeque::new()), + capacity: 1, + interrupted: AtomicUsize::new(0), + pairable: true, + }); + attach(&hub, &stage, "session-a", &handle); - let handle2 = SessionControlHandle::new(); - assert!(!hub.attach_handle(&stage, "session-b", control_handle(&handle2))); - assert_eq!(handle2.queue_len(), 0); - } - - #[test] - fn stale_detach_does_not_remove_active_session() { - let hub = SteeringHub::for_tests(); - let stage = StageId::new("a", 1); - let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage, "session-a", control_handle(&handle))); - - assert!(!hub.detach(&stage, "session-b")); - hub.deliver_steer("still-active".into(), None); + hub.deliver_steer( + "first".into(), + Some(Principal::System { + system_kind: SystemActorKind::Engine, + }), + ); + hub.deliver_steer("second".into(), None); assert_eq!(handle.queue_len(), 1); - assert_eq!(hub.active_count(), 1); + let events = events.lock().unwrap(); + let dropped = events + .iter() + .find(|event| event.event_name() == "agent.steer.dropped") + .expect("the eviction is recorded"); + assert_eq!(dropped.node_id.as_deref(), Some("a")); + assert_eq!( + dropped.actor, + Some(Principal::System { + system_kind: SystemActorKind::Engine, + }), + "a system author survives the round trip through pebble's actor" + ); } #[test] - fn detach_if_no_pending_control_work_respects_session_id_and_queue_state() { + fn detach_if_idle_respects_session_id_and_queue_state() { let hub = SteeringHub::for_tests(); let stage = StageId::new("a", 1); let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage, "session-a", control_handle(&handle))); + attach(&hub, &stage, "session-a", &handle); - assert!(!hub.detach_if_no_pending_control_work(&stage, "session-b", &handle)); + assert!(!hub.detach_if_idle(&stage, "session-b")); hub.deliver_steer("queued".into(), None); - assert!(!hub.detach_if_no_pending_control_work(&stage, "session-a", &handle)); + assert!(!hub.detach_if_idle(&stage, "session-a")); assert_eq!(hub.active_count(), 1); - } - - #[test] - fn detach_if_no_pending_control_work_removes_matching_empty_session() { - let hub = SteeringHub::for_tests(); - let stage = StageId::new("a", 1); - let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage, "session-a", control_handle(&handle))); - - assert!(hub.detach_if_no_pending_control_work(&stage, "session-a", &handle)); + handle.queue.lock().unwrap().clear(); + assert!(hub.detach_if_idle(&stage, "session-a")); assert_eq!(hub.active_count(), 0); } @@ -976,12 +747,12 @@ mod tests { let (hub, events) = hub_with_events(); let stage = StageId::new("a", 1); let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage, "session-a", control_handle(&handle))); + attach(&hub, &stage, "session-a", &handle); hub.interrupt(None); hub.interrupt(None); - assert!(handle.is_waiting_for_steer()); + assert_eq!(handle.interrupt_count(), 2); assert_eq!(handle.queue_len(), 0); assert_eq!(hub.pending_len(), 0); let events = events.lock().unwrap(); @@ -998,16 +769,25 @@ mod tests { assert_eq!(events[3].session_id.as_deref(), Some("session-a")); } + #[test] + fn an_interrupt_with_no_session_emits_nothing() { + let (hub, names) = hub_with_event_names(); + hub.interrupt(None); + hub.interrupt_then_steer("stop", None); + assert!(names.lock().unwrap().is_empty()); + assert_eq!(hub.pending_len(), 0, "an interrupt is not buffered"); + } + #[test] fn interrupt_then_steer_cancels_and_queues_text() { let (hub, events) = hub_with_events(); let stage = StageId::new("a", 1); let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage, "session-a", control_handle(&handle))); + attach(&hub, &stage, "session-a", &handle); hub.interrupt_then_steer("stop", None); - assert!(!handle.is_waiting_for_steer()); + assert_eq!(handle.interrupt_count(), 1); assert_eq!(handle.queue_len(), 1); assert_eq!(hub.pending_len(), 0); let events = events.lock().unwrap(); @@ -1026,19 +806,19 @@ mod tests { let (hub, events) = hub_with_events(); let stage_id = StageId::new("code", 1); let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage_id, "ses_01", control_handle(&handle))); + attach(&hub, &stage_id, "ses_01", &handle); let pair_id = PairId::new(); let started = hub - .start_pair( - RunId::new(), - pair_id, - pair_target(&stage_id, "ses_01"), - None, - ) + .start_pair(RunId::new(), pair_id, pair_target(&stage_id), None) .unwrap(); assert_eq!(started.status, fabro_types::PairStatus::Active); assert_eq!(handle.queue_len(), 1); + assert_eq!( + handle.interrupt_count(), + 1, + "the paired session alone is told" + ); assert!(hub.pair_is_active_for(&stage_id, "ses_01")); let message = hub @@ -1074,20 +854,43 @@ mod tests { } #[test] - fn pair_start_rejects_missing_target() { + fn pair_start_rejects_missing_or_unpairable_targets_and_a_second_pair() { let hub = SteeringHub::for_tests(); let stage_id = StageId::new("code", 1); + let acp_stage = StageId::new("acp", 1); let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage_id, "ses_01", control_handle(&handle))); + attach(&hub, &stage_id, "ses_01", &handle); + attach( + &hub, + &acp_stage, + "ses_acp", + &SessionControlHandle::unpairable(), + ); let missing_stage = StageId::new("other", 1); - let result = hub.start_pair( - RunId::new(), - PairId::new(), - pair_target(&missing_stage, "ses_01"), - None, + assert_eq!( + hub.start_pair( + RunId::new(), + PairId::new(), + pair_target(&missing_stage), + None + ) + .unwrap_err(), + PairControlError::TargetNotActive + ); + assert_eq!( + hub.start_pair(RunId::new(), PairId::new(), pair_target(&acp_stage), None) + .unwrap_err(), + PairControlError::TargetNotActive, + "a session that cannot be held open cannot be paired with" + ); + hub.start_pair(RunId::new(), PairId::new(), pair_target(&stage_id), None) + .unwrap(); + assert_eq!( + hub.start_pair(RunId::new(), PairId::new(), pair_target(&stage_id), None) + .unwrap_err(), + PairControlError::AlreadyPaired ); - assert_eq!(result.unwrap_err(), PairControlError::TargetNotActive); } #[test] @@ -1095,15 +898,10 @@ mod tests { let (hub, events) = hub_with_events(); let stage_id = StageId::new("code", 1); let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage_id, "ses_01", control_handle(&handle))); + attach(&hub, &stage_id, "ses_01", &handle); let pair_id = PairId::new(); - hub.start_pair( - RunId::new(), - pair_id, - pair_target(&stage_id, "ses_01"), - None, - ) - .unwrap(); + hub.start_pair(RunId::new(), pair_id, pair_target(&stage_id), None) + .unwrap(); assert!(hub.detach(&stage_id, "ses_01")); @@ -1120,17 +918,4 @@ mod tests { "run.pair.ended" ]); } - - #[test] - fn per_session_queue_evicts_oldest_at_cap() { - let hub = SteeringHub::for_tests(); - let stage = StageId::new("a", 1); - let handle = SessionControlHandle::new(); - assert!(hub.attach_handle(&stage, "session-a", control_handle(&handle))); - - for i in 0..(super::PER_SESSION_QUEUE_CAP + 5) { - hub.deliver_steer(format!("m{i}"), None); - } - assert_eq!(handle.queue_len(), super::PER_SESSION_QUEUE_CAP); - } }