use std::collections::HashMap;
use std::sync::LazyLock;
use std::sync::Mutex;
use std::sync::atomic::AtomicU64;
use crate::util::UnwrapPoison;
use chrono::{DateTime, Utc};
use serde::Serialize;
use tokio_util::sync::CancellationToken;
static NEXT_ENTRY_GENERATION: AtomicU64 = AtomicU64::new(1);
static NEXT_TOOL_ID: AtomicU64 = AtomicU64::new(1);
#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize)]
pub enum ParentKey {
Ticket(String),
AnalyzeRound(String),
Research(String),
}
#[derive(Clone, Debug, Serialize)]
pub struct RunningTool {
pub name: String,
pub args: Vec<(String, String)>,
}
#[derive(Clone, Debug, Serialize)]
pub struct AgentHandle {
pub agent_id: String,
pub role: String,
pub ticket_id: Option<String>,
pub workspace_path: String,
pub workspace_name: String,
pub parent_key: Option<ParentKey>,
pub parent_label: Option<String>,
pub started_at: DateTime<Utc>,
pub label: String,
pub current_tools: Vec<RunningTool>,
pub last_tool: Option<RunningTool>,
pub activity: Option<String>,
pub session_tokens: Option<u64>,
}
#[derive(Clone, Debug)]
pub(crate) struct AgentTracking {
pub agent_id: String,
pub generation: u64,
pub role: String,
pub workspace: String,
}
struct AgentEntry {
generation: u64,
handle: AgentHandle,
cancel_token: CancellationToken,
current_tools: Vec<RunningToolEntry>,
last_tool: Option<RunningTool>,
activity: Option<String>,
}
#[derive(Clone, Debug)]
struct RunningToolEntry {
id: u64,
tool: RunningTool,
}
impl RunningToolEntry {
fn to_handle(&self) -> RunningTool {
self.tool.clone()
}
}
fn tool_arg_pairs(args: &serde_json::Value) -> Vec<(String, String)> {
match args {
serde_json::Value::Object(map) => map
.iter()
.map(|(k, v)| (k.clone(), json_value_to_display_string(v)))
.collect(),
other => vec![("args".to_string(), json_value_to_display_string(other))],
}
}
fn json_value_to_display_string(v: &serde_json::Value) -> String {
match v {
serde_json::Value::String(s) => s.clone(),
other => serde_json::to_string(other).unwrap_or_default(),
}
}
#[derive(Default)]
pub struct AgentRegistry {
inner: Mutex<HashMap<String, AgentEntry>>,
}
impl AgentRegistry {
#[expect(clippy::too_many_arguments)] pub fn register(
&self,
agent_id: String,
role: String,
ticket_id: Option<String>,
ws: &crate::Workspace,
label: String,
cancel_token: CancellationToken,
parent_key: Option<ParentKey>,
parent_label: Option<String>,
) -> u64 {
let generation = NEXT_ENTRY_GENERATION.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let handle = AgentHandle {
agent_id: agent_id.clone(),
role,
ticket_id,
workspace_path: ws.path.clone(),
workspace_name: ws.name.clone(),
parent_key,
parent_label,
started_at: Utc::now(),
label,
current_tools: Vec::new(),
last_tool: None,
activity: None,
session_tokens: None,
};
let mut map = self.inner.lock().unwrap_poison();
if let Some(old) = map.remove(&agent_id) {
old.cancel_token.cancel();
}
map.insert(
agent_id,
AgentEntry {
generation,
handle,
cancel_token,
current_tools: Vec::new(),
last_tool: None,
activity: None,
},
);
generation
}
pub fn tool_started(
&self,
agent_id: &str,
generation: u64,
name: &str,
args: &serde_json::Value,
) -> RunningToolGuard {
let tool_id = NEXT_TOOL_ID.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let args = tool_arg_pairs(args);
{
let mut map = self.inner.lock().unwrap_poison();
if let Some(entry) = map.get_mut(agent_id)
&& entry.generation == generation
{
entry.current_tools.push(RunningToolEntry {
id: tool_id,
tool: RunningTool {
name: name.to_string(),
args,
},
});
}
}
RunningToolGuard {
agent_id: agent_id.to_string(),
generation,
tool_id,
}
}
fn tool_finished(&self, agent_id: &str, generation: u64, tool_id: u64) {
let mut map = self.inner.lock().unwrap_poison();
if let Some(entry) = map.get_mut(agent_id)
&& entry.generation == generation
&& let Some(pos) = entry.current_tools.iter().position(|t| t.id == tool_id)
{
let removed = entry.current_tools.remove(pos);
entry.last_tool = Some(removed.tool);
}
}
pub fn activity_started(&self, agent_id: &str, generation: u64, label: &str) -> ActivityGuard {
{
let mut map = self.inner.lock().unwrap_poison();
if let Some(entry) = map.get_mut(agent_id)
&& entry.generation == generation
{
entry.activity = Some(label.to_string());
}
}
ActivityGuard {
agent_id: agent_id.to_string(),
generation,
}
}
fn activity_finished(&self, agent_id: &str, generation: u64) {
let mut map = self.inner.lock().unwrap_poison();
if let Some(entry) = map.get_mut(agent_id)
&& entry.generation == generation
{
entry.activity = None;
}
}
pub fn set_session_tokens(&self, agent_id: &str, generation: u64, token_length: u64) {
let mut map = self.inner.lock().unwrap_poison();
if let Some(entry) = map.get_mut(agent_id)
&& entry.generation == generation
{
entry.handle.session_tokens = Some(token_length);
}
}
fn cancel(&self, agent_id: &str) {
let mut map = self.inner.lock().unwrap_poison();
if let Some(entry) = map.remove(agent_id) {
entry.cancel_token.cancel();
}
}
fn cancel_matching<F>(&self, predicate: F)
where
F: Fn(&AgentEntry) -> bool,
{
let to_cancel: Vec<String> = {
let map = self.inner.lock().unwrap_poison();
map.iter()
.filter(|(_, entry)| predicate(entry))
.map(|(id, _)| id.clone())
.collect()
};
for agent_id in to_cancel {
self.cancel(&agent_id);
}
}
pub fn cancel_by_ticket_id(&self, ticket_id: &str) {
self.cancel_matching(|entry| entry.handle.ticket_id.as_deref() == Some(ticket_id));
}
pub fn cancel_by_role_and_workspace_path(&self, role: &str, ws_path: &str) {
self.cancel_matching(|entry| {
entry.handle.role == role && entry.handle.workspace_path == ws_path
});
}
pub fn cancel_by_parent_key(&self, parent: &ParentKey) {
let parent = parent.clone();
self.cancel_matching(move |entry| entry.handle.parent_key.as_ref() == Some(&parent));
}
#[must_use]
pub fn list(&self) -> Vec<AgentHandle> {
self.inner
.lock()
.unwrap_poison()
.values()
.map(|e| {
let mut handle = e.handle.clone();
handle.current_tools = e
.current_tools
.iter()
.map(RunningToolEntry::to_handle)
.collect();
handle.last_tool.clone_from(&e.last_tool);
handle.activity.clone_from(&e.activity);
handle
})
.collect()
}
#[must_use]
pub fn contains(&self, agent_id: &str) -> bool {
self.inner.lock().unwrap_poison().contains_key(agent_id)
}
pub fn shutdown_all(&self) {
let entries: Vec<(String, CancellationToken)> = self
.inner
.lock()
.unwrap_poison()
.drain()
.map(|(id, entry)| (id, entry.cancel_token))
.collect();
for (_id, token) in entries {
token.cancel();
}
}
pub fn deregister(&self, agent_id: &str, generation: u64) {
let mut map = self.inner.lock().unwrap_poison();
if let Some(entry) = map.get(agent_id)
&& entry.generation == generation
{
map.remove(agent_id);
}
}
}
pub struct RunningToolGuard {
agent_id: String,
generation: u64,
tool_id: u64,
}
impl Drop for RunningToolGuard {
fn drop(&mut self) {
AGENT_REGISTRY.tool_finished(&self.agent_id, self.generation, self.tool_id);
}
}
pub struct ActivityGuard {
agent_id: String,
generation: u64,
}
impl Drop for ActivityGuard {
fn drop(&mut self) {
AGENT_REGISTRY.activity_finished(&self.agent_id, self.generation);
}
}
pub static AGENT_REGISTRY: LazyLock<AgentRegistry> = LazyLock::new(AgentRegistry::default);
#[cfg(test)]
mod tests {
use super::*;
fn register_test_agent(agent_id: &str, ws_path: &str) -> u64 {
let ws = crate::Workspace {
name: agent_id.to_string(),
path: ws_path.to_string(),
..Default::default()
};
AGENT_REGISTRY.register(
agent_id.to_string(),
"analyst".to_string(),
None,
&ws,
"test".to_string(),
CancellationToken::new(),
None,
None,
)
}
#[test]
fn tool_started_shows_in_list_and_guard_removes_it() {
let agent_id = format!("tool_guard_{}", crate::generate_suffix());
let generation = register_test_agent(&agent_id, "/tmp/ws");
{
let guard = AGENT_REGISTRY.tool_started(
&agent_id,
generation,
"read",
&serde_json::json!({"path": "/tmp/ws/file.rs"}),
);
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent registered");
assert_eq!(h.current_tools.len(), 1);
assert_eq!(h.current_tools[0].name, "read");
assert!(
h.current_tools[0]
.args
.iter()
.any(|(k, v)| k == "path" && v == "/tmp/ws/file.rs")
);
drop(guard);
}
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent still registered");
assert!(
h.current_tools.is_empty(),
"guard drop must remove the tool entry"
);
AGENT_REGISTRY.deregister(&agent_id, generation);
}
#[test]
fn parallel_tools_coexist_and_remove_individually() {
let agent_id = format!("tool_par_{}", crate::generate_suffix());
let generation = register_test_agent(&agent_id, "/tmp/ws");
let guard_a = AGENT_REGISTRY.tool_started(
&agent_id,
generation,
"search",
&serde_json::json!({"query": "alpha"}),
);
let guard_b = AGENT_REGISTRY.tool_started(
&agent_id,
generation,
"read",
&serde_json::json!({"path": "/tmp/ws/b.rs"}),
);
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent registered");
assert_eq!(
h.current_tools.len(),
2,
"parallel tools must both be visible (honest representation)"
);
drop(guard_a);
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent registered");
assert_eq!(h.current_tools.len(), 1);
assert_eq!(h.current_tools[0].name, "read");
drop(guard_b);
AGENT_REGISTRY.deregister(&agent_id, generation);
}
#[test]
fn activity_shows_in_list_and_guard_clears_it() {
let agent_id = format!("activity_guard_{}", crate::generate_suffix());
let generation = register_test_agent(&agent_id, "/tmp/ws");
{
let guard = AGENT_REGISTRY.activity_started(&agent_id, generation, "extracting");
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent registered");
assert_eq!(
h.activity.as_deref(),
Some("extracting"),
"activity must be visible on the live card"
);
drop(guard);
}
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent still registered");
assert!(
h.activity.is_none(),
"guard drop must clear the activity label"
);
AGENT_REGISTRY.deregister(&agent_id, generation);
}
#[test]
fn stale_activity_guard_cannot_clear_replacement_agent() {
let agent_id = format!("activity_stale_{}", crate::generate_suffix());
let generation = register_test_agent(&agent_id, "/tmp/ws");
let stale_guard = AGENT_REGISTRY.activity_started(&agent_id, generation, "extracting");
AGENT_REGISTRY.deregister(&agent_id, generation);
let new_gen = register_test_agent(&agent_id, "/tmp/ws");
let fresh_guard = AGENT_REGISTRY.activity_started(&agent_id, new_gen, "transcribing");
drop(stale_guard);
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("replacement agent registered");
assert_eq!(
h.activity.as_deref(),
Some("transcribing"),
"stale guard must not clear the replacement's activity"
);
drop(fresh_guard);
AGENT_REGISTRY.deregister(&agent_id, new_gen);
}
#[test]
fn stale_guard_cannot_mutate_replacement_agent() {
let agent_id = format!("tool_stale_{}", crate::generate_suffix());
let generation = register_test_agent(&agent_id, "/tmp/ws");
let stale_guard = AGENT_REGISTRY.tool_started(
&agent_id,
generation,
"shell",
&serde_json::json!({"command": "ls"}),
);
AGENT_REGISTRY.deregister(&agent_id, generation);
let new_gen = register_test_agent(&agent_id, "/tmp/ws");
drop(stale_guard);
let fresh_guard = AGENT_REGISTRY.tool_started(
&agent_id,
new_gen,
"read",
&serde_json::json!({"path": "/tmp/ws/new.rs"}),
);
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("replacement registered");
assert_eq!(
h.current_tools.len(),
1,
"stale guard must not mutate the replacement's card"
);
assert_eq!(h.current_tools[0].name, "read");
drop(fresh_guard);
AGENT_REGISTRY.deregister(&agent_id, new_gen);
}
#[test]
fn tool_args_are_full_and_structured() {
let agent_id = format!("tool_full_{}", crate::generate_suffix());
let generation = register_test_agent(&agent_id, "/tmp/ws");
let secret = format!("token-{}", crate::generate_suffix());
let long_command = format!("echo {}", "a".repeat(1000));
let args = serde_json::json!({
"command": long_command,
"api_key": secret,
});
let guard = AGENT_REGISTRY.tool_started(&agent_id, generation, "shell", &args);
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent registered");
assert_eq!(h.current_tools.len(), 1);
let pairs = &h.current_tools[0].args;
assert!(
pairs.iter().any(|(k, v)| k == "api_key" && v == &secret),
"live view shows full unscrubbed values (deliberate divergence from durable logs)"
);
assert!(
pairs
.iter()
.any(|(k, v)| k == "command" && v == &long_command),
"values are not truncated at registration"
);
drop(guard);
AGENT_REGISTRY.deregister(&agent_id, generation);
}
#[test]
fn parent_key_and_workspace_name_round_trip() {
let agent_id = format!("parent_key_{}", crate::generate_suffix());
let ws = crate::Workspace {
name: "ws_parent".to_string(),
path: "/tmp/ws_parent".to_string(),
..Default::default()
};
let generation = AGENT_REGISTRY.register(
agent_id.clone(),
"engineer".to_string(),
Some("T-42".to_string()),
&ws,
"label".to_string(),
CancellationToken::new(),
Some(ParentKey::Ticket("T-42".to_string())),
Some("Fix login bug".to_string()),
);
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent registered");
assert_eq!(h.workspace_name, "ws_parent");
assert_eq!(h.parent_key, Some(ParentKey::Ticket("T-42".to_string())));
assert_eq!(
h.parent_label.as_deref(),
Some("Fix login bug"),
"parent label round-trips"
);
AGENT_REGISTRY.deregister(&agent_id, generation);
}
#[test]
fn completed_tool_is_kept_as_last_tool_until_replaced() {
let agent_id = format!("last_tool_{}", crate::generate_suffix());
let generation = register_test_agent(&agent_id, "/tmp/ws");
{
let guard = AGENT_REGISTRY.tool_started(
&agent_id,
generation,
"read",
&serde_json::json!({"path": "/tmp/ws/a.rs"}),
);
drop(guard);
}
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent registered");
assert!(h.current_tools.is_empty(), "no running tools left");
let last = h
.last_tool
.as_ref()
.expect("completed tool kept as last_tool");
assert_eq!(last.name, "read");
assert!(
last.args
.iter()
.any(|(k, v)| k == "path" && v == "/tmp/ws/a.rs")
);
{
let guard = AGENT_REGISTRY.tool_started(
&agent_id,
generation,
"edit",
&serde_json::json!({"path": "/tmp/ws/b.rs"}),
);
drop(guard);
}
let handles = AGENT_REGISTRY.list();
let h = handles
.iter()
.find(|h| h.agent_id == agent_id)
.expect("agent registered");
assert_eq!(
h.last_tool.as_ref().map(|t| t.name.as_str()),
Some("edit"),
"newer completed tool replaces the old one"
);
AGENT_REGISTRY.deregister(&agent_id, generation);
}
#[test]
fn cancel_by_parent_key_cancels_matching_agents_only() {
let a1 = format!("cancel_parent_a1_{}", crate::generate_suffix());
let a2 = format!("cancel_parent_a2_{}", crate::generate_suffix());
let b1 = format!("cancel_parent_b1_{}", crate::generate_suffix());
let ws = crate::Workspace {
name: "ws_cancel_parent".to_string(),
path: "/tmp/ws_cancel_parent".to_string(),
..Default::default()
};
let register_with_parent = |agent_id: &str, parent: ParentKey| {
AGENT_REGISTRY.register(
agent_id.to_string(),
"analyst".to_string(),
None,
&ws,
"test".to_string(),
CancellationToken::new(),
Some(parent),
None,
)
};
let gen_a1 = register_with_parent(&a1, ParentKey::Research("runA".to_string()));
let gen_a2 = register_with_parent(&a2, ParentKey::Research("runA".to_string()));
let gen_b = register_with_parent(&b1, ParentKey::Research("runB".to_string()));
AGENT_REGISTRY.cancel_by_parent_key(&ParentKey::Research("runA".to_string()));
assert!(
!AGENT_REGISTRY.contains(&a1) && !AGENT_REGISTRY.contains(&a2),
"runA agents cancelled (entry removed)"
);
assert!(
AGENT_REGISTRY.contains(&b1),
"runB agent untouched by the runA group cancel"
);
let b = AGENT_REGISTRY
.list()
.into_iter()
.find(|h| h.agent_id == b1)
.expect("runB agent still registered");
assert_eq!(
b.parent_key,
Some(ParentKey::Research("runB".to_string())),
"runB agent keeps its parent key"
);
AGENT_REGISTRY.deregister(&b1, gen_b);
let _ = (gen_a1, gen_a2);
}
}