diff --git a/app/src/services/api/agentTeamApi.test.ts b/app/src/services/api/agentTeamApi.test.ts index 88dcaac90..ab9eca54b 100644 --- a/app/src/services/api/agentTeamApi.test.ts +++ b/app/src/services/api/agentTeamApi.test.ts @@ -123,3 +123,78 @@ describe('agentTeamApi.listMessages', () => { }); }); }); + +describe('agentTeamApi.completeTask', () => { + it('unwraps { result } and defaults evidence + requireEvidence', async () => { + mockCall.mockResolvedValueOnce({ result: { kind: 'completed', id: 'task-1', status: 'done' } }); + const outcome = await agentTeamApi.completeTask({ + teamId: 'team-1', + taskId: 'task-1', + memberId: 'm1', + }); + expect(outcome.kind).toBe('completed'); + expect(mockCall).toHaveBeenCalledWith({ + method: 'openhuman.agent_team_complete_task', + params: { + teamId: 'team-1', + taskId: 'task-1', + memberId: 'm1', + evidence: [], + requireEvidence: false, + }, + }); + }); + + it('forwards evidence + requireEvidence and surfaces a gateFailed outcome', async () => { + mockCall.mockResolvedValueOnce({ + result: { kind: 'gateFailed', reasons: ['completion requires at least one evidence link'] }, + }); + const outcome = await agentTeamApi.completeTask({ + teamId: 'team-1', + taskId: 'task-1', + memberId: 'm1', + evidence: ['proof'], + requireEvidence: true, + }); + expect(outcome).toEqual({ + kind: 'gateFailed', + reasons: ['completion requires at least one evidence link'], + }); + expect(mockCall).toHaveBeenCalledWith({ + method: 'openhuman.agent_team_complete_task', + params: { + teamId: 'team-1', + taskId: 'task-1', + memberId: 'm1', + evidence: ['proof'], + requireEvidence: true, + }, + }); + }); + + it('throws when a required id is missing, without calling core', async () => { + await expect( + agentTeamApi.completeTask({ teamId: '', taskId: 't', memberId: 'm' }) + ).rejects.toThrow('required'); + expect(mockCall).not.toHaveBeenCalled(); + }); +}); + +describe('agentTeamApi.shutdownMember', () => { + it('unwraps { result } to the MemberShutdown', async () => { + const result = { member: { id: 'm1', memberStatus: 'stopped' }, releasedTaskIds: ['task-1'] }; + mockCall.mockResolvedValueOnce({ result }); + const outcome = await agentTeamApi.shutdownMember('team-1', 'm1'); + expect(outcome).toBe(result); + expect(mockCall).toHaveBeenCalledWith({ + method: 'openhuman.agent_team_shutdown_member', + params: { teamId: 'team-1', memberId: 'm1' }, + }); + }); + + it('throws when teamId or memberId is empty, without calling core', async () => { + await expect(agentTeamApi.shutdownMember('', 'm1')).rejects.toThrow('required'); + await expect(agentTeamApi.shutdownMember('team-1', '')).rejects.toThrow('required'); + expect(mockCall).not.toHaveBeenCalled(); + }); +}); diff --git a/app/src/services/api/agentTeamApi.ts b/app/src/services/api/agentTeamApi.ts index 6fe8aa94b..8ebaf657d 100644 --- a/app/src/services/api/agentTeamApi.ts +++ b/app/src/services/api/agentTeamApi.ts @@ -1,11 +1,13 @@ /** * Frontend client for the durable agent-team coordination surface (#3374). * - * Wraps the read paths of the `openhuman.agent_team_*` controller family that - * landed with the durable team ledger (PR1, #3546): `agent_team_list`, - * `agent_team_get`, and `agent_team_list_messages`. PR2 is a READ-ONLY surface - * — creating teams, assigning/claiming tasks and posting messages is the - * agents' job (and a later PR), so only the three read methods live here. + * Wraps the `openhuman.agent_team_*` controller family from the durable team + * ledger (PR1, #3546): the read paths `agent_team_list`, `agent_team_get`, and + * `agent_team_list_messages`, plus two lifecycle writes added with quality-gated + * completion — `agent_team_complete_task` and `agent_team_shutdown_member`. + * Creating teams, assigning and claiming tasks, and posting messages stay the + * agents' job (driven over the same controllers from the run loop), so those + * write methods are deliberately absent here. * * The Rust controllers serialize their row types with * `#[serde(rename_all = "camelCase")]`, so the wire payload is already camelCase @@ -86,6 +88,24 @@ export interface TeamView { tasks: AgentTeamTask[]; } +/** + * Outcome of a completion attempt. Mirrors Rust `CompletionOutcome`, which is an + * internally-tagged enum on `kind`: `completed` flattens the resulting task onto + * the object; `gateFailed` carries the unmet-invariant reasons; the rest are + * bare tags. + */ +export type CompletionOutcome = + | ({ kind: 'completed' } & AgentTeamTask) + | { kind: 'gateFailed'; reasons: string[] } + | { kind: 'notClaimed' } + | { kind: 'unknownTask' }; + +/** Result of stopping a member. Mirrors Rust `MemberShutdown`. */ +export interface MemberShutdown { + member: AgentTeamMember; + releasedTaskIds: string[]; +} + /** Parsed payload of a `team_message` run event. Mirrors the Rust `json!` body. */ export interface TeamMessagePayload { from: string; @@ -194,4 +214,48 @@ export const agentTeamApi = { log('listMessages received=%d', messages.length); return messages; }, + + /** + * Complete a claimed task, gating its transition to `done`. The core checks + * the quality invariants (dependencies done, the completer is the claimant and + * any pre-assigned owner, evidence present when `requireEvidence`) and returns + * a {@link CompletionOutcome}: `completed`, `gateFailed` (with reasons), + * `notClaimed`, or `unknownTask`. Evidence links accumulate across retries. + */ + completeTask: async (params: { + teamId: string; + taskId: string; + memberId: string; + evidence?: string[]; + requireEvidence?: boolean; + }): Promise => { + const { teamId, taskId, memberId, evidence = [], requireEvidence = false } = params; + if (!teamId || !taskId || !memberId) { + throw new Error('agentTeamApi.completeTask: teamId, taskId and memberId are required'); + } + log('completeTask teamId=%s taskId=%s requireEvidence=%o', teamId, taskId, requireEvidence); + const response = await callCoreRpc<{ result: CompletionOutcome }>({ + method: 'openhuman.agent_team_complete_task', + params: { teamId, taskId, memberId, evidence, requireEvidence }, + }); + log('completeTask kind=%s', response.result.kind); + return response.result; + }, + + /** + * Stop a member and release any task it is actively working on back to `todo`. + * Returns the stopped member plus the ids that were released. + */ + shutdownMember: async (teamId: string, memberId: string): Promise => { + if (!teamId || !memberId) { + throw new Error('agentTeamApi.shutdownMember: teamId and memberId are required'); + } + log('shutdownMember teamId=%s memberId=%s', teamId, memberId); + const response = await callCoreRpc<{ result: MemberShutdown }>({ + method: 'openhuman.agent_team_shutdown_member', + params: { teamId, memberId }, + }); + log('shutdownMember released=%d', response.result.releasedTaskIds.length); + return response.result; + }, }; diff --git a/src/openhuman/agent_orchestration/agent_teams/mod.rs b/src/openhuman/agent_orchestration/agent_teams/mod.rs index dfe2a0629..5ef5fca43 100644 --- a/src/openhuman/agent_orchestration/agent_teams/mod.rs +++ b/src/openhuman/agent_orchestration/agent_teams/mod.rs @@ -8,11 +8,13 @@ //! never in the main chat context — so a coordination session can be listed, //! inspected, and resumed. //! -//! PR1 scope (this module today): the durable model + 8 read/write controllers +//! Scope (this module today): the durable model + 10 read/write controllers //! (`create`, `list`, `get`, `assign_task`, `claim_task`, `message_member`, -//! `list_messages`, `close`), the atomic compare-and-swap claim primitive, and -//! dependency validation (self / unknown / cycle). Live agent execution -//! (spawning workers, driving the run loop) and the UI are follow-up PRs. +//! `list_messages`, `complete_task`, `shutdown_member`, `close`), the atomic +//! compare-and-swap claim primitive, dependency validation (self / unknown / +//! cycle), and quality-gated task completion (dependencies done, claimant owns +//! the task, evidence present when required). Live agent execution (spawning +//! workers, driving the run loop) and the message-send UI are a follow-up PR. //! //! Namespace note: `agent_team` is distinct from the existing `team` domain, //! which manages backend org/team membership. @@ -22,11 +24,11 @@ mod schemas; pub mod types; pub use ops::{ - assign_task, claim_task, close_team, create_team, get_team, list_messages, list_teams, - message_member, NewMember, + assign_task, claim_task, close_team, complete_task, create_team, get_team, list_messages, + list_teams, message_member, shutdown_member, NewMember, }; pub use schemas::{ all_controller_schemas as all_agent_team_controller_schemas, all_registered_controllers as all_agent_team_registered_controllers, }; -pub use types::{TeamError, TeamView}; +pub use types::{MemberShutdown, TeamError, TeamView}; diff --git a/src/openhuman/agent_orchestration/agent_teams/ops.rs b/src/openhuman/agent_orchestration/agent_teams/ops.rs index 7a653241d..526e225a2 100644 --- a/src/openhuman/agent_orchestration/agent_teams/ops.rs +++ b/src/openhuman/agent_orchestration/agent_teams/ops.rs @@ -17,11 +17,11 @@ use crate::openhuman::config::Config; use crate::openhuman::session_db::run_ledger::{ self, AgentTeam, AgentTeamListRequest, AgentTeamListResponse, AgentTeamMemberStatus, AgentTeamMemberUpsert, AgentTeamStatus, AgentTeamTask, AgentTeamTaskStatus, - AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, RunEvent, RunEventAppend, - RunEventListRequest, + AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, CompletionOutcome, RunEvent, + RunEventAppend, RunEventListRequest, }; -use super::types::{TeamError, TeamView}; +use super::types::{MemberShutdown, TeamError, TeamView}; const LOG_PREFIX: &str = "[agent_team]"; const TEAM_MESSAGE_EVENT: &str = "team_message"; @@ -287,6 +287,63 @@ pub fn close_team(config: &Config, team_id: &str, summary: Option<&str>) -> Resu Ok(team) } +/// Complete a claimed task, gating its transition to `done`. +/// +/// Validates the completing member belongs to the team, then delegates to the +/// run-ledger completion CAS, which enforces the quality gate (dependencies +/// done, claimant owns the task, evidence present when `require_evidence`) and +/// only flips the task to `done` when every invariant holds. +pub fn complete_task( + config: &Config, + team_id: &str, + task_id: &str, + member_id: &str, + evidence: &[String], + require_evidence: bool, +) -> Result { + log::debug!( + "{LOG_PREFIX} complete_task.entry team={team_id} task={task_id} member={member_id}" + ); + let members = run_ledger::list_agent_team_members(config, team_id)?; + if !members.iter().any(|m| m.id == member_id) { + return Err(anyhow!(TeamError::UnknownMember { + member_id: member_id.to_string(), + })); + } + let outcome = run_ledger::complete_agent_team_task( + config, + team_id, + task_id, + member_id, + evidence, + require_evidence, + )?; + log::debug!("{LOG_PREFIX} complete_task.exit team={team_id} task={task_id}"); + Ok(outcome) +} + +/// Stop a team member, releasing any task it was actively working on. +/// +/// Unknown member ids surface as [`TeamError::UnknownMember`]; otherwise returns +/// the stopped member plus the ids of tasks released back to `todo`. +pub fn shutdown_member(config: &Config, team_id: &str, member_id: &str) -> Result { + log::debug!("{LOG_PREFIX} shutdown_member.entry team={team_id} member={member_id}"); + let (member, released_task_ids) = + run_ledger::shutdown_agent_team_member(config, team_id, member_id)?.ok_or_else(|| { + anyhow!(TeamError::UnknownMember { + member_id: member_id.to_string(), + }) + })?; + log::debug!( + "{LOG_PREFIX} shutdown_member.exit team={team_id} member={member_id} released={}", + released_task_ids.len() + ); + Ok(MemberShutdown { + member, + released_task_ids, + }) +} + fn team_view(config: &Config, team_id: &str) -> Result { let team = run_ledger::get_agent_team(config, team_id)? .ok_or_else(|| anyhow!("team missing after creation: {team_id}"))?; @@ -524,4 +581,229 @@ mod tests { assert_eq!(messages[0].payload["content"], "first"); assert_eq!(messages[1].payload["content"], "second"); } + + /// Create a single-member team and return `(team_id, member_id)`. + fn solo_team(config: &Config, name: &str) -> (String, String) { + let view = create_team( + config, + "lead", + None, + None, + &[NewMember { + name: name.into(), + agent_id: None, + }], + ) + .unwrap(); + let member_id = view.members[0].id.clone(); + (view.team.id, member_id) + } + + #[test] + fn complete_task_gate_passes_and_marks_done() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + let (team_id, alice) = solo_team(&config, "alice"); + + let task = assign_task(&config, &team_id, "ship it", None, None, &[]).unwrap(); + let claim = claim_task(&config, &team_id, &task.id, &alice, "tok-1").unwrap(); + assert!(matches!(claim, ClaimOutcome::Claimed(_))); + + let outcome = complete_task( + &config, + &team_id, + &task.id, + &alice, + &["https://ci/run/1".to_string()], + true, + ) + .unwrap(); + match outcome { + CompletionOutcome::Completed(done) => { + assert_eq!(done.status, AgentTeamTaskStatus::Done); + assert_eq!(done.gate_status, "passed"); + assert_eq!(done.gate_reason, None); + assert_eq!(done.evidence, vec!["https://ci/run/1".to_string()]); + } + other => panic!("expected Completed, got {other:?}"), + } + } + + #[test] + fn complete_task_requires_evidence_then_recovers() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + let (team_id, alice) = solo_team(&config, "alice"); + + let task = assign_task(&config, &team_id, "ship it", None, None, &[]).unwrap(); + claim_task(&config, &team_id, &task.id, &alice, "tok-1").unwrap(); + + // No evidence + require_evidence → gate fails, task stays in progress. + let failed = complete_task(&config, &team_id, &task.id, &alice, &[], true).unwrap(); + match failed { + CompletionOutcome::GateFailed { reasons } => { + assert!( + reasons.iter().any(|r| r.contains("evidence")), + "{reasons:?}" + ); + } + other => panic!("expected GateFailed, got {other:?}"), + } + let mid = run_ledger::get_agent_team_task(&config, &task.id) + .unwrap() + .unwrap(); + assert_eq!(mid.status, AgentTeamTaskStatus::InProgress); + assert_eq!(mid.gate_status, "failed"); + + // Retry with evidence → passes. + let ok = complete_task( + &config, + &team_id, + &task.id, + &alice, + &["proof".to_string()], + true, + ) + .unwrap(); + assert!(matches!(ok, CompletionOutcome::Completed(_))); + } + + #[test] + fn complete_task_is_not_double_completable() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + let (team_id, alice) = solo_team(&config, "alice"); + + let task = assign_task(&config, &team_id, "ship it", None, None, &[]).unwrap(); + claim_task(&config, &team_id, &task.id, &alice, "tok-1").unwrap(); + + let first = complete_task(&config, &team_id, &task.id, &alice, &[], false).unwrap(); + assert!(matches!(first, CompletionOutcome::Completed(_))); + + // A task that is already `done` is no longer in progress, so a second + // completion is rejected (the `status = 'in_progress'` UPDATE guard makes + // the CAS airtight even under a concurrent double-complete). + let second = complete_task(&config, &team_id, &task.id, &alice, &[], false).unwrap(); + assert_eq!(second, CompletionOutcome::NotClaimed); + } + + #[test] + fn complete_task_rejects_non_claimant() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + let view = create_team( + &config, + "lead", + None, + None, + &[ + NewMember { + name: "alice".into(), + agent_id: None, + }, + NewMember { + name: "bob".into(), + agent_id: None, + }, + ], + ) + .unwrap(); + let team_id = view.team.id.clone(); + let alice = view.members[0].id.clone(); + let bob = view.members[1].id.clone(); + + let task = assign_task(&config, &team_id, "ship it", None, None, &[]).unwrap(); + claim_task(&config, &team_id, &task.id, &alice, "tok-1").unwrap(); + + // Bob is a member but not the claimant → NotClaimed. + let outcome = complete_task(&config, &team_id, &task.id, &bob, &[], false).unwrap(); + assert_eq!(outcome, CompletionOutcome::NotClaimed); + + // Unknown member → typed error (not an outcome). + let err = complete_task(&config, &team_id, &task.id, "ghost", &[], false).unwrap_err(); + assert_eq!( + team_err(err), + TeamError::UnknownMember { + member_id: "ghost".into() + } + ); + } + + #[test] + fn complete_task_owner_mismatch_fails_gate() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + let view = create_team( + &config, + "lead", + None, + None, + &[ + NewMember { + name: "alice".into(), + agent_id: None, + }, + NewMember { + name: "bob".into(), + agent_id: None, + }, + ], + ) + .unwrap(); + let team_id = view.team.id.clone(); + let alice = view.members[0].id.clone(); + let bob = view.members[1].id.clone(); + + // Task owned by bob, but alice claims + tries to complete. + let task = assign_task(&config, &team_id, "ship it", None, Some(&bob), &[]).unwrap(); + claim_task(&config, &team_id, &task.id, &alice, "tok-1").unwrap(); + + let outcome = complete_task(&config, &team_id, &task.id, &alice, &[], false).unwrap(); + match outcome { + CompletionOutcome::GateFailed { reasons } => { + assert!( + reasons.iter().any(|r| r.contains("owned by")), + "{reasons:?}" + ); + } + other => panic!("expected GateFailed, got {other:?}"), + } + } + + #[test] + fn shutdown_member_releases_in_progress_tasks() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + let (team_id, alice) = solo_team(&config, "alice"); + + let task = assign_task(&config, &team_id, "ship it", None, None, &[]).unwrap(); + claim_task(&config, &team_id, &task.id, &alice, "tok-1").unwrap(); + + let result = shutdown_member(&config, &team_id, &alice).unwrap(); + assert_eq!(result.released_task_ids, vec![task.id.clone()]); + assert_eq!(result.member.member_status, AgentTeamMemberStatus::Stopped); + + // Task is back to todo and unclaimed → another teammate could claim it. + let released = run_ledger::get_agent_team_task(&config, &task.id) + .unwrap() + .unwrap(); + assert_eq!(released.status, AgentTeamTaskStatus::Todo); + assert_eq!(released.claimed_by_member_id, None); + assert_eq!(released.claim_token, None); + } + + #[test] + fn shutdown_member_unknown_errors() { + let dir = TempDir::new().unwrap(); + let config = test_config(&dir); + let (team_id, _alice) = solo_team(&config, "alice"); + + let err = shutdown_member(&config, &team_id, "ghost").unwrap_err(); + assert_eq!( + team_err(err), + TeamError::UnknownMember { + member_id: "ghost".into() + } + ); + } } diff --git a/src/openhuman/agent_orchestration/agent_teams/schemas.rs b/src/openhuman/agent_orchestration/agent_teams/schemas.rs index 9fa6fafb4..412f8ddcb 100644 --- a/src/openhuman/agent_orchestration/agent_teams/schemas.rs +++ b/src/openhuman/agent_orchestration/agent_teams/schemas.rs @@ -1,9 +1,10 @@ //! Controller schemas + JSON-RPC dispatchers for durable agent-team //! coordination (#3374). Namespace `agent_team`. //! -//! Surface (PR1): create a team with members, list/get teams, assign -//! dependency-aware tasks, atomically claim a task, exchange teammate messages, -//! list those messages, and close a team. Live agent execution is a follow-up. +//! Surface: create a team with members, list/get teams, assign dependency-aware +//! tasks, atomically claim a task, exchange + list teammate messages, complete a +//! claimed task behind a quality gate, shut a member down (releasing its tasks), +//! and close a team. Live agent execution is a follow-up. use serde_json::{Map, Value}; @@ -25,6 +26,8 @@ pub fn all_controller_schemas() -> Vec { schema_for("agent_team_claim_task"), schema_for("agent_team_message_member"), schema_for("agent_team_list_messages"), + schema_for("agent_team_complete_task"), + schema_for("agent_team_shutdown_member"), schema_for("agent_team_close"), ] } @@ -60,6 +63,14 @@ pub fn all_registered_controllers() -> Vec { schema: schema_for("agent_team_list_messages"), handler: handle_list_messages, }, + RegisteredController { + schema: schema_for("agent_team_complete_task"), + handler: handle_complete_task, + }, + RegisteredController { + schema: schema_for("agent_team_shutdown_member"), + handler: handle_shutdown_member, + }, RegisteredController { schema: schema_for("agent_team_close"), handler: handle_close, @@ -157,6 +168,45 @@ fn schema_for(function: &str) -> ControllerSchema { ], outputs: vec![json_output("messages", "Array of message events.")], }, + "agent_team_complete_task" => ControllerSchema { + namespace: "agent_team", + function: "complete_task", + description: + "Complete a claimed task, gating its transition to done behind quality checks.", + inputs: vec![ + required_str("teamId", "Team id."), + required_str("taskId", "Task id to complete."), + required_str( + "memberId", + "Member completing the task (must be the claimant).", + ), + json_input( + "evidence", + "Array of evidence links to attach on completion.", + ), + optional_bool( + "requireEvidence", + "Require at least one evidence link to pass the gate (default false).", + ), + ], + outputs: vec![json_output( + "result", + "CompletionOutcome: completed | gateFailed | notClaimed | unknownTask.", + )], + }, + "agent_team_shutdown_member" => ControllerSchema { + namespace: "agent_team", + function: "shutdown_member", + description: "Stop a team member and release any task it is actively working on.", + inputs: vec![ + required_str("teamId", "Team id."), + required_str("memberId", "Member id to stop."), + ], + outputs: vec![json_output( + "result", + "MemberShutdown: the stopped member and releasedTaskIds.", + )], + }, "agent_team_close" => ControllerSchema { namespace: "agent_team", function: "close", @@ -324,6 +374,52 @@ fn handle_list_messages(params: Map) -> ControllerFuture { }) } +fn handle_complete_task(params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] complete_task.entry"); + let config = config_rpc::load_config_with_timeout().await.inspect_err(|err| { + log::warn!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] complete_task.config_failed err={err}"); + })?; + let team_id = require_str(¶ms, "teamId")?; + let task_id = require_str(¶ms, "taskId")?; + let member_id = require_str(¶ms, "memberId")?; + let evidence = opt_str_array(¶ms, "evidence"); + let require_evidence = params + .get("requireEvidence") + .and_then(Value::as_bool) + .unwrap_or(false); + log::debug!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] complete_task.parsed team={team_id} task={task_id} evidence={} requireEvidence={require_evidence}", evidence.len()); + let outcome = ops::complete_task( + &config, + &team_id, + &task_id, + &member_id, + &evidence, + require_evidence, + ) + .map_err(|e| log_err(&cid, "complete_task", e))?; + log::debug!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] complete_task.success team={team_id} task={task_id}"); + to_json(serde_json::json!({ "result": outcome })) + }) +} + +fn handle_shutdown_member(params: Map) -> ControllerFuture { + Box::pin(async move { + let cid = new_correlation_id(); + log::debug!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] shutdown_member.entry"); + let config = config_rpc::load_config_with_timeout().await.inspect_err(|err| { + log::warn!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] shutdown_member.config_failed err={err}"); + })?; + let team_id = require_str(¶ms, "teamId")?; + let member_id = require_str(¶ms, "memberId")?; + let result = ops::shutdown_member(&config, &team_id, &member_id) + .map_err(|e| log_err(&cid, "shutdown_member", e))?; + log::debug!(target: "agent_team_rpc", "[agent_team_rpc][{cid}] shutdown_member.success team={team_id} member={member_id} released={}", result.released_task_ids.len()); + to_json(serde_json::json!({ "result": result })) + }) +} + fn handle_close(params: Map) -> ControllerFuture { Box::pin(async move { let cid = new_correlation_id(); @@ -438,6 +534,15 @@ fn optional_u64(name: &'static str, comment: &'static str) -> FieldSchema { } } +fn optional_bool(name: &'static str, comment: &'static str) -> FieldSchema { + FieldSchema { + name, + ty: TypeSchema::Option(Box::new(TypeSchema::Bool)), + comment, + required: false, + } +} + fn json_input(name: &'static str, comment: &'static str) -> FieldSchema { FieldSchema { name, @@ -465,8 +570,16 @@ mod tests { let schemas = all_controller_schemas(); let registered = all_registered_controllers(); assert_eq!(schemas.len(), registered.len()); - assert_eq!(schemas.len(), 8); + assert_eq!(schemas.len(), 10); assert!(schemas.iter().all(|s| s.namespace == "agent_team")); assert_eq!(schema_for("agent_team_claim_task").function, "claim_task"); + assert_eq!( + schema_for("agent_team_complete_task").function, + "complete_task" + ); + assert_eq!( + schema_for("agent_team_shutdown_member").function, + "shutdown_member" + ); } } diff --git a/src/openhuman/agent_orchestration/agent_teams/types.rs b/src/openhuman/agent_orchestration/agent_teams/types.rs index e750a16ac..5b697c795 100644 --- a/src/openhuman/agent_orchestration/agent_teams/types.rs +++ b/src/openhuman/agent_orchestration/agent_teams/types.rs @@ -18,6 +18,15 @@ pub struct TeamView { pub tasks: Vec, } +/// Result of shutting a member down: the stopped member plus the ids of any +/// `in_progress` tasks that were released back to `todo` for another teammate. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct MemberShutdown { + pub member: AgentTeamMember, + pub released_task_ids: Vec, +} + /// A validation / coordination problem surfaced by the agent-team ops. #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(rename_all = "camelCase", tag = "kind", content = "detail")] diff --git a/src/openhuman/session_db/run_ledger/mod.rs b/src/openhuman/session_db/run_ledger/mod.rs index a65cb60d6..f11034ce9 100644 --- a/src/openhuman/session_db/run_ledger/mod.rs +++ b/src/openhuman/session_db/run_ledger/mod.rs @@ -10,18 +10,18 @@ pub mod store; pub mod types; pub use ops::{ - append_run_event, claim_agent_team_task, get_agent_run, get_agent_team, get_agent_team_member, - get_agent_team_task, get_workflow_run, list_agent_runs, list_agent_team_members, - list_agent_team_tasks, list_agent_teams, list_recent_run_events, list_workflow_runs, - upsert_agent_run, upsert_agent_team, upsert_agent_team_member, upsert_agent_team_task, - upsert_run_telemetry, upsert_workflow_run, + append_run_event, claim_agent_team_task, complete_agent_team_task, get_agent_run, + get_agent_team, get_agent_team_member, get_agent_team_task, get_workflow_run, list_agent_runs, + list_agent_team_members, list_agent_team_tasks, list_agent_teams, list_recent_run_events, + list_workflow_runs, shutdown_agent_team_member, upsert_agent_run, upsert_agent_team, + upsert_agent_team_member, upsert_agent_team_task, upsert_run_telemetry, upsert_workflow_run, }; pub use types::{ AgentRun, AgentRunKind, AgentRunListRequest, AgentRunListResponse, AgentRunStatus, AgentRunUpsert, AgentTeam, AgentTeamListRequest, AgentTeamListResponse, AgentTeamMember, AgentTeamMemberStatus, AgentTeamMemberUpsert, AgentTeamStatus, AgentTeamTask, - AgentTeamTaskStatus, AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, RunEvent, - RunEventAppend, RunEventListRequest, RunEventListResponse, RunTelemetry, RunTelemetryUpsert, - WorkflowRun, WorkflowRunListRequest, WorkflowRunListResponse, WorkflowRunStatus, - WorkflowRunUpsert, + AgentTeamTaskStatus, AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, CompletionOutcome, + RunEvent, RunEventAppend, RunEventListRequest, RunEventListResponse, RunTelemetry, + RunTelemetryUpsert, WorkflowRun, WorkflowRunListRequest, WorkflowRunListResponse, + WorkflowRunStatus, WorkflowRunUpsert, }; diff --git a/src/openhuman/session_db/run_ledger/ops.rs b/src/openhuman/session_db/run_ledger/ops.rs index 675d86f1a..c8719af66 100644 --- a/src/openhuman/session_db/run_ledger/ops.rs +++ b/src/openhuman/session_db/run_ledger/ops.rs @@ -10,9 +10,9 @@ use super::types::{ AgentRun, AgentRunListRequest, AgentRunListResponse, AgentRunStatus, AgentRunUpsert, AgentTeam, AgentTeamListRequest, AgentTeamListResponse, AgentTeamMember, AgentTeamMemberStatus, AgentTeamMemberUpsert, AgentTeamStatus, AgentTeamTask, AgentTeamTaskStatus, - AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, RunEvent, RunEventAppend, - RunEventListRequest, RunEventListResponse, RunTelemetry, RunTelemetryUpsert, WorkflowRun, - WorkflowRunListRequest, WorkflowRunListResponse, WorkflowRunUpsert, + AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, CompletionOutcome, RunEvent, + RunEventAppend, RunEventListRequest, RunEventListResponse, RunTelemetry, RunTelemetryUpsert, + WorkflowRun, WorkflowRunListRequest, WorkflowRunListResponse, WorkflowRunUpsert, }; const LOG_PREFIX: &str = "[session_db:run_ledger]"; @@ -857,6 +857,238 @@ pub fn claim_agent_team_task( Ok(outcome) } +/// Quality-gate a task's completion and, on pass, transition it to `done`. +/// +/// Runs inside a single transaction so the gate evaluation and the status flip +/// observe one consistent snapshot: +/// 1. Resolve the task by `(id, team_id)`; absent → [`CompletionOutcome::UnknownTask`]. +/// 2. The completer must be the current claimant and the task must be +/// `in_progress`; otherwise [`CompletionOutcome::NotClaimed`]. +/// 3. Evaluate the quality gate (every dependency `done`, claimant matches any +/// pre-assigned owner, evidence present when `require_evidence`). Any unmet +/// invariant records `gate_status = "failed"` + the joined reasons and leaves +/// the task `in_progress` → [`CompletionOutcome::GateFailed`]. +/// 4. On pass, merge `evidence`, set `status = "done"`, `gate_status = "passed"`, +/// clear `gate_reason`, re-fetch → [`CompletionOutcome::Completed`]. +pub fn complete_agent_team_task( + config: &Config, + team_id: &str, + task_id: &str, + member_id: &str, + evidence: &[String], + require_evidence: bool, +) -> Result { + log::debug!( + "{LOG_PREFIX} complete_agent_team_task.entry team={team_id} task={task_id} member={member_id}" + ); + let outcome = crate::openhuman::session_db::store::with_connection(config, |conn| { + init_run_ledger_schema(conn)?; + + // 1. Resolve the task within this team. + let task = match get_agent_team_task_inner(conn, task_id)? { + Some(task) if task.team_id == team_id => task, + _ => { + log::debug!( + "{LOG_PREFIX} complete_agent_team_task.unknown team={team_id} task={task_id}" + ); + return Ok(CompletionOutcome::UnknownTask); + } + }; + + // 2. Only the current claimant may complete, and only while in progress. + let is_claimant = task.claimed_by_member_id.as_deref() == Some(member_id); + let in_progress = task.status == AgentTeamTaskStatus::InProgress; + if !is_claimant || !in_progress { + log::debug!( + "{LOG_PREFIX} complete_agent_team_task.not_claimed team={team_id} task={task_id} claimant={is_claimant} in_progress={in_progress}" + ); + return Ok(CompletionOutcome::NotClaimed); + } + + // Merge prior evidence with the newly-supplied links (de-duplicated, + // order-preserving) so a retry that adds evidence accumulates it. + let mut merged_evidence = task.evidence.clone(); + for link in evidence { + if !merged_evidence.iter().any(|e| e == link) { + merged_evidence.push(link.clone()); + } + } + + // 3. Quality gate. + let reasons = + evaluate_completion_gate(conn, team_id, &task, &merged_evidence, require_evidence)?; + let now = Utc::now(); + if !reasons.is_empty() { + let joined = reasons.join("; "); + conn.execute( + "UPDATE agent_team_tasks + SET gate_status = 'failed', gate_reason = ?1, updated_at = ?2 + WHERE id = ?3 AND team_id = ?4", + params![joined, now.to_rfc3339(), task_id, team_id], + ) + .context("record failed completion gate")?; + log::debug!( + "{LOG_PREFIX} complete_agent_team_task.gate_failed team={team_id} task={task_id} reasons={}", + reasons.len() + ); + return Ok(CompletionOutcome::GateFailed { reasons }); + } + + // 4. Gate passed — flip to done. The WHERE clause is the real CAS: the + // `claimed_by_member_id` guard stops a concurrent shutdown/unclaim from + // completing a task it no longer holds, and the `status = 'in_progress'` + // guard stops a concurrent double-complete by the same member (the + // snapshot check above is a read, not part of the swap — only one of two + // racing UPDATEs flips `in_progress -> done`). + let evidence_json = + serde_json::to_string(&merged_evidence).context("serialize completion evidence")?; + let rows_affected = conn + .execute( + "UPDATE agent_team_tasks + SET status = 'done', gate_status = 'passed', gate_reason = NULL, + evidence_json = ?1, updated_at = ?2 + WHERE id = ?3 AND team_id = ?4 AND claimed_by_member_id = ?5 + AND status = 'in_progress'", + params![evidence_json, now.to_rfc3339(), task_id, team_id, member_id], + ) + .context("complete agent team task")?; + if rows_affected == 0 { + log::debug!( + "{LOG_PREFIX} complete_agent_team_task.lost_claim team={team_id} task={task_id}" + ); + return Ok(CompletionOutcome::NotClaimed); + } + + let done = get_agent_team_task_inner(conn, task_id)? + .context("completed task missing after update")?; + Ok(CompletionOutcome::Completed(Box::new(done))) + })?; + log::debug!( + "{LOG_PREFIX} complete_agent_team_task.exit team={team_id} task={task_id} outcome={}", + match &outcome { + CompletionOutcome::Completed(_) => "completed", + CompletionOutcome::GateFailed { .. } => "gate_failed", + CompletionOutcome::NotClaimed => "not_claimed", + CompletionOutcome::UnknownTask => "unknown", + } + ); + Ok(outcome) +} + +/// Evaluate the quality-gate invariants for a completing task. Returns one +/// human-readable reason per unmet invariant (empty = gate passes). +fn evaluate_completion_gate( + conn: &Connection, + team_id: &str, + task: &AgentTeamTask, + merged_evidence: &[String], + require_evidence: bool, +) -> Result> { + let mut reasons = Vec::new(); + + // Every dependency must still be `done` (defends against a dependency that + // regressed after this task was claimed). + for dep_id in &task.depends_on { + let dep_status: Option = conn + .query_row( + "SELECT status FROM agent_team_tasks WHERE id = ?1 AND team_id = ?2", + params![dep_id, team_id], + |row| row.get(0), + ) + .optional()?; + if dep_status.as_deref() != Some(AgentTeamTaskStatus::Done.as_str()) { + reasons.push(format!("dependency {dep_id} is not done")); + } + } + + // No overlapping ownership: a pre-assigned owner must be the one completing. + if let Some(owner) = &task.owner_member_id { + if Some(owner.as_str()) != task.claimed_by_member_id.as_deref() { + reasons.push(format!( + "task is owned by {owner} but claimed by {}", + task.claimed_by_member_id.as_deref().unwrap_or("nobody") + )); + } + } + + // Evidence gate. + if require_evidence && merged_evidence.is_empty() { + reasons.push("completion requires at least one evidence link".to_string()); + } + + Ok(reasons) +} + +/// Stop a team member and release any task it is actively working on. +/// +/// In one transaction: unclaim the member's `in_progress` tasks back to `todo` +/// (clearing claimant + token so another teammate can pick them up), then mark +/// the member `stopped` and clear its `current_task_id`. Returns the updated +/// member plus the ids of the tasks that were released, or `None` if the member +/// is not part of the team. +pub fn shutdown_agent_team_member( + config: &Config, + team_id: &str, + member_id: &str, +) -> Result)>> { + log::debug!("{LOG_PREFIX} shutdown_agent_team_member.entry team={team_id} member={member_id}"); + let result = crate::openhuman::session_db::store::with_connection(config, |conn| { + init_run_ledger_schema(conn)?; + + // Existence + team-membership check only; the row is intentionally not + // reused — the caller-facing member is re-read after the UPDATEs below so + // it reflects the stopped state. + match get_agent_team_member_inner(conn, member_id)? { + Some(found) if found.team_id == team_id => {} + _ => { + log::debug!( + "{LOG_PREFIX} shutdown_agent_team_member.unknown team={team_id} member={member_id}" + ); + return Ok(None); + } + } + + // Collect the ids first so the caller can report exactly what was freed. + let released: Vec = { + let mut stmt = conn.prepare( + "SELECT id FROM agent_team_tasks + WHERE team_id = ?1 AND claimed_by_member_id = ?2 AND status = 'in_progress'", + )?; + let ids = stmt.query_map(params![team_id, member_id], |row| row.get::<_, String>(0))?; + let mut out = Vec::new(); + for id in ids { + out.push(id?); + } + out + }; + + let now = Utc::now(); + conn.execute( + "UPDATE agent_team_tasks + SET claimed_by_member_id = NULL, claim_token = NULL, status = 'todo', updated_at = ?1 + WHERE team_id = ?2 AND claimed_by_member_id = ?3 AND status = 'in_progress'", + params![now.to_rfc3339(), team_id, member_id], + ) + .context("release tasks on member shutdown")?; + conn.execute( + "UPDATE agent_team_members + SET member_status = 'stopped', current_task_id = NULL, updated_at = ?1 + WHERE id = ?2 AND team_id = ?3", + params![now.to_rfc3339(), member_id, team_id], + ) + .context("stop agent team member")?; + + let member = get_agent_team_member_inner(conn, member_id)? + .context("member missing after shutdown")?; + Ok(Some((member, released))) + })?; + log::debug!( + "{LOG_PREFIX} shutdown_agent_team_member.exit team={team_id} member={member_id} released={}", + result.as_ref().map(|(_, r)| r.len()).unwrap_or(0) + ); + Ok(result) +} + fn get_agent_team_inner(conn: &Connection, id: &str) -> Result> { let mut stmt = conn.prepare( "SELECT id, parent_thread_id, lead_agent_id, status, summary, diff --git a/src/openhuman/session_db/run_ledger/types.rs b/src/openhuman/session_db/run_ledger/types.rs index 8b73375b3..1f093e0b2 100644 --- a/src/openhuman/session_db/run_ledger/types.rs +++ b/src/openhuman/session_db/run_ledger/types.rs @@ -524,3 +524,24 @@ pub enum ClaimOutcome { /// No task matched the given team + task id. UnknownTask, } + +/// Outcome of a completion attempt on a team task. +/// +/// Completion gates a task's transition to `done` behind quality invariants +/// (dependencies done, claimer owns the task, evidence present when required). +/// A failed gate leaves the task `in_progress` with `gate_status = "failed"` +/// and the reasons recorded, so a teammate can fix and retry. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(rename_all = "camelCase", tag = "kind")] +pub enum CompletionOutcome { + /// The task passed its quality gate and is now `done`. Boxed to keep the + /// enum small (the task payload dwarfs the other variants). + Completed(Box), + /// One or more quality-gate invariants failed; carries human-readable + /// reasons for each unmet invariant. + GateFailed { reasons: Vec }, + /// The task is not claimed by the completing member, or is not in progress. + NotClaimed, + /// No task matched the given team + task id. + UnknownTask, +} diff --git a/tests/json_rpc_e2e.rs b/tests/json_rpc_e2e.rs index b3053cbb7..83a1ea2de 100644 --- a/tests/json_rpc_e2e.rs +++ b/tests/json_rpc_e2e.rs @@ -3380,6 +3380,82 @@ async fn json_rpc_agent_team_coordination_roundtrip() { Some("claimed") ); + // Complete B with no evidence but requireEvidence → quality gate fails. + let complete_b_blocked = post_json_rpc( + &rpc_base, + 9360, + "openhuman.agent_team_complete_task", + json!({ + "teamId": team_id, + "taskId": task_b_id, + "memberId": bob_id, + "evidence": [], + "requireEvidence": true + }), + ) + .await; + let complete_b_blocked_outer = + assert_no_jsonrpc_error(&complete_b_blocked, "agent_team_complete_task B gate-fail"); + assert_eq!( + complete_b_blocked_outer + .get("result") + .and_then(|r| r.get("kind")) + .and_then(serde_json::Value::as_str), + Some("gateFailed") + ); + + // Complete B again with evidence → passes the gate, task is now done. + let complete_b_ok = post_json_rpc( + &rpc_base, + 9361, + "openhuman.agent_team_complete_task", + json!({ + "teamId": team_id, + "taskId": task_b_id, + "memberId": bob_id, + "evidence": ["https://ci/run/42"], + "requireEvidence": true + }), + ) + .await; + let complete_b_ok_outer = + assert_no_jsonrpc_error(&complete_b_ok, "agent_team_complete_task B done"); + assert_eq!( + complete_b_ok_outer + .get("result") + .and_then(|r| r.get("kind")) + .and_then(serde_json::Value::as_str), + Some("completed") + ); + + // Shut alice down → member stopped (her task A is already done, so nothing + // is released back to the queue). + let shutdown_alice = post_json_rpc( + &rpc_base, + 9362, + "openhuman.agent_team_shutdown_member", + json!({ "teamId": team_id, "memberId": alice_id }), + ) + .await; + let shutdown_alice_outer = + assert_no_jsonrpc_error(&shutdown_alice, "agent_team_shutdown_member alice"); + assert_eq!( + shutdown_alice_outer + .get("result") + .and_then(|r| r.get("member")) + .and_then(|m| m.get("memberStatus")) + .and_then(serde_json::Value::as_str), + Some("stopped") + ); + assert_eq!( + shutdown_alice_outer + .get("result") + .and_then(|r| r.get("releasedTaskIds")) + .and_then(serde_json::Value::as_array) + .map(|ids| ids.len()), + Some(0) + ); + // Message bob from alice, then list messages. let message = post_json_rpc( &rpc_base,