scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-14 23:22:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

Run steering over pebble's bus and keep only fabro's attribution

Browse source 
Pebble's `SteeringBus` now owns the map of live sessions, the buffer for
steers that arrive between sessions, fan-out of steers and interrupts, the
close-the-door detach, and the hold that keeps a paired session open. The
hub keeps what only fabro knows: pair records, principals, stage ids, and
the run events that put bus activity on the run's stream in the order its
consumers expect. `PebbleControlHandle` is gone, since the coding agent's
control handle is a bus session natively; the ACP session joins the bus
through a small adapter and carries pebble's steering message end to end,
so a human steer keeps its author on the ACP `agent.steering.injected`
event. A pair message that evicted an older steer is now accepted and the
eviction recorded, where before it was queued and reported as refused.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
Bryan Helmkamp 2026-09-11 20:35:31 -06:00
parent d0f97c25f7
commit a33c9069cf
No known key found for this signature in database
11 changed files with 500 additions and 812 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
Cargo.lock generated
View file

@ -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",

4
Cargo.toml
View file

@ -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" }

2
lib/components/fabro-acp/Cargo.toml
View file

@ -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

11
lib/components/fabro-acp/src/session.rs
View file

@ -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<dyn Fn() -> bool + Send + Sync>;
pub type AcpSteerPromptCallback = Arc<dyn Fn(String, Option<Principal>) + Send + Sync>;
pub type AcpSteerPromptCallback = Arc<dyn Fn(String, Option<Actor>) + 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<Principal>) {
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;

13
lib/components/fabro-acp/tests/session.rs
View file

@ -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)),

2
lib/components/fabro-workflow/src/event.rs
View file

@ -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;

2
lib/components/fabro-workflow/src/event/stored_fields.rs
View file

@ -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<Principal> {
pub fn principal_from_actor(actor: &Actor) -> Option<Principal> {
match actor {
Actor::Agent { id } => Some(Principal::Agent {
session_id: id.clone(),

69
lib/components/fabro-workflow/src/handler/llm/acp.rs
View file

@ -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<Principal>| {
Arc::new(move |text: String, actor: Option<Actor>| {
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<dyn Fn(String, Option<Principal>) + Send + Sync>
}) as Arc<dyn Fn(String, Option<Actor>) + 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<dyn ActiveControlHandle>),
Arc::new(AcpSteerable(handle.clone())),
)
.map(Some)
}
}
impl ActiveControlHandle for AcpControlHandle {
fn enqueue_bounded(&self, item: SteeringItem, cap: usize) -> Option<SteeringItem> {
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<Principal>) {
Self::interrupt(self, actor);
fn interrupt(&self) -> bool {
self.0.interrupt();
true
}
fn interrupt_then_enqueue_bounded(
&self,
item: SteeringItem,
cap: usize,
) -> Option<SteeringItem> {
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()
}
}

111
lib/components/fabro-workflow/src/handler/llm/activation_lease.rs
View file

@ -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<dyn ActiveControlHandle>,
session: Arc<dyn SteerableSession>,
) -> Result<Arc<Self>, 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<Mutex<Vec<SteeringItem>>>,
queue: Mutex<Vec<SteeringMessage>>,
}
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<SteeringItem> {
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<Principal>) {}
fn interrupt_then_enqueue_bounded(
&self,
item: SteeringItem,
cap: usize,
) -> Option<SteeringItem> {
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<dyn ActiveControlHandle> {
Arc::new(handle.clone())
fn session(handle: &Arc<SessionControlHandle>) -> Arc<dyn SteerableSession> {
Arc::clone(handle) as Arc<dyn SteerableSession>
}
#[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!(

130
lib/components/fabro-workflow/src/handler/llm/pebble.rs
View file

@ -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<Option<SteeringLease>>,
}
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<SteeringItem> {
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<SteeringItem> {
let outcome = self.control.queue_steering(Self::message(&item));
Self::rejected(item, outcome)
}
fn interrupt(&self, _actor: Option<Principal>) {
self.control.interrupt();
}
fn interrupt_then_enqueue_bounded(
&self,
item: SteeringItem,
_cap: usize,
) -> Option<SteeringItem> {
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<PebbleControlHandle>,
handle: CodingAgentControlHandle,
lease: Option<Arc<ActivationLease>>,
total_usage: TokenCounts,
total_cost: Option<UsdMicros>,
@ -375,7 +274,7 @@ struct LiveAgent {
}
impl LiveAgent {
fn new(agent: CodingAgent, handle: Arc<PebbleControlHandle>) -> 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<dyn ActiveControlHandle> =
Arc::clone(&live.handle) as Arc<dyn ActiveControlHandle>;
let session: Arc<dyn SteerableSession> = 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) =

963
lib/components/fabro-workflow/src/steering_hub.rs
View file

File diff suppressed because it is too large Load diff