From ac497565398bf0860d4d376763657ed37a768e38 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 20 Sep 2026 01:40:31 +0300 Subject: [PATCH 1/5] Split run ledger operations into modules Co-authored-by: Medulla --- .../tinyagents-session/src/run_ledger/ops.rs | 1074 +---------------- .../src/run_ledger/ops/README.md | 14 + .../src/run_ledger/ops/rows.rs | 246 ++++ .../src/run_ledger/ops/team.rs | 819 +++++++++++++ 4 files changed, 1104 insertions(+), 1049 deletions(-) create mode 100644 crates/tinyagents-session/src/run_ledger/ops/README.md create mode 100644 crates/tinyagents-session/src/run_ledger/ops/rows.rs create mode 100644 crates/tinyagents-session/src/run_ledger/ops/team.rs diff --git a/crates/tinyagents-session/src/run_ledger/ops.rs b/crates/tinyagents-session/src/run_ledger/ops.rs index 09292f840..de29a8bd9 100644 --- a/crates/tinyagents-session/src/run_ledger/ops.rs +++ b/crates/tinyagents-session/src/run_ledger/ops.rs @@ -25,13 +25,31 @@ use tinyagents_harness::error::Result; use super::super::context::StorageContext; use super::store::init_run_ledger_schema; use super::types::{ - AgentRun, AgentRunListRequest, AgentRunListResponse, AgentRunStatus, AgentRunUpsert, AgentTeam, - AgentTeamListRequest, AgentTeamListResponse, AgentTeamMember, AgentTeamMemberStatus, - AgentTeamMemberUpsert, AgentTeamStatus, AgentTeamTask, AgentTeamTaskStatus, - AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, CompletionOutcome, RunEvent, - RunEventAppend, RunEventListRequest, RunEventListResponse, RunTelemetry, RunTelemetryUpsert, - WorkflowLeaseClaim, WorkflowRun, WorkflowRunListRequest, WorkflowRunListResponse, - WorkflowRunUpsert, + AgentRun, AgentRunKind, AgentRunListRequest, AgentRunListResponse, AgentRunStatus, + AgentRunUpsert, AgentTeam, AgentTeamListRequest, AgentTeamListResponse, AgentTeamMember, + AgentTeamMemberStatus, AgentTeamMemberUpsert, AgentTeamStatus, AgentTeamTask, + AgentTeamTaskStatus, AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, CompletionOutcome, + RunEvent, RunEventAppend, RunEventListRequest, RunEventListResponse, RunTelemetry, + RunTelemetryUpsert, WorkflowLeaseClaim, WorkflowRun, WorkflowRunListRequest, + WorkflowRunListResponse, WorkflowRunStatus, WorkflowRunUpsert, +}; + +mod rows; +mod team; + +use rows::{ + get_agent_run_inner, get_agent_team_inner, get_agent_team_member_inner, + get_agent_team_task_inner, get_run_telemetry_inner, map_agent_run_row, + map_agent_team_member_row, map_agent_team_row, map_agent_team_task_row, map_run_event_row, + map_run_telemetry_row, map_workflow_run_row, +}; + +pub use team::{ + claim_agent_team_task, complete_agent_team_task, get_agent_team, get_agent_team_member, + get_agent_team_task, list_agent_team_members, list_agent_team_tasks, list_agent_teams, + mark_agent_team_member_idle, mark_agent_team_member_running, release_agent_team_task, + shutdown_agent_team_member, upsert_agent_team, upsert_agent_team_member, + upsert_agent_team_task, }; /// Grep prefix for run-ledger logging. @@ -788,1045 +806,3 @@ pub fn list_workflow_runs( Ok(WorkflowRunListResponse { runs, count }) }) } - -// --------------------------------------------------------------------------- -// Agent-team coordination (issue #3374) -// --------------------------------------------------------------------------- - -/// Insert or update a team row. -pub fn upsert_agent_team(workspace_dir: &Path, upsert: AgentTeamUpsert) -> Result { - let now = Utc::now(); - let created_at = upsert.created_at.unwrap_or(now); - tinyagents_tracing::debug!( - "{LOG_PREFIX} upsert_agent_team.entry id={} lead={} status={}", - upsert.id, - upsert.lead_agent_id, - upsert.status.as_str() - ); - // One transaction for the write *and* the read-back. Committing the insert - // on an autocommit connection, closing it, then re-opening to `get_*` hands - // the caller whatever a concurrent writer left behind rather than what this - // call wrote — an upsert that reports someone else's row. - let team = crate::store::with_transaction(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - conn.execute( - "INSERT INTO agent_teams ( - id, parent_thread_id, lead_agent_id, status, summary, - created_at, updated_at, closed_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) - ON CONFLICT(id) DO UPDATE SET - parent_thread_id = COALESCE(excluded.parent_thread_id, agent_teams.parent_thread_id), - lead_agent_id = excluded.lead_agent_id, - status = excluded.status, - summary = COALESCE(excluded.summary, agent_teams.summary), - updated_at = excluded.updated_at, - closed_at = COALESCE(excluded.closed_at, agent_teams.closed_at)", - params![ - upsert.id, - upsert.parent_thread_id, - upsert.lead_agent_id, - upsert.status.as_str(), - upsert.summary, - created_at.to_rfc3339(), - now.to_rfc3339(), - upsert.closed_at.map(|dt| dt.to_rfc3339()), - ], - ) - .storage_context("upsert agent team")?; - get_agent_team_inner(conn, &upsert.id)?.storage_context("agent team missing after upsert") - })?; - tinyagents_tracing::debug!("{LOG_PREFIX} upsert_agent_team.exit id={}", team.id); - Ok(team) -} - -/// Fetch a single team by id. -pub fn get_agent_team(workspace_dir: &Path, id: &str) -> Result> { - tinyagents_tracing::debug!("{LOG_PREFIX} get_agent_team.entry id={id}"); - crate::store::with_connection(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - let team = get_agent_team_inner(conn, id)?; - tinyagents_tracing::debug!( - "{LOG_PREFIX} get_agent_team.exit id={id} found={}", - team.is_some() - ); - Ok(team) - }) -} - -/// List teams, most-recently-updated first, with optional thread/status filters. -pub fn list_agent_teams( - workspace_dir: &Path, - request: &AgentTeamListRequest, -) -> Result { - tinyagents_tracing::debug!( - "{LOG_PREFIX} list_agent_teams.entry parent_thread={:?} status={:?} limit={:?} offset={:?}", - request.parent_thread_id, - request.status, - request.limit, - request.offset - ); - crate::store::with_connection(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - let mut where_clauses = Vec::new(); - let mut values: Vec> = Vec::new(); - - if let Some(thread) = request - .parent_thread_id - .as_deref() - .filter(|s| !s.trim().is_empty()) - { - values.push(Box::new(thread.to_string())); - where_clauses.push(format!("parent_thread_id = ?{}", values.len())); - } - if let Some(status) = request.status.as_deref().filter(|s| !s.trim().is_empty()) { - values.push(Box::new(status.to_string())); - where_clauses.push(format!("status = ?{}", values.len())); - } - - let where_sql = if where_clauses.is_empty() { - String::new() - } else { - format!("WHERE {}", where_clauses.join(" AND ")) - }; - let count_sql = format!("SELECT COUNT(*) FROM agent_teams {where_sql}"); - let params_ref: Vec<&dyn rusqlite::types::ToSql> = - values.iter().map(|v| v.as_ref()).collect(); - let count = conn.query_row(&count_sql, params_ref.as_slice(), |row| { - row.get::<_, i64>(0) - })? as usize; - - let limit = request.limit.unwrap_or(50).min(500) as i64; - // `offset` is `u64`; convert checked so a value > i64::MAX surfaces a - // clear error instead of wrapping negative and corrupting pagination. - let offset = i64::try_from(request.offset.unwrap_or(0)) - .storage_context("agent team list offset exceeds i64::MAX")?; - values.push(Box::new(limit)); - let limit_idx = values.len(); - values.push(Box::new(offset)); - let offset_idx = values.len(); - - let query_sql = format!( - "SELECT id, parent_thread_id, lead_agent_id, status, summary, - created_at, updated_at, closed_at - FROM agent_teams {where_sql} - ORDER BY updated_at DESC - LIMIT ?{limit_idx} OFFSET ?{offset_idx}" - ); - let params_ref: Vec<&dyn rusqlite::types::ToSql> = - values.iter().map(|v| v.as_ref()).collect(); - let mut stmt = conn.prepare(&query_sql)?; - let rows = stmt.query_map(params_ref.as_slice(), map_agent_team_row)?; - let mut teams = Vec::new(); - for row in rows { - teams.push(row?); - } - tinyagents_tracing::debug!( - "{LOG_PREFIX} list_agent_teams.exit count={count} returned={}", - teams.len() - ); - Ok(AgentTeamListResponse { teams, count }) - }) -} - -/// Insert or update a team member. `UNIQUE(team_id, name)` enforces unique names. -pub fn upsert_agent_team_member( - workspace_dir: &Path, - upsert: AgentTeamMemberUpsert, -) -> Result { - let now = Utc::now(); - let created_at = upsert.created_at.unwrap_or(now); - tinyagents_tracing::debug!( - "{LOG_PREFIX} upsert_agent_team_member.entry id={} team={} name={} status={}", - upsert.id, - upsert.team_id, - upsert.name, - upsert.member_status.as_str() - ); - // One transaction for the write *and* the read-back. Committing the insert - // on an autocommit connection, closing it, then re-opening to `get_*` hands - // the caller whatever a concurrent writer left behind rather than what this - // call wrote — an upsert that reports someone else's row. - let member = crate::store::with_transaction(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - conn.execute( - "INSERT INTO agent_team_members ( - id, team_id, name, agent_id, member_status, - current_task_id, worker_thread_id, run_id, created_at, updated_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10) - ON CONFLICT(id) DO UPDATE SET - name = excluded.name, - agent_id = COALESCE(excluded.agent_id, agent_team_members.agent_id), - member_status = excluded.member_status, - current_task_id = COALESCE(excluded.current_task_id, agent_team_members.current_task_id), - worker_thread_id = COALESCE(excluded.worker_thread_id, agent_team_members.worker_thread_id), - run_id = COALESCE(excluded.run_id, agent_team_members.run_id), - updated_at = excluded.updated_at", - params![ - upsert.id, - upsert.team_id, - upsert.name, - upsert.agent_id, - upsert.member_status.as_str(), - upsert.current_task_id, - upsert.worker_thread_id, - upsert.run_id, - created_at.to_rfc3339(), - now.to_rfc3339(), - ], - ) - .storage_context("upsert agent team member")?; - get_agent_team_member_inner(conn, &upsert.id)? - .storage_context("agent team member missing after upsert") - })?; - tinyagents_tracing::debug!( - "{LOG_PREFIX} upsert_agent_team_member.exit id={}", - member.id - ); - Ok(member) -} - -/// Fetch a single member by id. -pub fn get_agent_team_member(workspace_dir: &Path, id: &str) -> Result> { - tinyagents_tracing::debug!("{LOG_PREFIX} get_agent_team_member.entry id={id}"); - crate::store::with_connection(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - let member = get_agent_team_member_inner(conn, id)?; - tinyagents_tracing::debug!( - "{LOG_PREFIX} get_agent_team_member.exit id={id} found={}", - member.is_some() - ); - Ok(member) - }) -} - -/// List all members of a team, by creation order. -pub fn list_agent_team_members( - workspace_dir: &Path, - team_id: &str, -) -> Result> { - tinyagents_tracing::debug!("{LOG_PREFIX} list_agent_team_members.entry team={team_id}"); - crate::store::with_connection(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - let mut stmt = conn.prepare( - "SELECT id, team_id, name, agent_id, member_status, - current_task_id, worker_thread_id, run_id, created_at, updated_at - FROM agent_team_members WHERE team_id = ?1 - ORDER BY created_at ASC", - )?; - let rows = stmt.query_map(params![team_id], map_agent_team_member_row)?; - let mut members = Vec::new(); - for row in rows { - members.push(row?); - } - tinyagents_tracing::debug!( - "{LOG_PREFIX} list_agent_team_members.exit team={team_id} count={}", - members.len() - ); - Ok(members) - }) -} - -/// Insert or update a team task. -pub fn upsert_agent_team_task( - workspace_dir: &Path, - upsert: AgentTeamTaskUpsert, -) -> Result { - let now = Utc::now(); - let created_at = upsert.created_at.unwrap_or(now); - let depends_on_json = - serde_json::to_string(&upsert.depends_on).storage_context("serialize task depends_on")?; - let evidence_json = - serde_json::to_string(&upsert.evidence).storage_context("serialize task evidence")?; - let gate_status = upsert.gate_status.unwrap_or_else(|| "pending".to_string()); - tinyagents_tracing::debug!( - "{LOG_PREFIX} upsert_agent_team_task.entry id={} team={} status={} deps={}", - upsert.id, - upsert.team_id, - upsert.status.as_str(), - upsert.depends_on.len() - ); - // One transaction for the write *and* the read-back. Committing the insert - // on an autocommit connection, closing it, then re-opening to `get_*` hands - // the caller whatever a concurrent writer left behind rather than what this - // call wrote — an upsert that reports someone else's row. - let task = crate::store::with_transaction(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - conn.execute( - "INSERT INTO agent_team_tasks ( - id, team_id, title, objective, status, owner_member_id, - claimed_by_member_id, claim_token, depends_on_json, gate_status, - gate_reason, evidence_json, source_run_id, order_index, - created_at, updated_at - ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, NULL, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14) - ON CONFLICT(id) DO UPDATE SET - title = excluded.title, - objective = COALESCE(excluded.objective, agent_team_tasks.objective), - status = excluded.status, - -- A claim only means anything while the task is in_progress. - -- Editing a claimed task through this upsert (to change - -- dependencies, or to reset status during recovery) used to - -- leave claimed_by_member_id/claim_token set on a todo row, - -- which strands it: a fresh claim returns AlreadyClaimed, - -- completion returns NotClaimed, and release/shutdown skip it - -- because they only match in_progress. Drop the claim whenever - -- the new status is not in_progress; preserve it otherwise so - -- an unrelated edit does not steal a live claim. - claimed_by_member_id = CASE - WHEN excluded.status = 'in_progress' - THEN agent_team_tasks.claimed_by_member_id ELSE NULL END, - claim_token = CASE - WHEN excluded.status = 'in_progress' - THEN agent_team_tasks.claim_token ELSE NULL END, - owner_member_id = COALESCE(excluded.owner_member_id, agent_team_tasks.owner_member_id), - depends_on_json = excluded.depends_on_json, - gate_status = excluded.gate_status, - gate_reason = COALESCE(excluded.gate_reason, agent_team_tasks.gate_reason), - evidence_json = excluded.evidence_json, - source_run_id = COALESCE(excluded.source_run_id, agent_team_tasks.source_run_id), - order_index = excluded.order_index, - updated_at = excluded.updated_at", - params![ - upsert.id, - upsert.team_id, - upsert.title, - upsert.objective, - upsert.status.as_str(), - upsert.owner_member_id, - depends_on_json, - gate_status, - upsert.gate_reason, - evidence_json, - upsert.source_run_id, - upsert.order_index, - created_at.to_rfc3339(), - now.to_rfc3339(), - ], - ) - .storage_context("upsert agent team task")?; - get_agent_team_task_inner(conn, &upsert.id)? - .storage_context("agent team task missing after upsert") - })?; - tinyagents_tracing::debug!("{LOG_PREFIX} upsert_agent_team_task.exit id={}", task.id); - Ok(task) -} - -/// Fetch a single task by id. -pub fn get_agent_team_task(workspace_dir: &Path, id: &str) -> Result> { - tinyagents_tracing::debug!("{LOG_PREFIX} get_agent_team_task.entry id={id}"); - crate::store::with_connection(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - let task = get_agent_team_task_inner(conn, id)?; - tinyagents_tracing::debug!( - "{LOG_PREFIX} get_agent_team_task.exit id={id} found={}", - task.is_some() - ); - Ok(task) - }) -} - -/// List all tasks of a team, by `order_index` then creation order. -pub fn list_agent_team_tasks(workspace_dir: &Path, team_id: &str) -> Result> { - tinyagents_tracing::debug!("{LOG_PREFIX} list_agent_team_tasks.entry team={team_id}"); - crate::store::with_connection(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - let mut stmt = conn.prepare( - "SELECT id, team_id, title, objective, status, owner_member_id, - claimed_by_member_id, claim_token, depends_on_json, gate_status, - gate_reason, evidence_json, source_run_id, order_index, - created_at, updated_at - FROM agent_team_tasks WHERE team_id = ?1 - ORDER BY order_index ASC, created_at ASC", - )?; - let rows = stmt.query_map(params![team_id], map_agent_team_task_row)?; - let mut tasks = Vec::new(); - for row in rows { - tasks.push(row?); - } - tinyagents_tracing::debug!( - "{LOG_PREFIX} list_agent_team_tasks.exit team={team_id} count={}", - tasks.len() - ); - Ok(tasks) - }) -} - -/// Atomically claim a task for a member. -/// -/// All steps run inside a single `with_connection` transaction so that the -/// dependency check and the compare-and-swap observe a consistent snapshot: -/// 1. Resolve the task by `(id, team_id)`; absent → [`ClaimOutcome::UnknownTask`]. -/// 2. For every dependency id, look up its status; collect those not `done` -/// into `unmet`. Non-empty → [`ClaimOutcome::Blocked`]. -/// 3. WHERE-guarded `UPDATE ... WHERE claimed_by_member_id IS NULL`: SQLite -/// serializes writers, so exactly one concurrent claimer flips the row from -/// unclaimed to claimed. `rows_affected == 0` → already taken -/// ([`ClaimOutcome::AlreadyClaimed`]); otherwise re-fetch and return -/// [`ClaimOutcome::Claimed`]. -pub fn claim_agent_team_task( - workspace_dir: &Path, - team_id: &str, - task_id: &str, - member_id: &str, - claim_token: &str, -) -> Result { - tinyagents_tracing::debug!( - "{LOG_PREFIX} claim_agent_team_task.entry team={team_id} task={task_id} member={member_id}" - ); - let outcome = crate::store::with_transaction(workspace_dir, |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, - _ => { - tinyagents_tracing::debug!( - "{LOG_PREFIX} claim_agent_team_task.unknown team={team_id} task={task_id}" - ); - return Ok(ClaimOutcome::UnknownTask); - } - }; - - // 2. Dependency gate: every dep must be `done`. - let mut unmet = Vec::new(); - 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()?; - let is_done = dep_status.as_deref() == Some(AgentTeamTaskStatus::Done.as_str()); - if !is_done { - unmet.push(dep_id.clone()); - } - } - if !unmet.is_empty() { - tinyagents_tracing::debug!( - "{LOG_PREFIX} claim_agent_team_task.blocked team={team_id} task={task_id} unmet={}", - unmet.len() - ); - return Ok(ClaimOutcome::Blocked { unmet }); - } - - // 3. Compare-and-swap on the unclaimed guard **and the status**. - // - // `claimed_by_member_id IS NULL` alone is not a guard: `upsert_agent_team_task` - // deliberately NULLs that column whenever the new status is not - // `in_progress`, so every `done` task also satisfies it. A stale worker - // re-claiming a finished task therefore flipped it straight back to - // `in_progress` — and stranded everything downstream, because the - // completion gate re-checks that each dependency is still `done` and now - // reports it unfinished. A terminal task is not claimable, whatever its - // claim column says. - let now = Utc::now(); - let rows_affected = conn - .execute( - "UPDATE agent_team_tasks - SET claimed_by_member_id = ?1, claim_token = ?2, status = 'in_progress', updated_at = ?3 - WHERE id = ?4 AND team_id = ?5 AND claimed_by_member_id IS NULL - AND status IN ('todo', 'ready', 'blocked')", - params![member_id, claim_token, now.to_rfc3339(), task_id, team_id], - ) - .storage_context("compare-and-swap claim agent team task")?; - if rows_affected == 0 { - tinyagents_tracing::debug!( - "{LOG_PREFIX} claim_agent_team_task.already_claimed team={team_id} task={task_id} \ - status={}", - task.status.as_str() - ); - return Ok(ClaimOutcome::AlreadyClaimed); - } - - let claimed = get_agent_team_task_inner(conn, task_id)? - .storage_context("claimed task missing after compare-and-swap")?; - Ok(ClaimOutcome::Claimed(Box::new(claimed))) - })?; - tinyagents_tracing::debug!( - "{LOG_PREFIX} claim_agent_team_task.exit team={team_id} task={task_id} outcome={}", - match &outcome { - ClaimOutcome::Claimed(_) => "claimed", - ClaimOutcome::AlreadyClaimed => "already_claimed", - ClaimOutcome::Blocked { .. } => "blocked", - ClaimOutcome::UnknownTask => "unknown", - } - ); - 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( - workspace_dir: &Path, - team_id: &str, - task_id: &str, - member_id: &str, - evidence: &[String], - require_evidence: bool, -) -> Result { - tinyagents_tracing::debug!( - "{LOG_PREFIX} complete_agent_team_task.entry team={team_id} task={task_id} member={member_id}" - ); - let outcome = crate::store::with_transaction(workspace_dir, |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, - _ => { - tinyagents_tracing::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 { - tinyagents_tracing::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("; "); - // Persist the merged evidence even though the gate failed. Evidence - // accumulates across attempts (see `merged_evidence` above), so - // dropping it here punished a caller for an unrelated gate failure: - // after a dependency was fixed, a retry that did not resend the same - // links would fail `require_evidence` on evidence it had already - // submitted. Only the gate verdict is a failure; the submission is - // still real. - let evidence_json = serde_json::to_string(&merged_evidence) - .storage_context("serialize completion evidence")?; - conn.execute( - "UPDATE agent_team_tasks - SET gate_status = 'failed', gate_reason = ?1, - evidence_json = ?2, updated_at = ?3 - WHERE id = ?4 AND team_id = ?5", - params![joined, evidence_json, now.to_rfc3339(), task_id, team_id], - ) - .storage_context("record failed completion gate")?; - tinyagents_tracing::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) - .storage_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], - ) - .storage_context("complete agent team task")?; - if rows_affected == 0 { - tinyagents_tracing::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)? - .storage_context("completed task missing after update")?; - Ok(CompletionOutcome::Completed(Box::new(done))) - })?; - tinyagents_tracing::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 - .as_ref() - .filter(|owner| 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( - workspace_dir: &Path, - team_id: &str, - member_id: &str, -) -> Result)>> { - tinyagents_tracing::debug!( - "{LOG_PREFIX} shutdown_agent_team_member.entry team={team_id} member={member_id}" - ); - let result = crate::store::with_transaction(workspace_dir, |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 => {} - _ => { - tinyagents_tracing::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], - ) - .storage_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], - ) - .storage_context("stop agent team member")?; - - let member = get_agent_team_member_inner(conn, member_id)? - .storage_context("member missing after shutdown")?; - Ok(Some((member, released))) - })?; - tinyagents_tracing::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) -} - -/// Mark a member as actively running a task: status → `active`, with the -/// current task id and the worker/run identifiers of the spawned agent. Used by -/// the live runtime right after it claims a task and dispatches a worker. -/// Returns the updated member, or `None` if the member is not in the team. -pub fn mark_agent_team_member_running( - workspace_dir: &Path, - team_id: &str, - member_id: &str, - task_id: &str, - worker_thread_id: &str, - run_id: &str, -) -> Result> { - tinyagents_tracing::debug!( - "{LOG_PREFIX} mark_agent_team_member_running.entry team={team_id} member={member_id} task={task_id} run={run_id}" - ); - crate::store::with_connection(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - let now = Utc::now(); - let changed = conn - .execute( - "UPDATE agent_team_members - SET member_status = 'active', current_task_id = ?1, - worker_thread_id = ?2, run_id = ?3, updated_at = ?4 - WHERE id = ?5 AND team_id = ?6", - params![ - task_id, - worker_thread_id, - run_id, - now.to_rfc3339(), - member_id, - team_id - ], - ) - .storage_context("mark agent team member running")?; - if changed == 0 { - return Ok(None); - } - get_agent_team_member_inner(conn, member_id) - }) -} - -/// Mark a member idle: status → `idle`, clearing `current_task_id`. The -/// `worker_thread_id` / `run_id` are intentionally retained as a pointer to the -/// member's last run for history. Returns the updated member, or `None` if the -/// member is not in the team. Used when a worker run finishes (completed, -/// gate-failed, or failed) so the member is free to pick up new work. -pub fn mark_agent_team_member_idle( - workspace_dir: &Path, - team_id: &str, - member_id: &str, -) -> Result> { - tinyagents_tracing::debug!( - "{LOG_PREFIX} mark_agent_team_member_idle.entry team={team_id} member={member_id}" - ); - crate::store::with_connection(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - let now = Utc::now(); - let changed = conn - .execute( - "UPDATE agent_team_members - SET member_status = 'idle', current_task_id = NULL, updated_at = ?1 - WHERE id = ?2 AND team_id = ?3", - params![now.to_rfc3339(), member_id, team_id], - ) - .storage_context("mark agent team member idle")?; - if changed == 0 { - return Ok(None); - } - get_agent_team_member_inner(conn, member_id) - }) -} - -/// Release a single `in_progress` task back to `todo`, clearing its claim and -/// resetting the quality gate. Returns `true` if a row was actually released -/// (the task existed, belonged to the team, and was `in_progress`). Used by the -/// live runtime when a worker run fails or is aborted, so the task is free for -/// another teammate — the per-task analogue of the bulk release in -/// `shutdown_agent_team_member`. -pub fn release_agent_team_task(workspace_dir: &Path, team_id: &str, task_id: &str) -> Result { - tinyagents_tracing::debug!( - "{LOG_PREFIX} release_agent_team_task.entry team={team_id} task={task_id}" - ); - crate::store::with_connection(workspace_dir, |conn| { - init_run_ledger_schema(conn)?; - let now = Utc::now(); - let changed = conn - .execute( - "UPDATE agent_team_tasks - SET status = 'todo', claimed_by_member_id = NULL, claim_token = NULL, - gate_status = 'pending', gate_reason = NULL, updated_at = ?1 - WHERE id = ?2 AND team_id = ?3 AND status = 'in_progress'", - params![now.to_rfc3339(), task_id, team_id], - ) - .storage_context("release agent team task")?; - tinyagents_tracing::debug!( - "{LOG_PREFIX} release_agent_team_task.exit team={team_id} task={task_id} released={}", - changed > 0 - ); - Ok(changed > 0) - }) -} - -/// Connection-scoped team lookup, so an upsert can read its own write back -/// inside the same transaction. -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, - created_at, updated_at, closed_at - FROM agent_teams WHERE id = ?1", - )?; - stmt.query_row(params![id], map_agent_team_row) - .optional() - .map_err(Into::into) -} - -/// Connection-scoped member lookup, so an upsert can read its own write back -/// inside the same transaction. -fn get_agent_team_member_inner(conn: &Connection, id: &str) -> Result> { - let mut stmt = conn.prepare( - "SELECT id, team_id, name, agent_id, member_status, - current_task_id, worker_thread_id, run_id, created_at, updated_at - FROM agent_team_members WHERE id = ?1", - )?; - stmt.query_row(params![id], map_agent_team_member_row) - .optional() - .map_err(Into::into) -} - -/// Connection-scoped task lookup, so a claim/completion transaction can read -/// its own write back inside the same transaction. -fn get_agent_team_task_inner(conn: &Connection, id: &str) -> Result> { - let mut stmt = conn.prepare( - "SELECT id, team_id, title, objective, status, owner_member_id, - claimed_by_member_id, claim_token, depends_on_json, gate_status, - gate_reason, evidence_json, source_run_id, order_index, - created_at, updated_at - FROM agent_team_tasks WHERE id = ?1", - )?; - stmt.query_row(params![id], map_agent_team_task_row) - .optional() - .map_err(Into::into) -} - -fn map_agent_team_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { - Ok(AgentTeam { - id: row.get(0)?, - parent_thread_id: row.get(1)?, - lead_agent_id: row.get(2)?, - status: AgentTeamStatus::parse(&row.get::<_, String>(3)?), - summary: row.get(4)?, - created_at: parse_rfc3339(&row.get::<_, String>(5)?)?, - updated_at: parse_rfc3339(&row.get::<_, String>(6)?)?, - closed_at: parse_rfc3339_opt(row.get(7)?)?, - }) -} - -fn map_agent_team_member_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { - Ok(AgentTeamMember { - id: row.get(0)?, - team_id: row.get(1)?, - name: row.get(2)?, - agent_id: row.get(3)?, - member_status: AgentTeamMemberStatus::parse(&row.get::<_, String>(4)?), - current_task_id: row.get(5)?, - worker_thread_id: row.get(6)?, - run_id: row.get(7)?, - created_at: parse_rfc3339(&row.get::<_, String>(8)?)?, - updated_at: parse_rfc3339(&row.get::<_, String>(9)?)?, - }) -} - -fn map_agent_team_task_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { - Ok(AgentTeamTask { - id: row.get(0)?, - team_id: row.get(1)?, - title: row.get(2)?, - objective: row.get(3)?, - status: AgentTeamTaskStatus::parse(&row.get::<_, String>(4)?), - owner_member_id: row.get(5)?, - claimed_by_member_id: row.get(6)?, - claim_token: row.get(7)?, - depends_on: serde_json::from_str(&row.get::<_, String>(8)?).unwrap_or_default(), - gate_status: row.get(9)?, - gate_reason: row.get(10)?, - evidence: serde_json::from_str(&row.get::<_, String>(11)?).unwrap_or_default(), - source_run_id: row.get(12)?, - order_index: row.get(13)?, - created_at: parse_rfc3339(&row.get::<_, String>(14)?)?, - updated_at: parse_rfc3339(&row.get::<_, String>(15)?)?, - }) -} - -/// Connection-scoped agent-run lookup, so an upsert can read its own write -/// back inside the same transaction. -fn get_agent_run_inner(conn: &Connection, id: &str) -> Result> { - let mut stmt = conn.prepare( - "SELECT id, kind, parent_run_id, parent_thread_id, agent_id, status, - prompt_ref, worker_thread_id, task_board_id, task_card_id, - checkpoint_path, checkpoint_json, summary, error, metadata_json, - started_at, updated_at, completed_at - FROM agent_runs WHERE id = ?1", - )?; - stmt.query_row(params![id], |row| map_agent_run_row(conn, row)) - .optional() - .map_err(Into::into) -} - -/// Connection-scoped telemetry lookup that errors when the row is absent — -/// used right after [`upsert_run_telemetry`] writes it, where a miss means -/// the write silently failed. -fn get_run_telemetry_inner(conn: &Connection, run_id: &str) -> Result { - let mut stmt = conn.prepare( - "SELECT run_id, input_tokens, output_tokens, cached_input_tokens, cost_usd, - elapsed_ms, tool_count, model, provider, error, updated_at - FROM run_telemetry WHERE run_id = ?1", - )?; - stmt.query_row(params![run_id], map_run_telemetry_row) - .storage_context("run telemetry missing after upsert") -} - -/// Connection-scoped telemetry lookup used when joining telemetry onto an -/// [`AgentRun`], where no telemetry row yet existing is a normal `None` -/// rather than an error. -fn get_optional_run_telemetry( - conn: &Connection, - run_id: &str, -) -> rusqlite::Result> { - let mut stmt = conn.prepare( - "SELECT run_id, input_tokens, output_tokens, cached_input_tokens, cost_usd, - elapsed_ms, tool_count, model, provider, error, updated_at - FROM run_telemetry WHERE run_id = ?1", - )?; - stmt.query_row(params![run_id], map_run_telemetry_row) - .optional() -} - -fn map_agent_run_row(conn: &Connection, row: &rusqlite::Row<'_>) -> rusqlite::Result { - let id: String = row.get(0)?; - let checkpoint_json: Option = row.get(11)?; - let metadata_json: String = row.get(14)?; - Ok(AgentRun { - id: id.clone(), - kind: super::types::AgentRunKind::parse(&row.get::<_, String>(1)?), - parent_run_id: row.get(2)?, - parent_thread_id: row.get(3)?, - agent_id: row.get(4)?, - status: AgentRunStatus::parse(&row.get::<_, String>(5)?), - prompt_ref: row.get(6)?, - worker_thread_id: row.get(7)?, - task_board_id: row.get(8)?, - task_card_id: row.get(9)?, - checkpoint_path: row.get(10)?, - checkpoint: parse_json_opt(checkpoint_json), - summary: row.get(12)?, - error: row.get(13)?, - metadata: parse_json(metadata_json), - telemetry: get_optional_run_telemetry(conn, &id)?, - started_at: parse_rfc3339(&row.get::<_, String>(15)?)?, - updated_at: parse_rfc3339(&row.get::<_, String>(16)?)?, - completed_at: parse_rfc3339_opt(row.get(17)?)?, - }) -} - -fn map_workflow_run_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { - Ok(WorkflowRun { - id: row.get(0)?, - definition_id: row.get(1)?, - parent_thread_id: row.get(2)?, - input: parse_json(row.get(3)?), - phase_states: parse_json(row.get(4)?), - child_run_ids: serde_json::from_str(&row.get::<_, String>(5)?).unwrap_or_default(), - status: super::types::WorkflowRunStatus::parse(&row.get::<_, String>(6)?), - summary: row.get(7)?, - started_at: parse_rfc3339(&row.get::<_, String>(8)?)?, - updated_at: parse_rfc3339(&row.get::<_, String>(9)?)?, - completed_at: parse_rfc3339_opt(row.get(10)?)?, - revision: row.get::<_, i64>(11)? as u64, - lease_owner: row.get(12)?, - lease_expires_at: parse_rfc3339_opt(row.get(13)?)?, - }) -} - -fn map_run_event_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { - Ok(RunEvent { - run_id: row.get(0)?, - sequence: row.get::<_, i64>(1)? as u64, - event_type: row.get(2)?, - payload: parse_json(row.get(3)?), - timestamp: parse_rfc3339(&row.get::<_, String>(4)?)?, - }) -} - -fn map_run_telemetry_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { - Ok(RunTelemetry { - run_id: row.get(0)?, - input_tokens: row.get::<_, i64>(1)? as u64, - output_tokens: row.get::<_, i64>(2)? as u64, - cached_input_tokens: row.get::<_, i64>(3)? as u64, - cost_usd: row.get(4)?, - elapsed_ms: row.get::<_, Option>(5)?.map(|v| v as u64), - tool_count: row.get::<_, i64>(6)? as u64, - model: row.get(7)?, - provider: row.get(8)?, - error: row.get(9)?, - updated_at: Some(parse_rfc3339(&row.get::<_, String>(10)?)?), - }) -} - -fn parse_json(raw: String) -> Value { - serde_json::from_str(&raw).unwrap_or_else(|_| json!({})) -} - -fn parse_json_opt(raw: Option) -> Option { - raw.and_then(|value| serde_json::from_str(&value).ok()) -} - -fn parse_rfc3339(raw: &str) -> rusqlite::Result> { - DateTime::parse_from_rfc3339(raw) - .map(|dt| dt.with_timezone(&Utc)) - .map_err(|err| { - rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(err)) - }) -} - -fn parse_rfc3339_opt(raw: Option) -> rusqlite::Result>> { - match raw { - Some(value) => parse_rfc3339(&value).map(Some), - None => Ok(None), - } -} diff --git a/crates/tinyagents-session/src/run_ledger/ops/README.md b/crates/tinyagents-session/src/run_ledger/ops/README.md new file mode 100644 index 000000000..76a6ddbb0 --- /dev/null +++ b/crates/tinyagents-session/src/run_ledger/ops/README.md @@ -0,0 +1,14 @@ +# Run-ledger operations + +This directory implements the durable run-ledger query and mutation surface. + +- `../ops.rs` owns agent-run, workflow-run, telemetry, and event operations, + and re-exports the agent-team API so existing callers retain the same path. +- `team.rs` owns team, member, task, claim, completion, and release lifecycle + operations. +- `rows.rs` owns SQLite row decoding and connection-scoped lookup helpers used + by transactional operations. + +The ledger's public types live in `../types.rs`; schema initialization remains +in `../store.rs`. Tests remain in `../test.rs` because they exercise the +public ledger surface across these implementation modules. diff --git a/crates/tinyagents-session/src/run_ledger/ops/rows.rs b/crates/tinyagents-session/src/run_ledger/ops/rows.rs new file mode 100644 index 000000000..e464364a6 --- /dev/null +++ b/crates/tinyagents-session/src/run_ledger/ops/rows.rs @@ -0,0 +1,246 @@ +//! SQLite row decoding and transaction-scoped lookup helpers. +//! +//! Keeping this detail outside the public ledger operations makes the +//! operation module focused on state transitions and query semantics. + +use super::*; + +/// Connection-scoped team lookup, so an upsert can read its own write back +/// inside the same transaction. +pub(super) 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, + created_at, updated_at, closed_at + FROM agent_teams WHERE id = ?1", + )?; + stmt.query_row(params![id], map_agent_team_row) + .optional() + .map_err(Into::into) +} + +/// Connection-scoped member lookup, so an upsert can read its own write back +/// inside the same transaction. +pub(super) fn get_agent_team_member_inner( + conn: &Connection, + id: &str, +) -> Result> { + let mut stmt = conn.prepare( + "SELECT id, team_id, name, agent_id, member_status, + current_task_id, worker_thread_id, run_id, created_at, updated_at + FROM agent_team_members WHERE id = ?1", + )?; + stmt.query_row(params![id], map_agent_team_member_row) + .optional() + .map_err(Into::into) +} + +/// Connection-scoped task lookup, so a claim/completion transaction can read +/// its own write back inside the same transaction. +pub(super) fn get_agent_team_task_inner( + conn: &Connection, + id: &str, +) -> Result> { + let mut stmt = conn.prepare( + "SELECT id, team_id, title, objective, status, owner_member_id, + claimed_by_member_id, claim_token, depends_on_json, gate_status, + gate_reason, evidence_json, source_run_id, order_index, + created_at, updated_at + FROM agent_team_tasks WHERE id = ?1", + )?; + stmt.query_row(params![id], map_agent_team_task_row) + .optional() + .map_err(Into::into) +} + +pub(super) fn map_agent_team_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(AgentTeam { + id: row.get(0)?, + parent_thread_id: row.get(1)?, + lead_agent_id: row.get(2)?, + status: AgentTeamStatus::parse(&row.get::<_, String>(3)?), + summary: row.get(4)?, + created_at: parse_rfc3339(&row.get::<_, String>(5)?)?, + updated_at: parse_rfc3339(&row.get::<_, String>(6)?)?, + closed_at: parse_rfc3339_opt(row.get(7)?)?, + }) +} + +pub(super) fn map_agent_team_member_row( + row: &rusqlite::Row<'_>, +) -> rusqlite::Result { + Ok(AgentTeamMember { + id: row.get(0)?, + team_id: row.get(1)?, + name: row.get(2)?, + agent_id: row.get(3)?, + member_status: AgentTeamMemberStatus::parse(&row.get::<_, String>(4)?), + current_task_id: row.get(5)?, + worker_thread_id: row.get(6)?, + run_id: row.get(7)?, + created_at: parse_rfc3339(&row.get::<_, String>(8)?)?, + updated_at: parse_rfc3339(&row.get::<_, String>(9)?)?, + }) +} + +pub(super) fn map_agent_team_task_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(AgentTeamTask { + id: row.get(0)?, + team_id: row.get(1)?, + title: row.get(2)?, + objective: row.get(3)?, + status: AgentTeamTaskStatus::parse(&row.get::<_, String>(4)?), + owner_member_id: row.get(5)?, + claimed_by_member_id: row.get(6)?, + claim_token: row.get(7)?, + depends_on: serde_json::from_str(&row.get::<_, String>(8)?).unwrap_or_default(), + gate_status: row.get(9)?, + gate_reason: row.get(10)?, + evidence: serde_json::from_str(&row.get::<_, String>(11)?).unwrap_or_default(), + source_run_id: row.get(12)?, + order_index: row.get(13)?, + created_at: parse_rfc3339(&row.get::<_, String>(14)?)?, + updated_at: parse_rfc3339(&row.get::<_, String>(15)?)?, + }) +} + +/// Connection-scoped agent-run lookup, so an upsert can read its own write +/// back inside the same transaction. +pub(super) fn get_agent_run_inner(conn: &Connection, id: &str) -> Result> { + let mut stmt = conn.prepare( + "SELECT id, kind, parent_run_id, parent_thread_id, agent_id, status, + prompt_ref, worker_thread_id, task_board_id, task_card_id, + checkpoint_path, checkpoint_json, summary, error, metadata_json, + started_at, updated_at, completed_at + FROM agent_runs WHERE id = ?1", + )?; + stmt.query_row(params![id], |row| map_agent_run_row(conn, row)) + .optional() + .map_err(Into::into) +} + +/// Connection-scoped telemetry lookup that errors when the row is absent — +/// used right after [`upsert_run_telemetry`] writes it, where a miss means +/// the write silently failed. +pub(super) fn get_run_telemetry_inner(conn: &Connection, run_id: &str) -> Result { + let mut stmt = conn.prepare( + "SELECT run_id, input_tokens, output_tokens, cached_input_tokens, cost_usd, + elapsed_ms, tool_count, model, provider, error, updated_at + FROM run_telemetry WHERE run_id = ?1", + )?; + stmt.query_row(params![run_id], map_run_telemetry_row) + .storage_context("run telemetry missing after upsert") +} + +/// Connection-scoped telemetry lookup used when joining telemetry onto an +/// [`AgentRun`], where no telemetry row yet existing is a normal `None` +/// rather than an error. +pub(super) fn get_optional_run_telemetry( + conn: &Connection, + run_id: &str, +) -> rusqlite::Result> { + let mut stmt = conn.prepare( + "SELECT run_id, input_tokens, output_tokens, cached_input_tokens, cost_usd, + elapsed_ms, tool_count, model, provider, error, updated_at + FROM run_telemetry WHERE run_id = ?1", + )?; + stmt.query_row(params![run_id], map_run_telemetry_row) + .optional() +} + +pub(super) fn map_agent_run_row( + conn: &Connection, + row: &rusqlite::Row<'_>, +) -> rusqlite::Result { + let id: String = row.get(0)?; + let checkpoint_json: Option = row.get(11)?; + let metadata_json: String = row.get(14)?; + Ok(AgentRun { + id: id.clone(), + kind: AgentRunKind::parse(&row.get::<_, String>(1)?), + parent_run_id: row.get(2)?, + parent_thread_id: row.get(3)?, + agent_id: row.get(4)?, + status: AgentRunStatus::parse(&row.get::<_, String>(5)?), + prompt_ref: row.get(6)?, + worker_thread_id: row.get(7)?, + task_board_id: row.get(8)?, + task_card_id: row.get(9)?, + checkpoint_path: row.get(10)?, + checkpoint: parse_json_opt(checkpoint_json), + summary: row.get(12)?, + error: row.get(13)?, + metadata: parse_json(metadata_json), + telemetry: get_optional_run_telemetry(conn, &id)?, + started_at: parse_rfc3339(&row.get::<_, String>(15)?)?, + updated_at: parse_rfc3339(&row.get::<_, String>(16)?)?, + completed_at: parse_rfc3339_opt(row.get(17)?)?, + }) +} + +pub(super) fn map_workflow_run_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(WorkflowRun { + id: row.get(0)?, + definition_id: row.get(1)?, + parent_thread_id: row.get(2)?, + input: parse_json(row.get(3)?), + phase_states: parse_json(row.get(4)?), + child_run_ids: serde_json::from_str(&row.get::<_, String>(5)?).unwrap_or_default(), + status: WorkflowRunStatus::parse(&row.get::<_, String>(6)?), + summary: row.get(7)?, + started_at: parse_rfc3339(&row.get::<_, String>(8)?)?, + updated_at: parse_rfc3339(&row.get::<_, String>(9)?)?, + completed_at: parse_rfc3339_opt(row.get(10)?)?, + revision: row.get::<_, i64>(11)? as u64, + lease_owner: row.get(12)?, + lease_expires_at: parse_rfc3339_opt(row.get(13)?)?, + }) +} + +pub(super) fn map_run_event_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(RunEvent { + run_id: row.get(0)?, + sequence: row.get::<_, i64>(1)? as u64, + event_type: row.get(2)?, + payload: parse_json(row.get(3)?), + timestamp: parse_rfc3339(&row.get::<_, String>(4)?)?, + }) +} + +pub(super) fn map_run_telemetry_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { + Ok(RunTelemetry { + run_id: row.get(0)?, + input_tokens: row.get::<_, i64>(1)? as u64, + output_tokens: row.get::<_, i64>(2)? as u64, + cached_input_tokens: row.get::<_, i64>(3)? as u64, + cost_usd: row.get(4)?, + elapsed_ms: row.get::<_, Option>(5)?.map(|v| v as u64), + tool_count: row.get::<_, i64>(6)? as u64, + model: row.get(7)?, + provider: row.get(8)?, + error: row.get(9)?, + updated_at: Some(parse_rfc3339(&row.get::<_, String>(10)?)?), + }) +} + +pub(super) fn parse_json(raw: String) -> Value { + serde_json::from_str(&raw).unwrap_or_else(|_| json!({})) +} + +pub(super) fn parse_json_opt(raw: Option) -> Option { + raw.and_then(|value| serde_json::from_str(&value).ok()) +} + +pub(super) fn parse_rfc3339(raw: &str) -> rusqlite::Result> { + DateTime::parse_from_rfc3339(raw) + .map(|dt| dt.with_timezone(&Utc)) + .map_err(|err| { + rusqlite::Error::FromSqlConversionFailure(0, rusqlite::types::Type::Text, Box::new(err)) + }) +} + +pub(super) fn parse_rfc3339_opt(raw: Option) -> rusqlite::Result>> { + match raw { + Some(value) => parse_rfc3339(&value).map(Some), + None => Ok(None), + } +} diff --git a/crates/tinyagents-session/src/run_ledger/ops/team.rs b/crates/tinyagents-session/src/run_ledger/ops/team.rs new file mode 100644 index 000000000..4c9373b8f --- /dev/null +++ b/crates/tinyagents-session/src/run_ledger/ops/team.rs @@ -0,0 +1,819 @@ +//! Durable agent-team coordination operations. +//! +//! Teams, members, and tasks share the run-ledger database but have distinct +//! lifecycle and claim semantics from agent and workflow runs. + +use super::*; + +// --------------------------------------------------------------------------- +// Agent-team coordination (issue #3374) +// --------------------------------------------------------------------------- + +/// Insert or update a team row. +pub fn upsert_agent_team(workspace_dir: &Path, upsert: AgentTeamUpsert) -> Result { + let now = Utc::now(); + let created_at = upsert.created_at.unwrap_or(now); + tinyagents_tracing::debug!( + "{LOG_PREFIX} upsert_agent_team.entry id={} lead={} status={}", + upsert.id, + upsert.lead_agent_id, + upsert.status.as_str() + ); + // One transaction for the write *and* the read-back. Committing the insert + // on an autocommit connection, closing it, then re-opening to `get_*` hands + // the caller whatever a concurrent writer left behind rather than what this + // call wrote — an upsert that reports someone else's row. + let team = crate::store::with_transaction(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + conn.execute( + "INSERT INTO agent_teams ( + id, parent_thread_id, lead_agent_id, status, summary, + created_at, updated_at, closed_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8) + ON CONFLICT(id) DO UPDATE SET + parent_thread_id = COALESCE(excluded.parent_thread_id, agent_teams.parent_thread_id), + lead_agent_id = excluded.lead_agent_id, + status = excluded.status, + summary = COALESCE(excluded.summary, agent_teams.summary), + updated_at = excluded.updated_at, + closed_at = COALESCE(excluded.closed_at, agent_teams.closed_at)", + params![ + upsert.id, + upsert.parent_thread_id, + upsert.lead_agent_id, + upsert.status.as_str(), + upsert.summary, + created_at.to_rfc3339(), + now.to_rfc3339(), + upsert.closed_at.map(|dt| dt.to_rfc3339()), + ], + ) + .storage_context("upsert agent team")?; + get_agent_team_inner(conn, &upsert.id)?.storage_context("agent team missing after upsert") + })?; + tinyagents_tracing::debug!("{LOG_PREFIX} upsert_agent_team.exit id={}", team.id); + Ok(team) +} + +/// Fetch a single team by id. +pub fn get_agent_team(workspace_dir: &Path, id: &str) -> Result> { + tinyagents_tracing::debug!("{LOG_PREFIX} get_agent_team.entry id={id}"); + crate::store::with_connection(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + let team = get_agent_team_inner(conn, id)?; + tinyagents_tracing::debug!( + "{LOG_PREFIX} get_agent_team.exit id={id} found={}", + team.is_some() + ); + Ok(team) + }) +} + +/// List teams, most-recently-updated first, with optional thread/status filters. +pub fn list_agent_teams( + workspace_dir: &Path, + request: &AgentTeamListRequest, +) -> Result { + tinyagents_tracing::debug!( + "{LOG_PREFIX} list_agent_teams.entry parent_thread={:?} status={:?} limit={:?} offset={:?}", + request.parent_thread_id, + request.status, + request.limit, + request.offset + ); + crate::store::with_connection(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + let mut where_clauses = Vec::new(); + let mut values: Vec> = Vec::new(); + + if let Some(thread) = request + .parent_thread_id + .as_deref() + .filter(|s| !s.trim().is_empty()) + { + values.push(Box::new(thread.to_string())); + where_clauses.push(format!("parent_thread_id = ?{}", values.len())); + } + if let Some(status) = request.status.as_deref().filter(|s| !s.trim().is_empty()) { + values.push(Box::new(status.to_string())); + where_clauses.push(format!("status = ?{}", values.len())); + } + + let where_sql = if where_clauses.is_empty() { + String::new() + } else { + format!("WHERE {}", where_clauses.join(" AND ")) + }; + let count_sql = format!("SELECT COUNT(*) FROM agent_teams {where_sql}"); + let params_ref: Vec<&dyn rusqlite::types::ToSql> = + values.iter().map(|v| v.as_ref()).collect(); + let count = conn.query_row(&count_sql, params_ref.as_slice(), |row| { + row.get::<_, i64>(0) + })? as usize; + + let limit = request.limit.unwrap_or(50).min(500) as i64; + // `offset` is `u64`; convert checked so a value > i64::MAX surfaces a + // clear error instead of wrapping negative and corrupting pagination. + let offset = i64::try_from(request.offset.unwrap_or(0)) + .storage_context("agent team list offset exceeds i64::MAX")?; + values.push(Box::new(limit)); + let limit_idx = values.len(); + values.push(Box::new(offset)); + let offset_idx = values.len(); + + let query_sql = format!( + "SELECT id, parent_thread_id, lead_agent_id, status, summary, + created_at, updated_at, closed_at + FROM agent_teams {where_sql} + ORDER BY updated_at DESC + LIMIT ?{limit_idx} OFFSET ?{offset_idx}" + ); + let params_ref: Vec<&dyn rusqlite::types::ToSql> = + values.iter().map(|v| v.as_ref()).collect(); + let mut stmt = conn.prepare(&query_sql)?; + let rows = stmt.query_map(params_ref.as_slice(), map_agent_team_row)?; + let mut teams = Vec::new(); + for row in rows { + teams.push(row?); + } + tinyagents_tracing::debug!( + "{LOG_PREFIX} list_agent_teams.exit count={count} returned={}", + teams.len() + ); + Ok(AgentTeamListResponse { teams, count }) + }) +} + +/// Insert or update a team member. `UNIQUE(team_id, name)` enforces unique names. +pub fn upsert_agent_team_member( + workspace_dir: &Path, + upsert: AgentTeamMemberUpsert, +) -> Result { + let now = Utc::now(); + let created_at = upsert.created_at.unwrap_or(now); + tinyagents_tracing::debug!( + "{LOG_PREFIX} upsert_agent_team_member.entry id={} team={} name={} status={}", + upsert.id, + upsert.team_id, + upsert.name, + upsert.member_status.as_str() + ); + // One transaction for the write *and* the read-back. Committing the insert + // on an autocommit connection, closing it, then re-opening to `get_*` hands + // the caller whatever a concurrent writer left behind rather than what this + // call wrote — an upsert that reports someone else's row. + let member = crate::store::with_transaction(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + conn.execute( + "INSERT INTO agent_team_members ( + id, team_id, name, agent_id, member_status, + current_task_id, worker_thread_id, run_id, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10) + ON CONFLICT(id) DO UPDATE SET + name = excluded.name, + agent_id = COALESCE(excluded.agent_id, agent_team_members.agent_id), + member_status = excluded.member_status, + current_task_id = COALESCE(excluded.current_task_id, agent_team_members.current_task_id), + worker_thread_id = COALESCE(excluded.worker_thread_id, agent_team_members.worker_thread_id), + run_id = COALESCE(excluded.run_id, agent_team_members.run_id), + updated_at = excluded.updated_at", + params![ + upsert.id, + upsert.team_id, + upsert.name, + upsert.agent_id, + upsert.member_status.as_str(), + upsert.current_task_id, + upsert.worker_thread_id, + upsert.run_id, + created_at.to_rfc3339(), + now.to_rfc3339(), + ], + ) + .storage_context("upsert agent team member")?; + get_agent_team_member_inner(conn, &upsert.id)? + .storage_context("agent team member missing after upsert") + })?; + tinyagents_tracing::debug!( + "{LOG_PREFIX} upsert_agent_team_member.exit id={}", + member.id + ); + Ok(member) +} + +/// Fetch a single member by id. +pub fn get_agent_team_member(workspace_dir: &Path, id: &str) -> Result> { + tinyagents_tracing::debug!("{LOG_PREFIX} get_agent_team_member.entry id={id}"); + crate::store::with_connection(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + let member = get_agent_team_member_inner(conn, id)?; + tinyagents_tracing::debug!( + "{LOG_PREFIX} get_agent_team_member.exit id={id} found={}", + member.is_some() + ); + Ok(member) + }) +} + +/// List all members of a team, by creation order. +pub fn list_agent_team_members( + workspace_dir: &Path, + team_id: &str, +) -> Result> { + tinyagents_tracing::debug!("{LOG_PREFIX} list_agent_team_members.entry team={team_id}"); + crate::store::with_connection(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + let mut stmt = conn.prepare( + "SELECT id, team_id, name, agent_id, member_status, + current_task_id, worker_thread_id, run_id, created_at, updated_at + FROM agent_team_members WHERE team_id = ?1 + ORDER BY created_at ASC", + )?; + let rows = stmt.query_map(params![team_id], map_agent_team_member_row)?; + let mut members = Vec::new(); + for row in rows { + members.push(row?); + } + tinyagents_tracing::debug!( + "{LOG_PREFIX} list_agent_team_members.exit team={team_id} count={}", + members.len() + ); + Ok(members) + }) +} + +/// Insert or update a team task. +pub fn upsert_agent_team_task( + workspace_dir: &Path, + upsert: AgentTeamTaskUpsert, +) -> Result { + let now = Utc::now(); + let created_at = upsert.created_at.unwrap_or(now); + let depends_on_json = + serde_json::to_string(&upsert.depends_on).storage_context("serialize task depends_on")?; + let evidence_json = + serde_json::to_string(&upsert.evidence).storage_context("serialize task evidence")?; + let gate_status = upsert.gate_status.unwrap_or_else(|| "pending".to_string()); + tinyagents_tracing::debug!( + "{LOG_PREFIX} upsert_agent_team_task.entry id={} team={} status={} deps={}", + upsert.id, + upsert.team_id, + upsert.status.as_str(), + upsert.depends_on.len() + ); + // One transaction for the write *and* the read-back. Committing the insert + // on an autocommit connection, closing it, then re-opening to `get_*` hands + // the caller whatever a concurrent writer left behind rather than what this + // call wrote — an upsert that reports someone else's row. + let task = crate::store::with_transaction(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + conn.execute( + "INSERT INTO agent_team_tasks ( + id, team_id, title, objective, status, owner_member_id, + claimed_by_member_id, claim_token, depends_on_json, gate_status, + gate_reason, evidence_json, source_run_id, order_index, + created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, NULL, NULL, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14) + ON CONFLICT(id) DO UPDATE SET + title = excluded.title, + objective = COALESCE(excluded.objective, agent_team_tasks.objective), + status = excluded.status, + -- A claim only means anything while the task is in_progress. + -- Editing a claimed task through this upsert (to change + -- dependencies, or to reset status during recovery) used to + -- leave claimed_by_member_id/claim_token set on a todo row, + -- which strands it: a fresh claim returns AlreadyClaimed, + -- completion returns NotClaimed, and release/shutdown skip it + -- because they only match in_progress. Drop the claim whenever + -- the new status is not in_progress; preserve it otherwise so + -- an unrelated edit does not steal a live claim. + claimed_by_member_id = CASE + WHEN excluded.status = 'in_progress' + THEN agent_team_tasks.claimed_by_member_id ELSE NULL END, + claim_token = CASE + WHEN excluded.status = 'in_progress' + THEN agent_team_tasks.claim_token ELSE NULL END, + owner_member_id = COALESCE(excluded.owner_member_id, agent_team_tasks.owner_member_id), + depends_on_json = excluded.depends_on_json, + gate_status = excluded.gate_status, + gate_reason = COALESCE(excluded.gate_reason, agent_team_tasks.gate_reason), + evidence_json = excluded.evidence_json, + source_run_id = COALESCE(excluded.source_run_id, agent_team_tasks.source_run_id), + order_index = excluded.order_index, + updated_at = excluded.updated_at", + params![ + upsert.id, + upsert.team_id, + upsert.title, + upsert.objective, + upsert.status.as_str(), + upsert.owner_member_id, + depends_on_json, + gate_status, + upsert.gate_reason, + evidence_json, + upsert.source_run_id, + upsert.order_index, + created_at.to_rfc3339(), + now.to_rfc3339(), + ], + ) + .storage_context("upsert agent team task")?; + get_agent_team_task_inner(conn, &upsert.id)? + .storage_context("agent team task missing after upsert") + })?; + tinyagents_tracing::debug!("{LOG_PREFIX} upsert_agent_team_task.exit id={}", task.id); + Ok(task) +} + +/// Fetch a single task by id. +pub fn get_agent_team_task(workspace_dir: &Path, id: &str) -> Result> { + tinyagents_tracing::debug!("{LOG_PREFIX} get_agent_team_task.entry id={id}"); + crate::store::with_connection(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + let task = get_agent_team_task_inner(conn, id)?; + tinyagents_tracing::debug!( + "{LOG_PREFIX} get_agent_team_task.exit id={id} found={}", + task.is_some() + ); + Ok(task) + }) +} + +/// List all tasks of a team, by `order_index` then creation order. +pub fn list_agent_team_tasks(workspace_dir: &Path, team_id: &str) -> Result> { + tinyagents_tracing::debug!("{LOG_PREFIX} list_agent_team_tasks.entry team={team_id}"); + crate::store::with_connection(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + let mut stmt = conn.prepare( + "SELECT id, team_id, title, objective, status, owner_member_id, + claimed_by_member_id, claim_token, depends_on_json, gate_status, + gate_reason, evidence_json, source_run_id, order_index, + created_at, updated_at + FROM agent_team_tasks WHERE team_id = ?1 + ORDER BY order_index ASC, created_at ASC", + )?; + let rows = stmt.query_map(params![team_id], map_agent_team_task_row)?; + let mut tasks = Vec::new(); + for row in rows { + tasks.push(row?); + } + tinyagents_tracing::debug!( + "{LOG_PREFIX} list_agent_team_tasks.exit team={team_id} count={}", + tasks.len() + ); + Ok(tasks) + }) +} + +/// Atomically claim a task for a member. +/// +/// All steps run inside a single `with_connection` transaction so that the +/// dependency check and the compare-and-swap observe a consistent snapshot: +/// 1. Resolve the task by `(id, team_id)`; absent → [`ClaimOutcome::UnknownTask`]. +/// 2. For every dependency id, look up its status; collect those not `done` +/// into `unmet`. Non-empty → [`ClaimOutcome::Blocked`]. +/// 3. WHERE-guarded `UPDATE ... WHERE claimed_by_member_id IS NULL`: SQLite +/// serializes writers, so exactly one concurrent claimer flips the row from +/// unclaimed to claimed. `rows_affected == 0` → already taken +/// ([`ClaimOutcome::AlreadyClaimed`]); otherwise re-fetch and return +/// [`ClaimOutcome::Claimed`]. +pub fn claim_agent_team_task( + workspace_dir: &Path, + team_id: &str, + task_id: &str, + member_id: &str, + claim_token: &str, +) -> Result { + tinyagents_tracing::debug!( + "{LOG_PREFIX} claim_agent_team_task.entry team={team_id} task={task_id} member={member_id}" + ); + let outcome = crate::store::with_transaction(workspace_dir, |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, + _ => { + tinyagents_tracing::debug!( + "{LOG_PREFIX} claim_agent_team_task.unknown team={team_id} task={task_id}" + ); + return Ok(ClaimOutcome::UnknownTask); + } + }; + + // 2. Dependency gate: every dep must be `done`. + let mut unmet = Vec::new(); + 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()?; + let is_done = dep_status.as_deref() == Some(AgentTeamTaskStatus::Done.as_str()); + if !is_done { + unmet.push(dep_id.clone()); + } + } + if !unmet.is_empty() { + tinyagents_tracing::debug!( + "{LOG_PREFIX} claim_agent_team_task.blocked team={team_id} task={task_id} unmet={}", + unmet.len() + ); + return Ok(ClaimOutcome::Blocked { unmet }); + } + + // 3. Compare-and-swap on the unclaimed guard **and the status**. + // + // `claimed_by_member_id IS NULL` alone is not a guard: `upsert_agent_team_task` + // deliberately NULLs that column whenever the new status is not + // `in_progress`, so every `done` task also satisfies it. A stale worker + // re-claiming a finished task therefore flipped it straight back to + // `in_progress` — and stranded everything downstream, because the + // completion gate re-checks that each dependency is still `done` and now + // reports it unfinished. A terminal task is not claimable, whatever its + // claim column says. + let now = Utc::now(); + let rows_affected = conn + .execute( + "UPDATE agent_team_tasks + SET claimed_by_member_id = ?1, claim_token = ?2, status = 'in_progress', updated_at = ?3 + WHERE id = ?4 AND team_id = ?5 AND claimed_by_member_id IS NULL + AND status IN ('todo', 'ready', 'blocked')", + params![member_id, claim_token, now.to_rfc3339(), task_id, team_id], + ) + .storage_context("compare-and-swap claim agent team task")?; + if rows_affected == 0 { + tinyagents_tracing::debug!( + "{LOG_PREFIX} claim_agent_team_task.already_claimed team={team_id} task={task_id} \ + status={}", + task.status.as_str() + ); + return Ok(ClaimOutcome::AlreadyClaimed); + } + + let claimed = get_agent_team_task_inner(conn, task_id)? + .storage_context("claimed task missing after compare-and-swap")?; + Ok(ClaimOutcome::Claimed(Box::new(claimed))) + })?; + tinyagents_tracing::debug!( + "{LOG_PREFIX} claim_agent_team_task.exit team={team_id} task={task_id} outcome={}", + match &outcome { + ClaimOutcome::Claimed(_) => "claimed", + ClaimOutcome::AlreadyClaimed => "already_claimed", + ClaimOutcome::Blocked { .. } => "blocked", + ClaimOutcome::UnknownTask => "unknown", + } + ); + 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( + workspace_dir: &Path, + team_id: &str, + task_id: &str, + member_id: &str, + evidence: &[String], + require_evidence: bool, +) -> Result { + tinyagents_tracing::debug!( + "{LOG_PREFIX} complete_agent_team_task.entry team={team_id} task={task_id} member={member_id}" + ); + let outcome = crate::store::with_transaction(workspace_dir, |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, + _ => { + tinyagents_tracing::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 { + tinyagents_tracing::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("; "); + // Persist the merged evidence even though the gate failed. Evidence + // accumulates across attempts (see `merged_evidence` above), so + // dropping it here punished a caller for an unrelated gate failure: + // after a dependency was fixed, a retry that did not resend the same + // links would fail `require_evidence` on evidence it had already + // submitted. Only the gate verdict is a failure; the submission is + // still real. + let evidence_json = serde_json::to_string(&merged_evidence) + .storage_context("serialize completion evidence")?; + conn.execute( + "UPDATE agent_team_tasks + SET gate_status = 'failed', gate_reason = ?1, + evidence_json = ?2, updated_at = ?3 + WHERE id = ?4 AND team_id = ?5", + params![joined, evidence_json, now.to_rfc3339(), task_id, team_id], + ) + .storage_context("record failed completion gate")?; + tinyagents_tracing::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) + .storage_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], + ) + .storage_context("complete agent team task")?; + if rows_affected == 0 { + tinyagents_tracing::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)? + .storage_context("completed task missing after update")?; + Ok(CompletionOutcome::Completed(Box::new(done))) + })?; + tinyagents_tracing::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 + .as_ref() + .filter(|owner| 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( + workspace_dir: &Path, + team_id: &str, + member_id: &str, +) -> Result)>> { + tinyagents_tracing::debug!( + "{LOG_PREFIX} shutdown_agent_team_member.entry team={team_id} member={member_id}" + ); + let result = crate::store::with_transaction(workspace_dir, |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 => {} + _ => { + tinyagents_tracing::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], + ) + .storage_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], + ) + .storage_context("stop agent team member")?; + + let member = get_agent_team_member_inner(conn, member_id)? + .storage_context("member missing after shutdown")?; + Ok(Some((member, released))) + })?; + tinyagents_tracing::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) +} + +/// Mark a member as actively running a task: status → `active`, with the +/// current task id and the worker/run identifiers of the spawned agent. Used by +/// the live runtime right after it claims a task and dispatches a worker. +/// Returns the updated member, or `None` if the member is not in the team. +pub fn mark_agent_team_member_running( + workspace_dir: &Path, + team_id: &str, + member_id: &str, + task_id: &str, + worker_thread_id: &str, + run_id: &str, +) -> Result> { + tinyagents_tracing::debug!( + "{LOG_PREFIX} mark_agent_team_member_running.entry team={team_id} member={member_id} task={task_id} run={run_id}" + ); + crate::store::with_connection(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + let now = Utc::now(); + let changed = conn + .execute( + "UPDATE agent_team_members + SET member_status = 'active', current_task_id = ?1, + worker_thread_id = ?2, run_id = ?3, updated_at = ?4 + WHERE id = ?5 AND team_id = ?6", + params![ + task_id, + worker_thread_id, + run_id, + now.to_rfc3339(), + member_id, + team_id + ], + ) + .storage_context("mark agent team member running")?; + if changed == 0 { + return Ok(None); + } + get_agent_team_member_inner(conn, member_id) + }) +} + +/// Mark a member idle: status → `idle`, clearing `current_task_id`. The +/// `worker_thread_id` / `run_id` are intentionally retained as a pointer to the +/// member's last run for history. Returns the updated member, or `None` if the +/// member is not in the team. Used when a worker run finishes (completed, +/// gate-failed, or failed) so the member is free to pick up new work. +pub fn mark_agent_team_member_idle( + workspace_dir: &Path, + team_id: &str, + member_id: &str, +) -> Result> { + tinyagents_tracing::debug!( + "{LOG_PREFIX} mark_agent_team_member_idle.entry team={team_id} member={member_id}" + ); + crate::store::with_connection(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + let now = Utc::now(); + let changed = conn + .execute( + "UPDATE agent_team_members + SET member_status = 'idle', current_task_id = NULL, updated_at = ?1 + WHERE id = ?2 AND team_id = ?3", + params![now.to_rfc3339(), member_id, team_id], + ) + .storage_context("mark agent team member idle")?; + if changed == 0 { + return Ok(None); + } + get_agent_team_member_inner(conn, member_id) + }) +} + +/// Release a single `in_progress` task back to `todo`, clearing its claim and +/// resetting the quality gate. Returns `true` if a row was actually released +/// (the task existed, belonged to the team, and was `in_progress`). Used by the +/// live runtime when a worker run fails or is aborted, so the task is free for +/// another teammate — the per-task analogue of the bulk release in +/// `shutdown_agent_team_member`. +pub fn release_agent_team_task(workspace_dir: &Path, team_id: &str, task_id: &str) -> Result { + tinyagents_tracing::debug!( + "{LOG_PREFIX} release_agent_team_task.entry team={team_id} task={task_id}" + ); + crate::store::with_connection(workspace_dir, |conn| { + init_run_ledger_schema(conn)?; + let now = Utc::now(); + let changed = conn + .execute( + "UPDATE agent_team_tasks + SET status = 'todo', claimed_by_member_id = NULL, claim_token = NULL, + gate_status = 'pending', gate_reason = NULL, updated_at = ?1 + WHERE id = ?2 AND team_id = ?3 AND status = 'in_progress'", + params![now.to_rfc3339(), task_id, team_id], + ) + .storage_context("release agent team task")?; + tinyagents_tracing::debug!( + "{LOG_PREFIX} release_agent_team_task.exit team={team_id} task={task_id} released={}", + changed > 0 + ); + Ok(changed > 0) + }) +} From e73fce9ba8b498acda384ca0eaaa05e8b20ada25 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 20 Sep 2026 01:40:35 +0300 Subject: [PATCH 2/5] Update TinyInference dependency cleanup Co-authored-by: Medulla --- vendor/tinyinference | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyinference b/vendor/tinyinference index 92445ea58..1e8ab6590 160000 --- a/vendor/tinyinference +++ b/vendor/tinyinference @@ -1 +1 @@ -Subproject commit 92445ea582c9b1c5dc78891ebbff4a817f9e8fa9 +Subproject commit 1e8ab659023d8040a2a8cb082ce367ca0ba99880 From 2a6b17403f68ad3d5c1a2c6bef4f6ae52ec24419 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 20 Sep 2026 01:40:49 +0300 Subject: [PATCH 3/5] Update TinyInference lockfile revision Co-authored-by: Medulla --- vendor/tinyinference | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/tinyinference b/vendor/tinyinference index 1e8ab6590..d67097cbf 160000 --- a/vendor/tinyinference +++ b/vendor/tinyinference @@ -1 +1 @@ -Subproject commit 1e8ab659023d8040a2a8cb082ce367ca0ba99880 +Subproject commit d67097cbf26a0e6bc857e4f65ef0ad1f46df04e3 From dc6c0e253a19410c0175477949c96a0397176b60 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 20 Sep 2026 01:53:32 +0300 Subject: [PATCH 4/5] fix: remove stale run ledger import Co-authored-by: Medulla --- crates/tinyagents-session/src/run_ledger/ops.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tinyagents-session/src/run_ledger/ops.rs b/crates/tinyagents-session/src/run_ledger/ops.rs index de29a8bd9..2536586a6 100644 --- a/crates/tinyagents-session/src/run_ledger/ops.rs +++ b/crates/tinyagents-session/src/run_ledger/ops.rs @@ -41,7 +41,7 @@ use rows::{ get_agent_run_inner, get_agent_team_inner, get_agent_team_member_inner, get_agent_team_task_inner, get_run_telemetry_inner, map_agent_run_row, map_agent_team_member_row, map_agent_team_row, map_agent_team_task_row, map_run_event_row, - map_run_telemetry_row, map_workflow_run_row, + map_workflow_run_row, }; pub use team::{ From d2ae2b7f3a25d3d45b1ad59a31a37ec6651e8f54 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Sun, 20 Sep 2026 01:59:00 +0300 Subject: [PATCH 5/5] refactor: make run ledger submodule imports explicit Co-authored-by: Medulla --- .../tinyagents-session/src/run_ledger/ops.rs | 17 ++++++---------- .../src/run_ledger/ops/rows.rs | 13 +++++++++++- .../src/run_ledger/ops/team.rs | 20 ++++++++++++++++++- .../tinyagents-session/src/run_ledger/test.rs | 20 +++++++++++++++++++ .../src/run_ledger/types.rs | 2 +- 5 files changed, 58 insertions(+), 14 deletions(-) diff --git a/crates/tinyagents-session/src/run_ledger/ops.rs b/crates/tinyagents-session/src/run_ledger/ops.rs index 2536586a6..a0cd5e250 100644 --- a/crates/tinyagents-session/src/run_ledger/ops.rs +++ b/crates/tinyagents-session/src/run_ledger/ops.rs @@ -18,29 +18,24 @@ use std::path::Path; use chrono::{DateTime, Utc}; use rusqlite::{Connection, OptionalExtension, params}; -use serde_json::{Value, json}; +use serde_json::json; use tinyagents_harness::error::Result; use super::super::context::StorageContext; use super::store::init_run_ledger_schema; use super::types::{ - AgentRun, AgentRunKind, AgentRunListRequest, AgentRunListResponse, AgentRunStatus, - AgentRunUpsert, AgentTeam, AgentTeamListRequest, AgentTeamListResponse, AgentTeamMember, - AgentTeamMemberStatus, AgentTeamMemberUpsert, AgentTeamStatus, AgentTeamTask, - AgentTeamTaskStatus, AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, CompletionOutcome, - RunEvent, RunEventAppend, RunEventListRequest, RunEventListResponse, RunTelemetry, - RunTelemetryUpsert, WorkflowLeaseClaim, WorkflowRun, WorkflowRunListRequest, - WorkflowRunListResponse, WorkflowRunStatus, WorkflowRunUpsert, + AgentRun, AgentRunListRequest, AgentRunListResponse, AgentRunStatus, AgentRunUpsert, RunEvent, + RunEventAppend, RunEventListRequest, RunEventListResponse, RunTelemetry, RunTelemetryUpsert, + WorkflowLeaseClaim, WorkflowRun, WorkflowRunListRequest, WorkflowRunListResponse, + WorkflowRunUpsert, }; mod rows; mod team; use rows::{ - get_agent_run_inner, get_agent_team_inner, get_agent_team_member_inner, - get_agent_team_task_inner, get_run_telemetry_inner, map_agent_run_row, - map_agent_team_member_row, map_agent_team_row, map_agent_team_task_row, map_run_event_row, + get_agent_run_inner, get_run_telemetry_inner, map_agent_run_row, map_run_event_row, map_workflow_run_row, }; diff --git a/crates/tinyagents-session/src/run_ledger/ops/rows.rs b/crates/tinyagents-session/src/run_ledger/ops/rows.rs index e464364a6..ed49915ba 100644 --- a/crates/tinyagents-session/src/run_ledger/ops/rows.rs +++ b/crates/tinyagents-session/src/run_ledger/ops/rows.rs @@ -3,7 +3,18 @@ //! Keeping this detail outside the public ledger operations makes the //! operation module focused on state transitions and query semantics. -use super::*; +use chrono::{DateTime, Utc}; +use rusqlite::{Connection, OptionalExtension, params}; +use serde_json::{Value, json}; + +use tinyagents_harness::error::Result; + +use super::super::types::{ + AgentRun, AgentRunKind, AgentRunStatus, AgentTeam, AgentTeamMember, AgentTeamMemberStatus, + AgentTeamStatus, AgentTeamTask, AgentTeamTaskStatus, RunEvent, RunTelemetry, WorkflowRun, + WorkflowRunStatus, +}; +use crate::context::StorageContext; /// Connection-scoped team lookup, so an upsert can read its own write back /// inside the same transaction. diff --git a/crates/tinyagents-session/src/run_ledger/ops/team.rs b/crates/tinyagents-session/src/run_ledger/ops/team.rs index 4c9373b8f..115ab2223 100644 --- a/crates/tinyagents-session/src/run_ledger/ops/team.rs +++ b/crates/tinyagents-session/src/run_ledger/ops/team.rs @@ -3,7 +3,25 @@ //! Teams, members, and tasks share the run-ledger database but have distinct //! lifecycle and claim semantics from agent and workflow runs. -use super::*; +use std::path::Path; + +use chrono::Utc; +use rusqlite::{Connection, OptionalExtension, params}; + +use tinyagents_harness::error::Result; + +use super::super::store::init_run_ledger_schema; +use super::super::types::{ + AgentTeam, AgentTeamListRequest, AgentTeamListResponse, AgentTeamMember, AgentTeamMemberUpsert, + AgentTeamTask, AgentTeamTaskStatus, AgentTeamTaskUpsert, AgentTeamUpsert, ClaimOutcome, + CompletionOutcome, +}; +use super::LOG_PREFIX; +use super::rows::{ + get_agent_team_inner, get_agent_team_member_inner, get_agent_team_task_inner, + map_agent_team_member_row, map_agent_team_row, map_agent_team_task_row, +}; +use crate::context::StorageContext; // --------------------------------------------------------------------------- // Agent-team coordination (issue #3374) diff --git a/crates/tinyagents-session/src/run_ledger/test.rs b/crates/tinyagents-session/src/run_ledger/test.rs index 24770e4e2..fbd343c7c 100644 --- a/crates/tinyagents-session/src/run_ledger/test.rs +++ b/crates/tinyagents-session/src/run_ledger/test.rs @@ -852,6 +852,26 @@ fn team_members_and_tasks_list_back() { assert_eq!(teams.count, 1); } +#[test] +fn team_list_count_is_total_before_pagination() { + let dir = TempDir::new().unwrap(); + let workspace_dir = test_workspace(&dir); + seed_team(workspace_dir, "team-1"); + seed_team(workspace_dir, "team-2"); + + let teams = list_agent_teams( + workspace_dir, + &AgentTeamListRequest { + limit: Some(1), + ..Default::default() + }, + ) + .unwrap(); + + assert_eq!(teams.teams.len(), 1); + assert_eq!(teams.count, 2); +} + fn seed_run(workspace_dir: &Path, id: &str, status: AgentRunStatus) { upsert_agent_run( workspace_dir, diff --git a/crates/tinyagents-session/src/run_ledger/types.rs b/crates/tinyagents-session/src/run_ledger/types.rs index 402fd1acb..1acd45a75 100644 --- a/crates/tinyagents-session/src/run_ledger/types.rs +++ b/crates/tinyagents-session/src/run_ledger/types.rs @@ -729,7 +729,7 @@ pub struct AgentTeamListRequest { #[serde(rename_all = "camelCase")] pub struct AgentTeamListResponse { pub teams: Vec, - /// Number of teams in `teams` (not the total matching count). + /// Total number of teams matching the request filters, before pagination. pub count: usize, }