From 9b85a64b581c0ada7a55c5d3b68f381514a9bd07 Mon Sep 17 00:00:00 2001 From: oxoxDev <164490987+oxoxDev@users.noreply.github.com> Date: Mon, 8 Jun 2026 19:56:25 +0530 Subject: [PATCH] feat(agent_orchestration): background command-center run list (#3373) (#3497) --- src/core/all.rs | 8 + .../agent_orchestration/command_center/mod.rs | 23 ++ .../agent_orchestration/command_center/ops.rs | 260 ++++++++++++++++++ .../command_center/schemas.rs | 111 ++++++++ .../command_center/types.rs | 121 ++++++++ src/openhuman/agent_orchestration/mod.rs | 4 + tests/json_rpc_e2e.rs | 92 +++++++ 7 files changed, 619 insertions(+) create mode 100644 src/openhuman/agent_orchestration/command_center/mod.rs create mode 100644 src/openhuman/agent_orchestration/command_center/ops.rs create mode 100644 src/openhuman/agent_orchestration/command_center/schemas.rs create mode 100644 src/openhuman/agent_orchestration/command_center/types.rs diff --git a/src/core/all.rs b/src/core/all.rs index fc90a8a88..1d8399c27 100644 --- a/src/core/all.rs +++ b/src/core/all.rs @@ -293,6 +293,9 @@ fn build_registered_controllers() -> Vec { controllers.extend(crate::openhuman::devices::all_devices_registered_controllers()); // Durable agent session database — queryable index over transcripts, lineage, tool calls controllers.extend(crate::openhuman::session_db::all_session_db_registered_controllers()); + // Background agent command center — read-only grouped view over the run ledger + controllers + .extend(crate::openhuman::agent_orchestration::all_command_center_registered_controllers()); controllers } @@ -419,6 +422,8 @@ fn build_declared_controller_schemas() -> Vec { schemas.extend(crate::openhuman::devices::all_devices_controller_schemas()); // Durable agent session database schemas.extend(crate::openhuman::session_db::all_session_db_controller_schemas()); + // Background agent command center + schemas.extend(crate::openhuman::agent_orchestration::all_command_center_controller_schemas()); schemas } @@ -502,6 +507,9 @@ pub fn namespace_description(namespace: &str) -> Option<&'static str> { "run_ledger" => Some( "Durable agent and workflow run state, child lineage, events, telemetry, and checkpoint references.", ), + "agent_work" => Some( + "Background agent command center — recent agent runs grouped by status (needs-input, working, completed, failed, stopped).", + ), "billing" => Some("Subscription plan, payment links, and credit top-up via the backend."), "team" => Some("Team member management, invites, and role changes via the backend."), "tool_registry" => Some( diff --git a/src/openhuman/agent_orchestration/command_center/mod.rs b/src/openhuman/agent_orchestration/command_center/mod.rs new file mode 100644 index 000000000..9838845d7 --- /dev/null +++ b/src/openhuman/agent_orchestration/command_center/mod.rs @@ -0,0 +1,23 @@ +//! Background agent command center (issue #3373). +//! +//! A read-only product surface over the durable run ledger +//! (`session_db::run_ledger`): it lists recent background agent runs grouped by +//! a normalized status model (needs-input / working / completed / failed / +//! stopped) so users can see what is in flight, what is blocked on them, and +//! what finished. Live run state already persists to the ledger via the spawn +//! tools + `channels::providers::web::progress_bridge`; this module only +//! projects and groups it for display. +//! +//! Control verbs (stop / retry / continue) are intentionally out of scope here +//! and tracked as follow-up work. + +mod ops; +mod schemas; +pub mod types; + +pub use ops::{bucket_for, build_view, list_agent_work}; +pub use schemas::{ + all_controller_schemas as all_command_center_controller_schemas, + all_registered_controllers as all_command_center_registered_controllers, +}; +pub use types::{AgentWorkBucket, AgentWorkRow, CommandCenterGroup, CommandCenterView}; diff --git a/src/openhuman/agent_orchestration/command_center/ops.rs b/src/openhuman/agent_orchestration/command_center/ops.rs new file mode 100644 index 000000000..a9f3fb525 --- /dev/null +++ b/src/openhuman/agent_orchestration/command_center/ops.rs @@ -0,0 +1,260 @@ +//! Read-only command-center projection over the durable run ledger. +//! +//! [`list_agent_work`] fetches recent background agent runs from +//! `session_db::run_ledger` and projects them into a [`CommandCenterView`] +//! grouped by normalized [`AgentWorkBucket`]. The projection is split so the +//! pure grouping logic ([`build_view`]) is unit-testable without a database, +//! while [`list_agent_work`] owns the one ledger read. + +use anyhow::Result; + +use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry; +use crate::openhuman::config::Config; +use crate::openhuman::session_db::run_ledger::{ + list_agent_runs, AgentRun, AgentRunListRequest, AgentRunStatus, +}; + +use super::types::{AgentWorkBucket, AgentWorkRow, CommandCenterGroup, CommandCenterView}; + +/// Default number of recent runs scanned for the command center. +const DEFAULT_LIMIT: u32 = 200; +/// Hard ceiling, mirroring the ledger's own `list_agent_runs` cap. +const MAX_LIMIT: u32 = 500; + +/// Map a fine-grained ledger status to its command-center bucket. +/// +/// Exhaustive on [`AgentRunStatus`] so a new ledger status variant fails to +/// compile here until its bucket is decided. +pub fn bucket_for(status: AgentRunStatus) -> AgentWorkBucket { + match status { + AgentRunStatus::AwaitingUser => AgentWorkBucket::NeedsInput, + AgentRunStatus::Pending | AgentRunStatus::Running | AgentRunStatus::Paused => { + AgentWorkBucket::Working + } + AgentRunStatus::Completed => AgentWorkBucket::Completed, + AgentRunStatus::Failed => AgentWorkBucket::Failed, + AgentRunStatus::Cancelled | AgentRunStatus::Interrupted => AgentWorkBucket::Stopped, + } +} + +/// List recent background agent work, grouped by command-center bucket. +/// +/// Reads at most `limit` (default 200, capped 500) most-recently-updated runs +/// across every parent thread and projects them. Read-only: no ledger writes. +pub fn list_agent_work(config: &Config, limit: Option) -> Result { + let limit = limit.unwrap_or(DEFAULT_LIMIT).min(MAX_LIMIT); + log::debug!( + target: "command_center", + "[command_center] list_agent_work.entry limit={limit}" + ); + let request = AgentRunListRequest { + status: None, + kind: None, + parent_run_id: None, + parent_thread_id: None, + limit: Some(limit), + offset: None, + }; + let response = list_agent_runs(config, &request)?; + let view = build_view(response.runs); + log::debug!( + target: "command_center", + "[command_center] list_agent_work.done total={}", + view.total + ); + Ok(view) +} + +/// Project + group a set of ledger runs into the command-center view. +/// +/// Pure: input order is preserved within each bucket, so callers that pass +/// runs already ordered most-recently-updated-first (as `list_agent_runs` +/// does) get recent-first rows per group. All five buckets are always present. +pub fn build_view(runs: Vec) -> CommandCenterView { + let rows: Vec = runs.into_iter().map(project_row).collect(); + let total = rows.len(); + let groups = AgentWorkBucket::ALL + .iter() + .map(|&bucket| { + let bucket_rows: Vec = rows + .iter() + .filter(|r| r.bucket == bucket) + .cloned() + .collect(); + CommandCenterGroup { + bucket, + count: bucket_rows.len(), + rows: bucket_rows, + } + }) + .collect(); + CommandCenterView { groups, total } +} + +/// Project one ledger run into a lean command-center row. +fn project_row(run: AgentRun) -> AgentWorkRow { + let display_name = run.agent_id.as_deref().and_then(resolve_display_name); + let telemetry = run.telemetry; + AgentWorkRow { + run_id: run.id, + kind: run.kind.as_str().to_string(), + agent_id: run.agent_id, + display_name, + bucket: bucket_for(run.status), + status: run.status.as_str().to_string(), + parent_thread_id: run.parent_thread_id, + worker_thread_id: run.worker_thread_id, + summary: run.summary, + error: run.error, + started_at: run.started_at.to_rfc3339(), + updated_at: run.updated_at.to_rfc3339(), + elapsed_ms: telemetry.as_ref().and_then(|t| t.elapsed_ms), + input_tokens: telemetry.as_ref().map(|t| t.input_tokens).unwrap_or(0), + output_tokens: telemetry.as_ref().map(|t| t.output_tokens).unwrap_or(0), + cost_usd: telemetry.as_ref().map(|t| t.cost_usd).unwrap_or(0.0), + tool_count: telemetry.as_ref().map(|t| t.tool_count).unwrap_or(0), + } +} + +/// Resolve an agent id to its registry display name, if the registry is up and +/// the agent is known. Returns `None` otherwise (e.g. custom/removed agents). +fn resolve_display_name(agent_id: &str) -> Option { + AgentDefinitionRegistry::global() + .and_then(|registry| registry.get(agent_id)) + .map(|definition| definition.display_name().to_string()) +} + +#[cfg(test)] +mod tests { + use super::*; + use chrono::{TimeZone, Utc}; + use serde_json::json; + + fn run_with(id: &str, status: AgentRunStatus, updated_secs: i64) -> AgentRun { + AgentRun { + id: id.to_string(), + kind: crate::openhuman::session_db::run_ledger::AgentRunKind::Subagent, + parent_run_id: None, + parent_thread_id: Some("thread-1".to_string()), + agent_id: Some("researcher".to_string()), + status, + prompt_ref: None, + worker_thread_id: None, + task_board_id: None, + task_card_id: None, + checkpoint_path: None, + checkpoint: None, + summary: None, + error: None, + metadata: json!({}), + telemetry: None, + started_at: Utc.timestamp_opt(1_700_000_000, 0).unwrap(), + updated_at: Utc.timestamp_opt(1_700_000_000 + updated_secs, 0).unwrap(), + completed_at: None, + } + } + + #[test] + fn bucket_for_maps_every_status_to_its_group() { + assert_eq!( + bucket_for(AgentRunStatus::AwaitingUser), + AgentWorkBucket::NeedsInput + ); + assert_eq!( + bucket_for(AgentRunStatus::Pending), + AgentWorkBucket::Working + ); + assert_eq!( + bucket_for(AgentRunStatus::Running), + AgentWorkBucket::Working + ); + assert_eq!(bucket_for(AgentRunStatus::Paused), AgentWorkBucket::Working); + assert_eq!( + bucket_for(AgentRunStatus::Completed), + AgentWorkBucket::Completed + ); + assert_eq!(bucket_for(AgentRunStatus::Failed), AgentWorkBucket::Failed); + assert_eq!( + bucket_for(AgentRunStatus::Cancelled), + AgentWorkBucket::Stopped + ); + assert_eq!( + bucket_for(AgentRunStatus::Interrupted), + AgentWorkBucket::Stopped + ); + } + + #[test] + fn build_view_always_emits_five_buckets_in_display_order() { + let view = build_view(vec![]); + assert_eq!(view.total, 0); + let order: Vec = view.groups.iter().map(|g| g.bucket).collect(); + assert_eq!(order, AgentWorkBucket::ALL.to_vec()); + assert!(view + .groups + .iter() + .all(|g| g.rows.is_empty() && g.count == 0)); + } + + #[test] + fn build_view_groups_runs_into_correct_buckets() { + let runs = vec![ + run_with("a", AgentRunStatus::Running, 1), + run_with("b", AgentRunStatus::AwaitingUser, 2), + run_with("c", AgentRunStatus::Completed, 3), + run_with("d", AgentRunStatus::Failed, 4), + run_with("e", AgentRunStatus::Cancelled, 5), + run_with("f", AgentRunStatus::Pending, 6), + ]; + let view = build_view(runs); + assert_eq!(view.total, 6); + + let group = |bucket: AgentWorkBucket| { + view.groups + .iter() + .find(|g| g.bucket == bucket) + .expect("bucket present") + }; + assert_eq!(group(AgentWorkBucket::NeedsInput).count, 1); + assert_eq!(group(AgentWorkBucket::Working).count, 2); // running + pending + assert_eq!(group(AgentWorkBucket::Completed).count, 1); + assert_eq!(group(AgentWorkBucket::Failed).count, 1); + assert_eq!(group(AgentWorkBucket::Stopped).count, 1); + } + + #[test] + fn build_view_preserves_input_order_within_a_bucket() { + // Caller passes recent-first; projection must not reorder. + let runs = vec![ + run_with("newest", AgentRunStatus::Running, 30), + run_with("middle", AgentRunStatus::Running, 20), + run_with("oldest", AgentRunStatus::Running, 10), + ]; + let view = build_view(runs); + let working = view + .groups + .iter() + .find(|g| g.bucket == AgentWorkBucket::Working) + .unwrap(); + let ids: Vec<&str> = working.rows.iter().map(|r| r.run_id.as_str()).collect(); + assert_eq!(ids, vec!["newest", "middle", "oldest"]); + } + + #[test] + fn project_row_defaults_telemetry_to_zero_when_absent() { + let view = build_view(vec![run_with("x", AgentRunStatus::Completed, 1)]); + let row = view + .groups + .iter() + .flat_map(|g| &g.rows) + .find(|r| r.run_id == "x") + .unwrap(); + assert_eq!(row.input_tokens, 0); + assert_eq!(row.output_tokens, 0); + assert_eq!(row.cost_usd, 0.0); + assert_eq!(row.tool_count, 0); + assert_eq!(row.elapsed_ms, None); + assert_eq!(row.status, "completed"); + assert_eq!(row.kind, "subagent"); + } +} diff --git a/src/openhuman/agent_orchestration/command_center/schemas.rs b/src/openhuman/agent_orchestration/command_center/schemas.rs new file mode 100644 index 000000000..6dd03e3e7 --- /dev/null +++ b/src/openhuman/agent_orchestration/command_center/schemas.rs @@ -0,0 +1,111 @@ +//! Controller schema + JSON-RPC dispatcher for the background agent command +//! center. Exposes `openhuman.agent_work_list` (read-only) over the durable +//! run ledger. Handlers delegate to [`super::ops`]; no business logic here. + +use serde_json::{Map, Value}; + +use crate::core::all::{ControllerFuture, RegisteredController}; +use crate::core::{ControllerSchema, FieldSchema, TypeSchema}; +use crate::openhuman::config::rpc as config_rpc; +use crate::rpc::RpcOutcome; + +/// Controller schemas exposed by the command center. +pub fn all_controller_schemas() -> Vec { + vec![schema_for("agent_work_list")] +} + +/// Registered controllers (schema + handler) for the command center. +pub fn all_registered_controllers() -> Vec { + vec![RegisteredController { + schema: schema_for("agent_work_list"), + handler: handle_agent_work_list, + }] +} + +fn schema_for(function: &str) -> ControllerSchema { + match function { + "agent_work_list" => ControllerSchema { + namespace: "agent_work", + function: "list", + description: "List recent background agent runs grouped by command-center status \ + bucket (needs_input / working / completed / failed / stopped).", + inputs: vec![optional_u64( + "limit", + "Max recent runs to scan (default 200, max 500).", + )], + outputs: vec![json_output( + "result", + "CommandCenterView with five status groups and a total count.", + )], + }, + _ => ControllerSchema { + namespace: "agent_work", + function: "unknown", + description: "unknown command center function", + inputs: vec![], + outputs: vec![], + }, + } +} + +fn handle_agent_work_list(params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "command_center_rpc", "[command_center_rpc][{cid}] list.entry"); + let config = config_rpc::load_config_with_timeout() + .await + .inspect_err(|err| { + log::warn!(target: "command_center_rpc", "[command_center_rpc][{cid}] list.config_failed err={err}"); + })?; + let limit = params + .get("limit") + .and_then(|v| v.as_u64()) + .map(|v| v as u32); + let view = super::ops::list_agent_work(&config, limit).map_err(|e| { + let s = e.to_string(); + log::warn!(target: "command_center_rpc", "[command_center_rpc][{cid}] list.error err={s}"); + s + })?; + to_json(view) + }) +} + +fn to_json(value: T) -> Result { + RpcOutcome::new(value, vec![]).into_cli_compatible_json() +} + +fn new_correlation_id() -> String { + uuid::Uuid::new_v4().simple().to_string()[..8].to_string() +} + +fn optional_u64(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::U64)), + comment, + required: false, + } +} + +fn json_output(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Json, + comment, + required: true, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registered_controllers_match_schemas() { + let schemas = all_controller_schemas(); + let registered = all_registered_controllers(); + assert_eq!(schemas.len(), registered.len()); + assert!(schemas.iter().all(|s| s.namespace == "agent_work")); + assert_eq!(schema_for("agent_work_list").function, "list"); + } +} diff --git a/src/openhuman/agent_orchestration/command_center/types.rs b/src/openhuman/agent_orchestration/command_center/types.rs new file mode 100644 index 000000000..4380f4ae4 --- /dev/null +++ b/src/openhuman/agent_orchestration/command_center/types.rs @@ -0,0 +1,121 @@ +//! Command-center view types for the background agent surface (issue #3373). +//! +//! The durable run ledger (`session_db::run_ledger`) stores fine-grained +//! `AgentRunStatus` values for every background agent run. The background +//! agent command center groups that work into five user-facing buckets so a +//! reviewer can see, at a glance, what needs input, what is still working, and +//! what finished, failed, or was stopped. These types are the read-only +//! projection the command center renders; nothing here mutates ledger state. + +use serde::Serialize; + +/// Normalized command-center status bucket. +/// +/// Collapses the ledger's eight `AgentRunStatus` values into the five groups +/// the command center renders. The mapping lives in +/// [`super::ops::bucket_for`]; the order of [`AgentWorkBucket::ALL`] is the +/// display order (needs-input first so blocked work is most visible). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum AgentWorkBucket { + /// A run paused waiting for user input (`awaiting_user`). + NeedsInput, + /// A run still executing or queued (`pending` / `running` / `paused`). + Working, + /// A run that finished successfully (`completed`). + Completed, + /// A run that ended in error (`failed`). + Failed, + /// A run cancelled or interrupted before completion. + Stopped, +} + +impl AgentWorkBucket { + /// Display order: needs-input first, then in-flight, then terminal states. + pub const ALL: [AgentWorkBucket; 5] = [ + AgentWorkBucket::NeedsInput, + AgentWorkBucket::Working, + AgentWorkBucket::Completed, + AgentWorkBucket::Failed, + AgentWorkBucket::Stopped, + ]; + + /// Stable wire string for this bucket. + pub fn as_str(self) -> &'static str { + match self { + AgentWorkBucket::NeedsInput => "needs_input", + AgentWorkBucket::Working => "working", + AgentWorkBucket::Completed => "completed", + AgentWorkBucket::Failed => "failed", + AgentWorkBucket::Stopped => "stopped", + } + } +} + +/// One command-center row, projected from a durable [`AgentRun`]. +/// +/// Kept deliberately lean — transcripts and checkpoints stay in the ledger / +/// thread stores and are fetched on demand when a user opens a row. +/// +/// [`AgentRun`]: crate::openhuman::session_db::run_ledger::AgentRun +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct AgentWorkRow { + /// Ledger run id (open/peek key). + pub run_id: String, + /// Run kind wire string (`subagent` / `worker_thread` / ...). + pub kind: String, + /// Agent definition id, when known. + pub agent_id: Option, + /// Human-friendly name resolved from the agent registry, when available. + pub display_name: Option, + /// Normalized command-center bucket. + pub bucket: AgentWorkBucket, + /// Raw ledger status wire string (preserved for detail views). + pub status: String, + /// Parent conversation thread, for deterministic "open thread". + pub parent_thread_id: Option, + /// Linked worker thread, for "open worker transcript". + pub worker_thread_id: Option, + /// Latest summary, when the run produced one. + pub summary: Option, + /// Failure reason, when the run failed. + pub error: Option, + /// RFC3339 start timestamp. + pub started_at: String, + /// RFC3339 last-activity timestamp. + pub updated_at: String, + /// Wall-clock elapsed milliseconds, when telemetry recorded it. + pub elapsed_ms: Option, + /// Input tokens spent (0 when no telemetry). + pub input_tokens: u64, + /// Output tokens spent (0 when no telemetry). + pub output_tokens: u64, + /// Cost in USD (0 when no telemetry). + pub cost_usd: f64, + /// Tool-call count (0 when no telemetry). + pub tool_count: u64, +} + +/// One status group in the command-center view. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CommandCenterGroup { + /// The bucket this group represents. + pub bucket: AgentWorkBucket, + /// Number of rows in this group. + pub count: usize, + /// Rows, most-recently-updated first. + pub rows: Vec, +} + +/// The full command-center view: all five buckets in display order plus a total. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CommandCenterView { + /// Always exactly five groups, in [`AgentWorkBucket::ALL`] order (empty + /// groups included so the UI can render stable section headers). + pub groups: Vec, + /// Total rows across all buckets. + pub total: usize, +} diff --git a/src/openhuman/agent_orchestration/mod.rs b/src/openhuman/agent_orchestration/mod.rs index d9627d728..191d32a65 100644 --- a/src/openhuman/agent_orchestration/mod.rs +++ b/src/openhuman/agent_orchestration/mod.rs @@ -5,6 +5,7 @@ //! [`crate::openhuman::agent::harness`] module remains responsible for prompt //! construction, tool filtering, and the actual sub-agent run loop. +pub mod command_center; mod ops; pub mod tools; pub mod types; @@ -12,6 +13,9 @@ pub mod types; #[cfg(test)] mod ops_tests; +pub use command_center::{ + all_command_center_controller_schemas, all_command_center_registered_controllers, +}; pub use ops::{AgentOrchestrationSession, OrchestrationError}; pub use types::{ AgentMessage, AgentOrchestrationEvent, AgentSnapshot, AgentStatus, CloseAgentRequest, diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index 787b26971..40d440a7b 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -2998,6 +2998,98 @@ async fn json_rpc_run_ledger_lifecycle() { rpc_join.abort(); } +#[tokio::test] +async fn json_rpc_agent_work_list_groups_runs_by_bucket() { + let _env_lock = json_rpc_e2e_env_lock(); + let tmp = tempdir().expect("tempdir"); + let home = tmp.path(); + let openhuman_home = home.join(".openhuman"); + + let _home_guard = EnvVarGuard::set_to_path("HOME", home); + let _workspace_guard = EnvVarGuard::unset("OPENHUMAN_WORKSPACE"); + let _backend_url_guard = EnvVarGuard::unset("BACKEND_URL"); + let _vite_backend_url_guard = EnvVarGuard::unset("VITE_BACKEND_URL"); + let _api_url_guard = EnvVarGuard::unset("OPENHUMAN_API_URL"); + + let (api_addr, api_join) = serve_on_ephemeral(mock_upstream_router()).await; + let api_origin = format!("http://{api_addr}"); + write_min_config(openhuman_home.as_path(), &api_origin); + + let (rpc_addr, rpc_join) = serve_on_ephemeral(build_core_http_router(false)).await; + let rpc_base = format!("http://{rpc_addr}"); + + let config = openhuman_core::openhuman::config::Config::load_or_init() + .await + .expect("load config"); + + use openhuman_core::openhuman::session_db::run_ledger::{ + upsert_agent_run, AgentRunKind, AgentRunStatus, AgentRunUpsert, + }; + let seed = |id: &str, status: AgentRunStatus| AgentRunUpsert { + id: id.to_string(), + kind: AgentRunKind::Subagent, + parent_run_id: None, + parent_thread_id: Some("thread-work-1".to_string()), + agent_id: Some("researcher".to_string()), + status, + prompt_ref: None, + worker_thread_id: None, + task_board_id: None, + task_card_id: None, + checkpoint_path: None, + checkpoint: None, + summary: None, + error: None, + metadata: json!({ "source": "json_rpc_e2e" }), + started_at: None, + completed_at: None, + }; + // Two awaiting-user (needs_input), one running (working), one completed. + upsert_agent_run(&config, seed("work-a", AgentRunStatus::AwaitingUser)).expect("seed a"); + upsert_agent_run(&config, seed("work-b", AgentRunStatus::AwaitingUser)).expect("seed b"); + upsert_agent_run(&config, seed("work-c", AgentRunStatus::Running)).expect("seed c"); + upsert_agent_run(&config, seed("work-d", AgentRunStatus::Completed)).expect("seed d"); + + let list = post_json_rpc(&rpc_base, 9131, "openhuman.agent_work_list", json!({})).await; + let outer = assert_no_jsonrpc_error(&list, "agent_work_list"); + + assert_eq!( + outer.get("total").and_then(serde_json::Value::as_u64), + Some(4) + ); + + let groups = outer + .get("groups") + .and_then(serde_json::Value::as_array) + .expect("groups array"); + // Always five buckets, in display order. + let buckets: Vec<&str> = groups + .iter() + .filter_map(|g| g.get("bucket").and_then(serde_json::Value::as_str)) + .collect(); + assert_eq!( + buckets, + vec!["needs_input", "working", "completed", "failed", "stopped"] + ); + + let count_of = |bucket: &str| -> u64 { + groups + .iter() + .find(|g| g.get("bucket").and_then(serde_json::Value::as_str) == Some(bucket)) + .and_then(|g| g.get("count")) + .and_then(serde_json::Value::as_u64) + .unwrap_or(0) + }; + assert_eq!(count_of("needs_input"), 2); + assert_eq!(count_of("working"), 1); + assert_eq!(count_of("completed"), 1); + assert_eq!(count_of("failed"), 0); + assert_eq!(count_of("stopped"), 0); + + api_join.abort(); + rpc_join.abort(); +} + #[tokio::test] async fn json_rpc_task_board_brief_roundtrips_across_todos_and_threads_rpc() { let _env_lock = json_rpc_e2e_env_lock();