Skip to content

feat(acp): persist bounded turn audit trail #5184

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

Closed
Tiago-Sonar wants to merge 1 commit into from
Closed

feat(acp): persist bounded turn audit trail #5184

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions crates/buzz-acp/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,6 +53,7 @@ url = { workspace = true }
sha2 = { workspace = true }
base64 = "0.22"
hex = { workspace = true }
atomic-write-file = "0.3"

# Logging
tracing = { workspace = true }
Expand Down
54 changes: 53 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 @@ -468,6 +468,26 @@ pub struct CliArgs {
#[arg(long, env = "BUZZ_ACP_RELAY_OBSERVER", default_value_t = false)]
pub relay_observer: bool,

/// Persist a bounded, content-free local audit trail for inbound turns.
/// `--turn-audit=false` disables it.
#[arg(
long,
env = "BUZZ_ACP_TURN_AUDIT",
default_value_t = true,
action = clap::ArgAction::Set,
num_args = 0..=1,
default_missing_value = "true",
)]
pub turn_audit: bool,

/// Maximum number of inbound event records kept in the local turn audit.
#[arg(long, env = "BUZZ_ACP_TURN_AUDIT_RETENTION", default_value_t = 1_000)]
pub turn_audit_retention: usize,

/// Directory for the local turn-audit file. Defaults to `<cwd>/.buzz-acp/`.
#[arg(long, env = "BUZZ_ACP_TURN_AUDIT_DIR")]
pub turn_audit_dir: Option<PathBuf>,

/// Exit after this many seconds with no dispatched events and no turn in flight.
/// 0 disables inactivity self-termination.
#[arg(long, env = "BUZZ_ACP_EXIT_AFTER_INACTIVITY", default_value_t = 0)]
Expand Down Expand Up @@ -549,6 +569,12 @@ pub struct Config {
pub has_generated_codex_config: bool,
/// Whether to publish encrypted observer frames through the relay.
pub relay_observer: bool,
/// Whether to persist the content-free local turn audit.
pub turn_audit: bool,
/// Maximum retained inbound event records in the turn audit.
pub turn_audit_retention: usize,
/// Audit directory override. `None` uses `<cwd>/.buzz-acp/`.
pub turn_audit_dir: Option<PathBuf>,
/// Seconds without dispatched events before an idle harness exits. 0 = disabled.
pub exit_after_inactivity_secs: u64,
/// Whether ACP/LLM subprocess initialization is deferred until accepted work arrives.
Expand Down Expand Up @@ -1099,6 +1125,9 @@ impl Config {
persona_env_vars,
has_generated_codex_config,
relay_observer: args.relay_observer,
turn_audit: args.turn_audit,
turn_audit_retention: args.turn_audit_retention.clamp(1, 10_000),
turn_audit_dir: args.turn_audit_dir,
exit_after_inactivity_secs: args.exit_after_inactivity,
lazy_pool: args.lazy_pool,
agent_owner: args.agent_owner.map(|s| s.trim().to_ascii_lowercase()),
Expand All @@ -1125,7 +1154,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={} turn_audit={} turn_audit_retention={} {}{}",
self.relay_url,
self.keys.public_key().to_hex(),
self.agent_command,
Expand All @@ -1146,6 +1175,8 @@ impl Config {
self.memory_enabled,
self.model.as_deref().unwrap_or("(agent default)"),
self.permission_mode,
self.turn_audit,
self.turn_audit_retention,
respond_to_detail,
allowed_respond_to_detail,
)
Expand Down Expand Up @@ -1470,6 +1501,9 @@ mod tests {
persona_env_vars: vec![],
has_generated_codex_config: false,
relay_observer: false,
turn_audit: true,
turn_audit_retention: 1_000,
turn_audit_dir: None,
exit_after_inactivity_secs: 0,
lazy_pool: false,
agent_owner: None,
Expand Down Expand Up @@ -2186,6 +2220,24 @@ channels = "ALL"
assert_eq!(configured.exit_after_inactivity, 120);
}

#[test]
fn turn_audit_defaults_on_and_can_be_disabled() {
let key = "0".repeat(64);
let default = CliArgs::parse_from([
"buzz-acp",
"--private-key",
&key,
"--turn-audit-retention",
"250",
]);
assert!(default.turn_audit);
assert_eq!(default.turn_audit_retention, 250);

let disabled =
CliArgs::parse_from(["buzz-acp", "--private-key", &key, "--turn-audit=false"]);
assert!(!disabled.turn_audit);
}

#[test]
fn lazy_pool_defaults_off() {
let key = "0".repeat(64);
Expand Down
133 changes: 130 additions & 3 deletions crates/buzz-acp/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@ mod pool_lifecycle;
mod queue;
mod relay;
mod setup_mode;
mod turn_audit;
mod usage;

pub use usage::TurnUsage;
Expand Down Expand Up @@ -114,6 +115,28 @@ fn emit_runtime_lifecycle(
}
}

fn emit_inbound_audit_stage(
observer: Option<&observer::ObserverHandle>,
kind: &str,
channel_id: Uuid,
event_id: &str,
reason: Option<&str>,
) {
let Some(observer) = observer else {
return;
};
let mut payload = serde_json::json!({"eventId": event_id});
if let Some(reason) = reason {
payload["reason"] = serde_json::Value::String(reason.to_string());
}
observer.emit(
kind,
None,
&observer::context_for(Some(channel_id), None, None),
payload,
);
}

/// Resolve the agent's owner pubkey at startup.
///
/// Priority:
Expand Down Expand Up @@ -1578,9 +1601,34 @@ async fn tokio_main() -> Result<()> {

tracing::info!("buzz-acp starting: {}", config.summary());

let observer = config
.relay_observer
.then(observer::ObserverHandle::in_process);
// The local audit consumes the same in-process observer feed as the
// optional encrypted relay observer, but persists only a strict metadata
// projection. Keep the bus alive when either consumer is enabled.
let observer = if config.relay_observer {
Some(observer::ObserverHandle::in_process())
} else if config.turn_audit {
Some(observer::ObserverHandle::in_process_unbuffered())
} else {
None
};
let turn_audit_task = if config.turn_audit {
observer.as_ref().map(|observer| {
let base_dir = config.turn_audit_dir.clone().unwrap_or_else(|| {
std::env::current_dir()
.unwrap_or_else(|_| std::path::PathBuf::from("."))
.join(".buzz-acp")
});
let path = turn_audit::audit_path(
&base_dir,
&config.keys.public_key().to_hex(),
&config.relay_url,
);
tracing::info!(target: "turn_audit", path = %path.display(), "local turn audit enabled");
turn_audit::spawn(observer, path, config.turn_audit_retention)
})
} else {
None
};
if let Some(handle) = &observer {
handle.emit(
"harness_started",
Expand Down Expand Up @@ -2326,8 +2374,24 @@ async fn tokio_main() -> Result<()> {
continue;
}

let inbound_event_id = buzz_event.event.id.to_hex();
emit_inbound_audit_stage(
observer.as_ref(),
"turn_received",
buzz_event.channel_id,
&inbound_event_id,
None,
);

if config.ignore_self && buzz_event.event.pubkey.to_hex() == pubkey_hex {
tracing::debug!(channel_id = %buzz_event.channel_id, "dropping self-authored event");
emit_inbound_audit_stage(
observer.as_ref(),
"turn_rejected",
buzz_event.channel_id,
&inbound_event_id,
Some("self_authored"),
);
continue;
}

Expand All @@ -2348,6 +2412,13 @@ async fn tokio_main() -> Result<()> {
"shutdown command from owner — exiting gracefully"
);
let _ = shutdown_tx.send(());
emit_inbound_audit_stage(
observer.as_ref(),
"turn_rejected",
buzz_event.channel_id,
&inbound_event_id,
Some("owner_shutdown_control"),
);
continue;
}
}
Expand Down Expand Up @@ -2383,6 +2454,13 @@ async fn tokio_main() -> Result<()> {
"!cancel received but no in-flight task — no-op"
);
}
emit_inbound_audit_stage(
observer.as_ref(),
"turn_rejected",
buzz_event.channel_id,
&inbound_event_id,
Some("owner_cancel_control"),
);
continue; // consume event — do NOT push to queue
}
}
Expand Down Expand Up @@ -2428,6 +2506,13 @@ async fn tokio_main() -> Result<()> {
"!rotate received — invalidated idle channel session(s)"
);
}
emit_inbound_audit_stage(
observer.as_ref(),
"turn_rejected",
buzz_event.channel_id,
&inbound_event_id,
Some("owner_rotate_control"),
);
continue; // consume event — do NOT push to queue
}
}
Expand Down Expand Up @@ -2469,6 +2554,13 @@ async fn tokio_main() -> Result<()> {
is_dm,
"inbound author gate — dropping event"
);
emit_inbound_audit_stage(
observer.as_ref(),
"turn_rejected",
buzz_event.channel_id,
&inbound_event_id,
Some("author_gate"),
);
continue;
}
}
Expand All @@ -2478,6 +2570,24 @@ async fn tokio_main() -> Result<()> {
Some(m) => m.prompt_tag,
None => {
tracing::debug!(channel_id = %buzz_event.channel_id, kind = buzz_event.event.kind.as_u16(), "event matched no rule — dropping");
let mentioned = buzz_event.event.tags.iter().any(|tag| {
let values = tag.as_slice();
values.first().map(|value| value.as_str()) == Some("p")
&& values.get(1).map(|value| value.as_str())
== Some(pubkey_hex.as_str())
});
let reason = if !mentioned && rules.iter().any(|rule| rule.require_mention) {
"mention_required"
} else {
"subscription_rule_no_match"
};
emit_inbound_audit_stage(
observer.as_ref(),
"turn_rejected",
buzz_event.channel_id,
&inbound_event_id,
Some(reason),
);
continue;
}
};
Expand All @@ -2503,6 +2613,13 @@ async fn tokio_main() -> Result<()> {
received_at: std::time::Instant::now(),
prompt_tag,
});
emit_inbound_audit_stage(
observer.as_ref(),
if accepted { "turn_queued" } else { "turn_rejected" },
buzz_event.channel_id,
&event_id_hex,
(!accepted).then_some("queue_policy_drop"),
);
// 👀 — immediate "seen" reaction, only if the event
// was actually queued (not dropped by DedupMode::Drop).
// Fire-and-forget: on rare fast-failure paths the
Expand Down Expand Up @@ -3044,6 +3161,10 @@ async fn tokio_main() -> Result<()> {
if let Some(handle) = relay_observer_publisher_task.take() {
handle.abort();
}
// Persist every frame already queued on the observer bus before shutdown.
if let Some(task) = turn_audit_task {
task.shutdown().await;
}

// Graceful relay shutdown — sends WebSocket close frame and waits up to 5s
// for the background task to finish, rather than aborting immediately (#40).
Expand Down Expand Up @@ -6206,6 +6327,9 @@ mod build_mcp_servers_tests {
persona_env_vars: vec![],
has_generated_codex_config: false,
relay_observer: false,
turn_audit: false,
turn_audit_retention: 1_000,
turn_audit_dir: None,
exit_after_inactivity_secs: 0,
lazy_pool: false,
agent_owner: None,
Expand Down Expand Up @@ -6428,6 +6552,9 @@ mod error_outcome_emission_tests {
persona_env_vars: vec![],
has_generated_codex_config: false,
relay_observer: false,
turn_audit: false,
turn_audit_retention: 1_000,
turn_audit_dir: None,
exit_after_inactivity_secs: 0,
lazy_pool: false,
agent_owner: None,
Expand Down
Loading