Skip to content

feat: let requesters observe shared agent turns #2779

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
akkikumar72 wants to merge 2 commits into
base: main
Choose a base branch
from
+1,277 −274
Open

feat: let requesters observe shared agent turns #2779

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bfcb73d
feat: let requesters observe shared agent turns
akkikumar72 Jul 24, 2026
88b990a
fix(acp): secure requester observer turns
akkikumar72 Jul 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 59 additions & 13 deletions crates/buzz-acp/src/acp.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -510,6 +510,12 @@ impl AcpClient {
self.observer_context = context;
}

/// Return the current observer routing context for terminal events emitted
/// after this client moves back to the harness loop.
pub(crate) fn observer_context(&self) -> &ObserverContext {
&self.observer_context
}

/// Return a clone of the observer handle, if attached.
pub(crate) fn observer_handle(&self) -> Option<ObserverHandle> {
self.observer.clone()
Expand Down Expand Up @@ -1211,14 +1217,18 @@ impl AcpClient {
// so the ack_tx oneshot is never leaked silently).
let mut steer_rx = self.steer_rx.take();

// Tracks the in-flight steer write: `(request_id, ack_tx)`. While
// Tracks the in-flight steer write:
// `(request_id, ack_tx, requester_pubkey)`. While
// `Some`, the steer arm is gated off so we don't stack writes,
// and a response matching `id` is routed to the ack_tx instead
// of being treated as the prompt result. Drained on every return
// path with `PromptCompletedNeutral` so callers are never left
// hanging.
let mut pending_steer: Option<(u64, tokio::sync::oneshot::Sender<crate::pool::SteerAck>)> =
None;
let mut pending_steer: Option<(
u64,
tokio::sync::oneshot::Sender<crate::pool::SteerAck>,
String,
)> = None;

let now = Instant::now();
let mut idle_deadline = now + idle_timeout;
Expand All @@ -1243,7 +1253,7 @@ impl AcpClient {
// exists). Check the classified deadline here so a steady-
// stream agent is still bounded.
if Instant::now() >= next_deadline {
if let Some((_, ack_tx)) = pending_steer.take() {
if let Some((_, ack_tx, _)) = pending_steer.take() {
// Prompt is timing out — release the withheld event via
// PromptCompletedNeutral (no fallback signal: there is
// no in-flight turn to signal once we return, and
Expand Down Expand Up @@ -1320,7 +1330,8 @@ impl AcpClient {
);
match self.write_ndjson(&msg).await {
Ok(()) => {
pending_steer = Some((id, req.ack_tx));
pending_steer =
Some((id, req.ack_tx, req.requester_pubkey));
}
Err(e) => {
tracing::warn!(
Expand All @@ -1343,7 +1354,7 @@ impl AcpClient {
// would catch this anyway, but firing the deadline arm
// here makes the wakeup immediate (no extra reader poll
// round-trip when stdout is idle).
if let Some((_, ack_tx)) = pending_steer.take() {
if let Some((_, ack_tx, _)) = pending_steer.take() {
let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
}
if idle_fires_first {
Expand All @@ -1367,21 +1378,21 @@ impl AcpClient {

match read_result {
None => {
if let Some((_, ack_tx)) = pending_steer.take() {
if let Some((_, ack_tx, _)) = pending_steer.take() {
let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
}
return Err(AcpError::AgentExited);
}
Some(Err(LinesCodecError::MaxLineLengthExceeded)) => {
if let Some((_, ack_tx)) = pending_steer.take() {
if let Some((_, ack_tx, _)) = pending_steer.take() {
let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
}
return Err(AcpError::Protocol(
"agent stdout line exceeded 10MB limit".into(),
));
}
Some(Err(e)) => {
if let Some((_, ack_tx)) = pending_steer.take() {
if let Some((_, ack_tx, _)) = pending_steer.take() {
let _ = ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
}
return Err(AcpError::Io(std::io::Error::other(e)));
Expand Down Expand Up @@ -1424,13 +1435,14 @@ impl AcpClient {
// share the `no method` guard.
if let Some(id) = msg.get("id") {
if msg.get("method").is_none() {
if let Some((steer_id, _)) = pending_steer.as_ref() {
if let Some((steer_id, _, _)) = pending_steer.as_ref() {
if *id == serde_json::json!(*steer_id) {
// Take the ack_tx out and route the
// response. We do not return — keep
// reading until the prompt response
// arrives.
let (_, ack_tx) = pending_steer.take().expect("just checked");
let (_, ack_tx, requester_pubkey) =
pending_steer.take().expect("just checked");
let ack = if let Some(error) = msg.get("error") {
let code = error
.get("code")
Expand All @@ -1441,6 +1453,12 @@ impl AcpClient {
crate::pool::SteerError::AgentError { code, message },
)
} else {
self.observer_context
.add_requester_pubkey(requester_pubkey);
self.observe(
"turn_liveness",
serde_json::json!({"source": "native_steer"}),
);
let renew_now = Instant::now();
let new_deadline = renew_now + max_duration;
if new_deadline > hard_deadline {
Expand All @@ -1458,13 +1476,13 @@ impl AcpClient {
}
if *id == serde_json::json!(expected_id) {
if let Some(error) = msg.get("error") {
if let Some((_, ack_tx)) = pending_steer.take() {
if let Some((_, ack_tx, _)) = pending_steer.take() {
let _ = ack_tx
.send(crate::pool::SteerAck::PromptCompletedNeutral);
}
return Err(agent_error_from_json(error));
}
if let Some((_, ack_tx)) = pending_steer.take() {
if let Some((_, ack_tx, _)) = pending_steer.take() {
let _ =
ack_tx.send(crate::pool::SteerAck::PromptCompletedNeutral);
}
Expand Down Expand Up @@ -3194,6 +3212,7 @@ mod tests {
steer_tx
.send(crate::pool::SteerRequest {
prompt_blocks: vec!["test steer body".into()],
requester_pubkey: "requester-a".into(),
ack_tx,
})
.await
Expand Down Expand Up @@ -3248,6 +3267,19 @@ mod tests {
echo '{\"jsonrpc\":\"2.0\",\"id\":0,\"result\":{\"stopReason\":\"end_turn\"}}'; \
sleep 10";
let mut client = spawn_script(script).await;
let observer = crate::observer::ObserverHandle::in_process();
let initial_requester = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
let steered_requester = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
client.set_observer(Some(observer.clone()), 0);
client.set_observer_context(
crate::observer::context_for_turn(
None,
Some("sess-test".into()),
"turn-test".into(),
"2026-07-24T10:00:00Z".into(),
)
.with_requester_pubkeys(vec![initial_requester.into()]),
);

// Set active_run_id via a synthesized session_info_update so the
// steer arm has a non-None value to read at write time.
Expand All @@ -3263,6 +3295,7 @@ mod tests {
steer_tx
.send(crate::pool::SteerRequest {
prompt_blocks: vec!["test steer body".into()],
requester_pubkey: steered_requester.into(),
ack_tx,
})
.await
Expand Down Expand Up @@ -3301,6 +3334,18 @@ mod tests {
crate::pool::SteerAck::Success => {}
other => panic!("expected SteerAck::Success, got {other:?}"),
}
let liveness = observer
.snapshot()
.into_iter()
.find(|event| {
event.kind == "turn_liveness"
&& event.payload["source"] == serde_json::json!("native_steer")
})
.expect("successful native steer should emit requester-visible liveness");
assert_eq!(
liveness.requester_pubkeys,
[initial_requester.to_string(), steered_requester.to_string()]
);
}

/// Steer-success renewal keeps the turn alive past the original hard
Expand Down Expand Up @@ -3335,6 +3380,7 @@ mod tests {
steer_tx
.send(crate::pool::SteerRequest {
prompt_blocks: vec!["steer body".into()],
requester_pubkey: "requester-b".into(),
ack_tx,
})
.await
Expand Down
56 changes: 55 additions & 1 deletion crates/buzz-acp/src/config.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,6 +111,26 @@ impl std::fmt::Display for RespondTo {
}
}

/// Who receives encrypted relay observer telemetry.
///
/// The registered owner always receives every frame. `Requester` additionally
/// sends turn-scoped activity to the authors whose events triggered that turn.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, clap::ValueEnum)]
pub enum ObserverVisibility {
#[default]
OwnerOnly,
Requester,
}

impl std::fmt::Display for ObserverVisibility {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::OwnerOnly => f.write_str("owner-only"),
Self::Requester => f.write_str("requester"),
}
}
}

/// Permission mode for agents that support `session/set_config_option` with
/// `configId: "mode"` (e.g. `claude-agent-acp`).
///
Expand Down Expand Up @@ -468,6 +488,15 @@ pub struct CliArgs {
#[arg(long, env = "BUZZ_ACP_RELAY_OBSERVER", default_value_t = false)]
pub relay_observer: bool,

/// Observer visibility policy. The owner always receives every frame;
/// `requester` additionally sends turn activity to triggering authors.
#[arg(
long,
env = "BUZZ_ACP_OBSERVER_VISIBILITY",
default_value_t = ObserverVisibility::OwnerOnly
)]
pub observer_visibility: ObserverVisibility,

/// Connect and subscribe before starting the ACP/LLM subprocess pool.
#[arg(long, env = "BUZZ_ACP_LAZY_POOL", default_value_t = false)]
pub lazy_pool: bool,
Expand Down Expand Up @@ -541,6 +570,8 @@ pub struct Config {
pub has_generated_codex_config: bool,
/// Whether to publish encrypted observer frames through the relay.
pub relay_observer: bool,
/// Who receives encrypted turn-scoped observer telemetry.
pub observer_visibility: ObserverVisibility,
/// Whether ACP/LLM subprocess initialization is deferred until accepted work arrives.
pub lazy_pool: bool,
/// Agent owner pubkey (hex). Used for `--respond-to=owner-only` gate.
Expand Down Expand Up @@ -999,6 +1030,7 @@ impl Config {
persona_env_vars,
has_generated_codex_config,
relay_observer: args.relay_observer,
observer_visibility: args.observer_visibility,
lazy_pool: args.lazy_pool,
agent_owner: args.agent_owner.map(|s| s.trim().to_ascii_lowercase()),
no_base_prompt: args.no_base_prompt,
Expand All @@ -1024,7 +1056,7 @@ impl Config {
format!(" allowed_respond_to=[{}]", modes.join(","))
};
format!(
"relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}{}",
"relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} observer_visibility={} {}{}",
self.relay_url,
self.keys.public_key().to_hex(),
self.agent_command,
Expand All @@ -1045,6 +1077,7 @@ impl Config {
self.memory_enabled,
self.model.as_deref().unwrap_or("(agent default)"),
self.permission_mode,
self.observer_visibility,
respond_to_detail,
allowed_respond_to_detail,
)
Expand Down Expand Up @@ -1368,6 +1401,7 @@ mod tests {
persona_env_vars: vec![],
has_generated_codex_config: false,
relay_observer: false,
observer_visibility: ObserverVisibility::OwnerOnly,
lazy_pool: false,
agent_owner: None,
no_base_prompt: false,
Expand Down Expand Up @@ -2048,6 +2082,26 @@ channels = "ALL"
assert!(CliArgs::parse_from(["buzz-acp", "--private-key", &key, "--lazy-pool"]).lazy_pool);
}

#[test]
fn observer_visibility_defaults_to_owner_only() {
let key = "0".repeat(64);
let args = CliArgs::parse_from(["buzz-acp", "--private-key", &key]);
assert_eq!(args.observer_visibility, ObserverVisibility::OwnerOnly);
}

#[test]
fn observer_visibility_accepts_requester() {
let key = "0".repeat(64);
let args = CliArgs::parse_from([
"buzz-acp",
"--private-key",
&key,
"--observer-visibility",
"requester",
]);
assert_eq!(args.observer_visibility, ObserverVisibility::Requester);
}

#[test]
fn test_summary_includes_agents_and_heartbeat() {
let config = test_config(SubscribeMode::Mentions);
Expand Down
Loading