From abdc69490211e03fcb37b397b9e03a3b274dee86 Mon Sep 17 00:00:00 2001 From: Shibo Sheng Date: Sun, 2 Aug 2026 23:08:48 +0800 Subject: [PATCH] fix(agent-org): guard runtime submissions during deletion --- .../core/coordination/agent_inbox/message.rs | 14 + .../agent_member_interventions.rs | 139 +++- .../coordination/agent_org_runs/deletion.rs | 473 ++++++++++++++ .../agent_org_runs/deletion_tests.rs | 602 ++++++++++++++++++ .../coordination/agent_org_runs/migration.rs | 7 +- .../core/coordination/agent_org_runs/mod.rs | 26 + .../core/coordination/agent_org_runs/store.rs | 147 +++-- .../core/coordination/agent_org_tasks/mod.rs | 13 +- .../agent_org_watchdog/inspect.rs | 4 +- .../agent_org_watchdog/recover.rs | 77 ++- .../agent_org_watchdog/reservation.rs | 22 +- .../coordination/agent_org_watchdog/tests.rs | 171 ++++- .../src/core/session/gateway_pipeline.rs | 6 + .../agent-core/src/core/session/launch/mod.rs | 22 + .../agent-core/src/core/session/scheduler.rs | 74 +++ .../agent-core/src/core/session/turn/entry.rs | 61 ++ .../turn/processor/inbox_drain/tests.rs | 8 +- src-tauri/crates/agent-core/src/init/mod.rs | 80 ++- .../agent-core/src/init/runtime_assemble.rs | 10 +- .../src/state/commands/session/compaction.rs | 1 + .../src/state/commands/session/identity.rs | 47 ++ .../commands/session/message/org_wake.rs | 10 +- .../state/commands/session/message/send.rs | 32 +- .../commands/session/org_tasks/lifecycle.rs | 20 +- .../commands/session/org_tasks/run_view.rs | 15 +- .../state/commands/session/org_tasks/tests.rs | 60 +- .../src/state/commands/session/persistence.rs | 178 ++++-- .../agent-core/src/state/session_runtime.rs | 26 +- src-tauri/crates/e2e-test/src/agent_org.rs | 191 ++++++ src-tauri/crates/e2e-test/src/main.rs | 10 + src-tauri/src/api/agent/mod.rs | 127 ++++ 31 files changed, 2480 insertions(+), 193 deletions(-) create mode 100644 src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/deletion.rs create mode 100644 src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/deletion_tests.rs diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_inbox/message.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_inbox/message.rs index 3f79f5270..0e186d4c5 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_inbox/message.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_inbox/message.rs @@ -708,6 +708,13 @@ mod tests { CREATE TABLE IF NOT EXISTS agent_org_tasks ( id TEXT PRIMARY KEY, org_run_id TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS agent_org_run_sessions ( + org_run_id TEXT NOT NULL, + member_id TEXT NOT NULL, + session_id TEXT NOT NULL, + role TEXT NOT NULL, + created_at TEXT NOT NULL );", ) .expect("initialize minimal delivery-repair dependencies"); @@ -1560,6 +1567,13 @@ mod tests { ], ) .expect("seed healthy replacement member"); + conn.execute( + "INSERT INTO agent_org_run_sessions ( + org_run_id, member_id, session_id, role, created_at + ) VALUES (?1, ?2, ?3, 'worker', ?4)", + params![run_id, member_id, session_id, &now], + ) + .expect("seed exact replacement-member ownership"); } let row_a = AgentInboxStore::insert(InsertInboxParams { diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_member_interventions.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_member_interventions.rs index 0a029657e..8036d2efc 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_member_interventions.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_member_interventions.rs @@ -9,7 +9,7 @@ use serde::Serialize; use database::db::{get_connection, with_sessions_writer}; -use super::agent_org_runs::COORDINATOR_MEMBER_ID; +use super::agent_org_runs::{is_run_writable_with_connection, COORDINATOR_MEMBER_ID}; pub const DEFAULT_INTERVENTION_TTL_SECS: i64 = 180; @@ -110,9 +110,13 @@ impl AgentMemberInterventionStore { let resume_after = (now + chrono::Duration::seconds(ttl_secs)).to_rfc3339(); let status = MemberInterventionStatus::UserIntervention; - with_sessions_writer(|| -> Result<(), String> { - let conn = get_connection().map_err(|err| err.to_string())?; - conn.execute( + let record = with_sessions_writer(|| -> Result { + let mut conn = get_connection().map_err(|err| err.to_string())?; + let tx = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|err| err.to_string())?; + ensure_run_writable_for_intervention(&tx, ¶ms.org_run_id)?; + tx.execute( "INSERT INTO agent_member_interventions ( org_run_id, member_id, @@ -146,14 +150,26 @@ impl AgentMemberInterventionStore { ], ) .map_err(|err| err.to_string())?; - Ok(()) - })?; - - let record = Self::get(¶ms.org_run_id, ¶ms.member_id)?.ok_or_else(|| { - format!( - "agent_member_interventions upsert did not return row for run={} member={}", - params.org_run_id, params.member_id - ) + let record = tx + .query_row( + "SELECT org_run_id, + member_id, + agent_id, + session_id, + status, + reason, + entered_at, + last_user_activity_at, + resume_after, + cleared_at + FROM agent_member_interventions + WHERE org_run_id = ?1 AND member_id = ?2", + params![params.org_run_id, params.member_id], + row_to_intervention, + ) + .map_err(|err| err.to_string())?; + tx.commit().map_err(|err| err.to_string())?; + Ok(record) })?; crate::coordination::agent_org_run_events::notify_agent_org_run_changed(&record.org_run_id); Ok(record) @@ -195,6 +211,7 @@ impl AgentMemberInterventionStore { let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; + ensure_run_writable_for_intervention(&tx, org_run_id)?; let updated = tx .execute( "UPDATE agent_member_interventions @@ -358,6 +375,15 @@ impl AgentMemberInterventionStore { } } +fn ensure_run_writable_for_intervention(conn: &Connection, run_id: &str) -> Result<(), String> { + if is_run_writable_with_connection(conn, run_id)? { + return Ok(()); + } + Err(format!( + "agent_org_run_not_writable: run {run_id} cannot accept member intervention work" + )) +} + fn resume_after_is_future(value: &str) -> bool { chrono::DateTime::parse_from_rfc3339(value) .map(|timestamp| timestamp.with_timezone(&chrono::Utc) > chrono::Utc::now()) @@ -394,12 +420,36 @@ mod tests { fn setup() -> test_helpers::test_env::SandboxGuard { let sandbox = test_helpers::test_env::sandbox(); let conn = get_connection().expect("db connection"); + crate::coordination::agent_org_runs::init_schema(&conn).expect("run schema"); + crate::coordination::agent_inbox::init_schema(&conn).expect("inbox schema"); init_schema(&conn).expect("schema"); conn.execute("DELETE FROM agent_member_interventions", []) .expect("clear"); + let now = chrono::Utc::now().to_rfc3339(); + conn.execute( + "INSERT INTO agent_org_runs ( + id, org_id, coordinator_agent_id, root_session_id, entry_mode, + status, created_at, updated_at + ) VALUES ('run-1', 'org-1', 'agent-coordinator', 'root-1', + 'standalone_session', 'running', ?1, ?1)", + [&now], + ) + .expect("seed writable run"); sandbox } + fn fence_run_without_changing_status(run_id: &str) { + let conn = get_connection().expect("db connection"); + let now = chrono::Utc::now().to_rfc3339(); + conn.execute( + "INSERT INTO agent_org_conversation_delete_fences ( + root_session_id, created_at, updated_at + ) SELECT root_session_id, ?2, ?2 FROM agent_org_runs WHERE id=?1", + params![run_id, now], + ) + .expect("fence run"); + } + #[test] fn enter_upserts_active_record_by_member_id() { let _sandbox = setup(); @@ -449,6 +499,71 @@ mod tests { assert!(error.contains("cannot enter member intervention")); } + #[test] + fn enter_rejects_fenced_non_running_and_missing_runs_without_orphans() { + let _sandbox = setup(); + let params_for = |run_id: &str, member_id: &str| EnterMemberInterventionParams { + org_run_id: run_id.to_string(), + member_id: member_id.to_string(), + agent_id: "agent-a".into(), + session_id: format!("session-{member_id}"), + reason: Some("user".into()), + ttl_secs: 60, + }; + + fence_run_without_changing_status("run-1"); + AgentMemberInterventionStore::enter(params_for("run-1", "member-fenced")) + .expect_err("fenced Run must reject intervention entry"); + + let conn = get_connection().expect("db connection"); + conn.execute( + "DELETE FROM agent_org_conversation_delete_fences WHERE root_session_id='root-1'", + [], + ) + .expect("remove test fence"); + conn.execute( + "UPDATE agent_org_runs SET status='paused' WHERE id='run-1'", + [], + ) + .expect("pause run"); + AgentMemberInterventionStore::enter(params_for("run-1", "member-paused")) + .expect_err("non-running Run must reject intervention entry"); + AgentMemberInterventionStore::enter(params_for("missing-run", "member-missing")) + .expect_err("missing Run must reject intervention entry"); + + let count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_member_interventions", + [], + |row| row.get(0), + ) + .expect("count intervention rows"); + assert_eq!(count, 0); + } + + #[test] + fn unread_boundary_does_not_release_a_fenced_intervention() { + let _sandbox = setup(); + AgentMemberInterventionStore::enter(EnterMemberInterventionParams { + org_run_id: "run-1".into(), + member_id: "member-a".into(), + agent_id: "agent-a".into(), + session_id: "session-a".into(), + reason: Some("user".into()), + ttl_secs: 60, + }) + .expect("enter before fence"); + fence_run_without_changing_status("run-1"); + + AgentMemberInterventionStore::clear_and_capture_unread_boundary("run-1", "member-a") + .expect_err("fenced Run must not be released into autonomous work"); + assert!(AgentMemberInterventionStore::get("run-1", "member-a") + .expect("load retained intervention") + .expect("intervention remains") + .cleared_at + .is_none()); + } + #[test] fn legacy_coordinator_intervention_is_hidden_without_mutating_on_read() { let _sandbox = setup(); diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/deletion.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/deletion.rs new file mode 100644 index 000000000..3410f4be0 --- /dev/null +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/deletion.rs @@ -0,0 +1,473 @@ +//! Durable Root deletion fencing and runtime-submission admission. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU8, Ordering}; +use std::sync::{Arc, Mutex, OnceLock}; + +use parking_lot::RwLock; +use rusqlite::{params, Connection, OptionalExtension}; + +use database::db::{get_connection, with_sessions_writer}; + +use super::AgentOrgRunStatus; + +pub(super) const CONVERSATION_DELETING_ERROR_CODE: &str = "conversation_deleting"; + +/// `Unknown` is resolved once from PR1's exact mapping. Fence state is never cached. +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum AgentOrgSubmissionScope { + Unknown, + Ordinary, + Run { run_id: String }, +} + +const SCOPE_UNKNOWN: u8 = 0; +const SCOPE_ORDINARY: u8 = 1; +const SCOPE_RUN: u8 = 2; + +/// Shared identity policy whose ordinary hot path is one atomic read. +pub(crate) struct AgentOrgSubmissionPolicy { + kind: AtomicU8, + run_id: RwLock>, + resolve_lock: tokio::sync::Mutex<()>, +} + +impl AgentOrgSubmissionPolicy { + pub(crate) fn new(scope: AgentOrgSubmissionScope) -> Self { + let policy = Self { + kind: AtomicU8::new(SCOPE_UNKNOWN), + run_id: RwLock::new(None), + resolve_lock: tokio::sync::Mutex::new(()), + }; + policy.store(scope); + policy + } + + pub(crate) fn snapshot(&self) -> AgentOrgSubmissionScope { + match self.kind.load(Ordering::Acquire) { + SCOPE_ORDINARY => AgentOrgSubmissionScope::Ordinary, + SCOPE_RUN => AgentOrgSubmissionScope::Run { + run_id: self + .run_id + .read() + .clone() + .expect("Run scope requires Run id"), + }, + _ => AgentOrgSubmissionScope::Unknown, + } + } + + pub(crate) fn store(&self, scope: AgentOrgSubmissionScope) { + if self.kind.load(Ordering::Acquire) == SCOPE_RUN + && !matches!(&scope, AgentOrgSubmissionScope::Run { .. }) + { + return; + } + let (kind, run_id) = match scope { + AgentOrgSubmissionScope::Unknown => (SCOPE_UNKNOWN, None), + AgentOrgSubmissionScope::Ordinary => (SCOPE_ORDINARY, None), + AgentOrgSubmissionScope::Run { run_id } => (SCOPE_RUN, Some(run_id)), + }; + *self.run_id.write() = run_id; + self.kind.store(kind, Ordering::Release); + } +} + +pub(crate) type SharedAgentOrgSubmissionScope = Arc; + +static ACTIVE_AGENT_ORG_SUBMISSIONS: OnceLock>> = OnceLock::new(); + +#[cfg(any(test, debug_assertions))] +static AGENT_ORG_SUBMISSION_QUERIES: std::sync::atomic::AtomicUsize = + std::sync::atomic::AtomicUsize::new(0); +#[cfg(any(test, debug_assertions))] +static AGENT_ORG_LEASE_MUTATIONS: std::sync::atomic::AtomicUsize = + std::sync::atomic::AtomicUsize::new(0); + +/// In-memory lifetime guard used only by known Agent Org sessions. +#[derive(Debug)] +pub(crate) struct AgentOrgSubmissionLease(String); + +impl AgentOrgSubmissionLease { + pub(crate) fn begin(session_id: &str) -> Self { + let mut active = ACTIVE_AGENT_ORG_SUBMISSIONS + .get_or_init(|| Mutex::new(HashMap::new())) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *active.entry(session_id.to_string()).or_default() += 1; + #[cfg(any(test, debug_assertions))] + AGENT_ORG_LEASE_MUTATIONS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + Self(session_id.to_string()) + } +} + +impl Drop for AgentOrgSubmissionLease { + fn drop(&mut self) { + let Some(active) = ACTIVE_AGENT_ORG_SUBMISSIONS.get() else { + return; + }; + let mut active = active + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(count) = active.get_mut(&self.0) { + *count -= 1; + if *count == 0 { + active.remove(&self.0); + } + #[cfg(any(test, debug_assertions))] + AGENT_ORG_LEASE_MUTATIONS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + } +} + +pub(crate) fn agent_org_submission_in_progress(session_id: &str) -> bool { + ACTIVE_AGENT_ORG_SUBMISSIONS + .get() + .and_then(|active| { + active + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .get(session_id) + .copied() + }) + .unwrap_or_default() + > 0 +} + +fn exact_submission_scope_with_connection( + conn: &Connection, + session_id: &str, +) -> Result { + #[cfg(any(test, debug_assertions))] + AGENT_ORG_SUBMISSION_QUERIES.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let mut stmt = conn + .prepare( + "SELECT mapping.org_run_id + FROM agent_org_run_sessions mapping + JOIN agent_org_runs run ON run.id=mapping.org_run_id + WHERE mapping.session_id=?1 + ORDER BY mapping.org_run_id + LIMIT 2", + ) + .map_err(|err| err.to_string())?; + let run_ids = stmt + .query_map([session_id], |row| row.get::<_, String>(0)) + .map_err(|err| err.to_string())? + .collect::, _>>() + .map_err(|err| err.to_string())?; + match run_ids.as_slice() { + [] => Ok(AgentOrgSubmissionScope::Ordinary), + [run_id] => Ok(AgentOrgSubmissionScope::Run { + run_id: run_id.clone(), + }), + _ => Err(format!( + "Agent Org submission ownership is ambiguous for session {session_id}" + )), + } +} + +pub(crate) fn exact_submission_scope(session_id: &str) -> Result { + let conn = get_connection().map_err(|err| err.to_string())?; + exact_submission_scope_with_connection(&conn, session_id) +} + +pub(crate) fn submission_scope_for_loaded_session( + session: &crate::session::persistence::UnifiedSessionRecord, +) -> Result { + let is_agent_org = session.session_type + == crate::session::persistence::session_type::ORG_MEMBER + || session.org_member_id.is_some(); + if !is_agent_org { + return Ok(AgentOrgSubmissionScope::Ordinary); + } + let conn = get_connection().map_err(|err| err.to_string())?; + let scope = exact_submission_scope_with_connection(&conn, &session.session_id)?; + if matches!(scope, AgentOrgSubmissionScope::Ordinary) { + return Err(format!( + "Agent Org session {} has no exact Run mapping", + session.session_id + )); + } + Ok(scope) +} + +async fn ensure_submission_scope_writable(scope: &AgentOrgSubmissionScope) -> Result<(), String> { + let AgentOrgSubmissionScope::Run { run_id } = scope else { + return Ok(()); + }; + let run_id = run_id.clone(); + #[cfg(any(test, debug_assertions))] + AGENT_ORG_SUBMISSION_QUERIES.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + tokio::task::spawn_blocking(move || { + let conn = get_connection().map_err(|err| err.to_string())?; + ensure_run_conversation_writable_with_connection(&conn, &run_id) + }) + .await + .map_err(|err| format!("Agent Org submission fence worker failed: {err}"))? +} + +/// Resolve transient unknown ownership once, then acquire a lease and read the +/// durable fence. The ordinary branch returns before either operation. +pub(crate) async fn admit_agent_org_submission( + scope: &SharedAgentOrgSubmissionScope, + session_id: &str, +) -> Result, String> { + let mut resolved = scope.snapshot(); + if resolved == AgentOrgSubmissionScope::Unknown { + let _resolver = scope.resolve_lock.lock().await; + resolved = scope.snapshot(); + if resolved == AgentOrgSubmissionScope::Unknown { + let session_id = session_id.to_string(); + resolved = tokio::task::spawn_blocking(move || { + let conn = get_connection().map_err(|err| err.to_string())?; + exact_submission_scope_with_connection(&conn, &session_id) + }) + .await + .map_err(|err| format!("Agent Org ownership worker failed: {err}"))??; + scope.store(resolved.clone()); + } + } + if resolved == AgentOrgSubmissionScope::Ordinary { + return Ok(None); + } + let lease = AgentOrgSubmissionLease::begin(session_id); + ensure_submission_scope_writable(&resolved).await?; + Ok(Some(lease)) +} + +pub(crate) async fn admit_known_agent_org_submission( + session_id: &str, + run_id: &str, +) -> Result { + let lease = AgentOrgSubmissionLease::begin(session_id); + recheck_known_agent_org_submission(run_id).await?; + Ok(lease) +} + +pub(crate) async fn recheck_agent_org_submission( + scope: &SharedAgentOrgSubmissionScope, +) -> Result<(), String> { + ensure_submission_scope_writable(&scope.snapshot()).await +} + +pub(crate) async fn recheck_known_agent_org_submission(run_id: &str) -> Result<(), String> { + ensure_submission_scope_writable(&AgentOrgSubmissionScope::Run { + run_id: run_id.to_string(), + }) + .await +} + +#[cfg(any(test, debug_assertions))] +pub fn reset_submission_metrics() { + AGENT_ORG_SUBMISSION_QUERIES.store(0, std::sync::atomic::Ordering::Relaxed); + AGENT_ORG_LEASE_MUTATIONS.store(0, std::sync::atomic::Ordering::Relaxed); +} + +#[cfg(any(test, debug_assertions))] +pub fn submission_metrics() -> (usize, usize) { + ( + AGENT_ORG_SUBMISSION_QUERIES.load(std::sync::atomic::Ordering::Relaxed), + AGENT_ORG_LEASE_MUTATIONS.load(std::sync::atomic::Ordering::Relaxed), + ) +} + +#[cfg(any(test, debug_assertions))] +pub(super) fn record_submission_query() { + AGENT_ORG_SUBMISSION_QUERIES.fetch_add(1, std::sync::atomic::Ordering::Relaxed); +} + +pub(super) fn is_conversation_deleting_with_connection( + conn: &Connection, + root_session_id: &str, +) -> Result { + conn.query_row( + "SELECT EXISTS( + SELECT 1 + FROM agent_org_conversation_delete_fences + WHERE root_session_id=?1 + )", + [root_session_id], + |row| row.get(0), + ) + .map_err(|err| err.to_string()) +} + +pub(crate) fn ensure_conversation_writable_with_connection( + conn: &Connection, + root_session_id: &str, +) -> Result<(), String> { + if is_conversation_deleting_with_connection(conn, root_session_id)? { + return Err(format!( + "{CONVERSATION_DELETING_ERROR_CODE}: Agent Org root {root_session_id} is being deleted" + )); + } + Ok(()) +} + +pub(crate) fn ensure_run_conversation_writable_with_connection( + conn: &Connection, + run_id: &str, +) -> Result<(), String> { + let root_session_id = conn + .query_row( + "SELECT root_session_id FROM agent_org_runs WHERE id=?1", + [run_id], + |row| row.get::<_, Option>(0), + ) + .optional() + .map_err(|err| err.to_string())? + .flatten() + .ok_or_else(|| format!("agent_org_run_not_found: run {run_id} has no Root"))?; + ensure_conversation_writable_with_connection(conn, &root_session_id) +} + +/// A Run is writable only while it is running and its root conversation is +/// not fenced. Callers that scanned earlier must repeat this check in the +/// transaction that performs their durable write. +pub(crate) fn is_run_writable_with_connection( + conn: &Connection, + run_id: &str, +) -> Result { + conn.query_row( + "SELECT EXISTS( + SELECT 1 + FROM agent_org_runs run + WHERE run.id=?1 + AND run.status='running' + AND run.root_session_id IS NOT NULL + AND NOT EXISTS ( + SELECT 1 + FROM agent_org_conversation_delete_fences fence + WHERE fence.root_session_id=run.root_session_id + ) + )", + [run_id], + |row| row.get(0), + ) + .map_err(|err| err.to_string()) +} + +/// Establish the root fence and cancel every live Run in one caller-owned +/// SQLite transaction. `conn` may be a `Transaction` coerced to `Connection`. +pub(crate) fn establish_conversation_delete_fence_with_connection( + conn: &Connection, + root_session_id: &str, +) -> Result, String> { + let run_rows = { + let mut stmt = conn + .prepare( + "SELECT id, status + FROM agent_org_runs + WHERE root_session_id=?1 + ORDER BY id", + ) + .map_err(|err| err.to_string())?; + let rows = stmt + .query_map([root_session_id], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + }) + .map_err(|err| err.to_string())?; + rows.collect::, _>>() + .map_err(|err| err.to_string())? + }; + if run_rows.is_empty() { + // A concurrent deletion may commit after this caller planned but + // before it acquires the writer transaction. Do not recreate an + // orphan fence for an already-removed conversation. + return Ok(Vec::new()); + } + + let mut cancelled_run_ids = Vec::new(); + for (run_id, status_raw) in &run_rows { + let status = AgentOrgRunStatus::parse(status_raw).ok_or_else(|| { + format!("unknown Agent Org run status {status_raw:?} for run {run_id}") + })?; + if matches!( + status, + AgentOrgRunStatus::Starting | AgentOrgRunStatus::Running | AgentOrgRunStatus::Paused + ) { + cancelled_run_ids.push(run_id.clone()); + } + } + + let now = chrono::Utc::now().to_rfc3339(); + conn.execute( + "INSERT INTO agent_org_conversation_delete_fences ( + root_session_id, created_at, updated_at + ) VALUES (?1, ?2, ?2) + ON CONFLICT(root_session_id) DO UPDATE SET updated_at=excluded.updated_at", + params![root_session_id, &now], + ) + .map_err(|err| err.to_string())?; + conn.execute( + "UPDATE agent_org_runs + SET status='cancelled', + updated_at=?2, + completed_at=COALESCE(completed_at, ?2) + WHERE root_session_id=?1 + AND status IN ('starting', 'running', 'paused')", + params![root_session_id, &now], + ) + .map_err(|err| err.to_string())?; + conn.execute( + "UPDATE agent_org_plan_approvals + SET status='cancelled', decision_by='system', resolved_at=?2 + WHERE status='pending' + AND org_run_id IN ( + SELECT id FROM agent_org_runs WHERE root_session_id=?1 + )", + params![root_session_id, &now], + ) + .map_err(|err| err.to_string())?; + + Ok(cancelled_run_ids) +} + +/// Own the short, serialized transaction used to establish a deletion fence. +/// Callers that must compare topology inside the same snapshot should instead +/// use [`establish_conversation_delete_fence_with_connection`]. +#[cfg(test)] +pub(crate) fn establish_conversation_delete_fence(root_session_id: &str) -> Result<(), String> { + let root_session_id = root_session_id.to_string(); + let outcome = with_sessions_writer(|| -> Result<_, String> { + let mut conn = get_connection().map_err(|err| err.to_string())?; + let tx = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|err| err.to_string())?; + let outcome = establish_conversation_delete_fence_with_connection(&tx, &root_session_id)?; + tx.commit().map_err(|err| err.to_string())?; + Ok(outcome) + })?; + for run_id in &outcome { + crate::coordination::agent_org_run_events::notify_agent_org_run_changed(run_id); + } + Ok(()) +} + +#[cfg(debug_assertions)] +pub fn debug_establish_e2e_conversation_delete_fence(root_session_id: &str) -> Result<(), String> { + if !root_session_id.contains("e2e-agent-org-fixture:") { + return Err("debug fence is restricted to disposable E2E fixtures".to_string()); + } + let root_session_id = root_session_id.to_string(); + with_sessions_writer(|| { + let mut conn = get_connection().map_err(|err| err.to_string())?; + let tx = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|err| err.to_string())?; + establish_conversation_delete_fence_with_connection(&tx, &root_session_id)?; + tx.commit().map_err(|err| err.to_string()) + }) +} + +pub(crate) fn remove_conversation_delete_fence_with_connection( + conn: &Connection, + root_session_id: &str, +) -> Result { + conn.execute( + "DELETE FROM agent_org_conversation_delete_fences WHERE root_session_id=?1", + [root_session_id], + ) + .map(|changed| changed > 0) + .map_err(|err| err.to_string()) +} diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/deletion_tests.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/deletion_tests.rs new file mode 100644 index 000000000..07873e79a --- /dev/null +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/deletion_tests.rs @@ -0,0 +1,602 @@ +use rusqlite::params; + +use crate::definitions::orgs::OrgDefinition; +use crate::session::persistence::UnifiedSessionRecord; +use crate::session::SessionStatus; + +use super::deletion::{ + establish_conversation_delete_fence, establish_conversation_delete_fence_with_connection, + is_conversation_deleting_with_connection, CONVERSATION_DELETING_ERROR_CODE, +}; +use super::*; + +fn sample_org() -> OrgDefinition { + serde_json::from_str(r#"{"id":"org-delete-fence-test","name":"Delete Fence Test","role":"lead","agentId":"agent-coordinator","hierarchyMode":"flat","planApprovalPolicy":"coordinator","children":[{"id":"member-worker","name":"Worker","role":"worker","agentId":"agent-worker","children":[]}]}"#).unwrap() +} + +fn ensure_schemas() { + let conn = database::db::get_connection().expect("test sqlite connection"); + crate::foundation::persistence::test_schema::ensure_agent_sessions_schema(&conn); + crate::foundation::persistence::session_snapshots::ensure_tables_with(&conn) + .expect("session snapshot schema"); + crate::session::persistence::init(&conn).expect("unified session schema"); + crate::coordination::init_agent_org_schemas(&conn).expect("Agent Org schemas"); +} + +fn run_params(root_session_id: &str, status: AgentOrgRunStatus) -> CreateAgentOrgRunParams { + let org = sample_org(); + CreateAgentOrgRunParams { + org_id: org.id.clone(), + coordinator_agent_id: org.agent_id.clone(), + root_session_id: Some(root_session_id.to_string()), + org_snapshot: org, + entry_mode: AgentOrgRunEntryMode::StandaloneSession, + status, + work_item_id: None, + project_slug: None, + routine_fire_id: None, + } +} + +fn create_run(root_session_id: &str, status: AgentOrgRunStatus) -> AgentOrgRunRecord { + AgentOrgRunStore::create(run_params(root_session_id, status)).expect("create test Run") +} + +fn load_status(run_id: &str) -> String { + database::db::get_connection() + .expect("test sqlite connection") + .query_row( + "SELECT status FROM agent_org_runs WHERE id=?1", + [run_id], + |row| row.get(0), + ) + .expect("load Run status") +} + +fn insert_pending_approval(run_id: &str, root_session_id: &str) { + let conn = database::db::get_connection().expect("test sqlite connection"); + conn.execute( + "INSERT INTO agent_org_plan_approvals (approval_id,plan_revision_id,request_id,org_run_id,source_task_id,source_member_id,source_session_id,root_session_id,policy,status,plan_title,plan_path,plan_content,created_at) VALUES (?1,?2,?3,?4,'task','member-worker','worker',?5,'coordinator','pending','Plan','/tmp/plan.md','# Plan',?6)", + params![ + format!("approval-{run_id}"), + format!("revision-{run_id}"), + format!("request-{run_id}"), + run_id, + root_session_id, + chrono::Utc::now().to_rfc3339(), + ], + ) + .expect("insert pending approval"); +} + +#[test] +fn starting_status_round_trips_and_is_non_terminal() { + assert_eq!( + AgentOrgRunStatus::parse(AgentOrgRunStatus::Starting.as_str()), + Some(AgentOrgRunStatus::Starting) + ); + assert!(!AgentOrgRunStatus::Starting.is_terminal()); +} + +#[test] +fn fence_schema_is_idempotent_and_survives_reopen() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + let root = "root-durable-fence"; + let conn = database::db::get_connection().expect("test sqlite connection"); + conn.execute("DROP TABLE agent_org_conversation_delete_fences", []) + .expect("simulate a pre-fence database"); + super::init_schema(&conn).expect("upgrade pre-fence database"); + let initially_empty: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_conversation_delete_fences", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(initially_empty, 0); + let run = create_run(root, AgentOrgRunStatus::Completed); + + establish_conversation_delete_fence(root).expect("establish fence"); + assert_eq!(load_status(&run.id), AgentOrgRunStatus::Completed.as_str()); + + let conn = database::db::get_connection().expect("reopened test sqlite connection"); + assert!(is_conversation_deleting_with_connection(&conn, root).unwrap()); + let created_at: String = conn + .query_row( + "SELECT created_at FROM agent_org_conversation_delete_fences + WHERE root_session_id=?1", + [root], + |row| row.get(0), + ) + .unwrap(); + super::init_schema(&conn).expect("repeat schema initialization"); + assert!(is_conversation_deleting_with_connection(&conn, root).unwrap()); + + establish_conversation_delete_fence(root).expect("repeat fence"); + let repeated_created_at: String = conn + .query_row( + "SELECT created_at FROM agent_org_conversation_delete_fences + WHERE root_session_id=?1", + [root], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(repeated_created_at, created_at); + let fence_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_conversation_delete_fences + WHERE root_session_id=?1", + [root], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(fence_count, 1); +} + +#[test] +fn fence_cancels_every_live_status_and_pending_approval() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + + for (suffix, status) in [ + ("starting", AgentOrgRunStatus::Starting), + ("running", AgentOrgRunStatus::Running), + ("paused", AgentOrgRunStatus::Paused), + ("completed", AgentOrgRunStatus::Completed), + ] { + let expected = if status == AgentOrgRunStatus::Completed { + status + } else { + AgentOrgRunStatus::Cancelled + }; + let root = format!("root-fence-{suffix}"); + let run = create_run(&root, status); + insert_pending_approval(&run.id, &root); + + establish_conversation_delete_fence(&root).expect("establish fence"); + assert_eq!(load_status(&run.id), expected.as_str()); + + let conn = database::db::get_connection().unwrap(); + let approval: (String, Option) = conn + .query_row( + "SELECT status, decision_by FROM agent_org_plan_approvals + WHERE org_run_id=?1", + [&run.id], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert_eq!( + approval, + ("cancelled".to_string(), Some("system".to_string())) + ); + } +} + +#[test] +fn one_fence_covers_historical_and_live_runs_for_the_same_root() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + let root = "root-fence-multi-run"; + let historical = create_run(root, AgentOrgRunStatus::Completed); + let live = create_run(root, AgentOrgRunStatus::Running); + insert_pending_approval(&historical.id, root); + insert_pending_approval(&live.id, root); + + establish_conversation_delete_fence(root).expect("fence the whole conversation"); + + assert_eq!( + load_status(&historical.id), + AgentOrgRunStatus::Completed.as_str() + ); + assert_eq!(load_status(&live.id), AgentOrgRunStatus::Cancelled.as_str()); + let conn = database::db::get_connection().unwrap(); + let cancelled_approvals: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_plan_approvals + WHERE org_run_id IN (?1, ?2) AND status='cancelled'", + params![historical.id, live.id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(cancelled_approvals, 2); +} + +#[test] +fn fence_establishment_rolls_back_as_one_transaction() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + let root = "root-fence-rollback"; + let run = create_run(root, AgentOrgRunStatus::Running); + insert_pending_approval(&run.id, root); + + let mut conn = database::db::get_connection().unwrap(); + let tx = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .unwrap(); + establish_conversation_delete_fence_with_connection(&tx, root).unwrap(); + tx.rollback().unwrap(); + + assert_eq!(load_status(&run.id), AgentOrgRunStatus::Running.as_str()); + let conn = database::db::get_connection().unwrap(); + assert!(!is_conversation_deleting_with_connection(&conn, root).unwrap()); + let approval_status: String = conn + .query_row( + "SELECT status FROM agent_org_plan_approvals WHERE org_run_id=?1", + [&run.id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(approval_status, "pending"); +} + +#[test] +fn fenced_root_rejects_run_creation_and_worker_materialization_without_orphans() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + let root = "root-fence-write-guards"; + let run = create_run(root, AgentOrgRunStatus::Running); + establish_conversation_delete_fence(root).expect("establish fence"); + + let create_error = AgentOrgRunStore::create(run_params(root, AgentOrgRunStatus::Running)) + .expect_err("fenced root must reject a new Run"); + assert!(create_error.starts_with(CONVERSATION_DELETING_ERROR_CODE)); + + let coordinator = UnifiedSessionRecord { + session_id: root.to_string(), + name: "Blocked coordinator".to_string(), + status: SessionStatus::Idle.as_str().to_string(), + session_type: "agent".to_string(), + agent_definition_id: Some("agent-coordinator".to_string()), + org_member_id: Some(COORDINATOR_MEMBER_ID.to_string()), + created_at: chrono::Utc::now().to_rfc3339(), + updated_at: chrono::Utc::now().to_rfc3339(), + ..Default::default() + }; + let coordinator_error = AgentOrgRunStore::create_with_coordinator_session( + run_params(root, AgentOrgRunStatus::Running), + Default::default(), + &coordinator, + ) + .expect_err("fenced root must reject coordinator materialization"); + assert!(coordinator_error.starts_with(CONVERSATION_DELETING_ERROR_CODE)); + + let worker = UnifiedSessionRecord { + session_id: "blocked-worker".to_string(), + name: "Blocked worker".to_string(), + status: SessionStatus::Pending.as_str().to_string(), + session_type: crate::session::persistence::session_type::ORG_MEMBER.to_string(), + agent_definition_id: Some("agent-worker".to_string()), + org_member_id: Some("member-worker".to_string()), + parent_session_id: Some(root.to_string()), + created_at: chrono::Utc::now().to_rfc3339(), + updated_at: chrono::Utc::now().to_rfc3339(), + ..Default::default() + }; + let worker_error = + AgentOrgRunStore::materialize_rust_worker_sessions(&run.id, &[worker.clone()]) + .expect_err("fenced root must reject worker materialization"); + assert!(worker_error.starts_with(CONVERSATION_DELETING_ERROR_CODE)); + + let conn = database::db::get_connection().unwrap(); + let session_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_sessions + WHERE session_id IN (?1, ?2)", + params![root, worker.session_id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(session_count, 0); + let run_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_runs WHERE root_session_id=?1", + [root], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(run_count, 1); +} + +#[test] +fn standalone_run_delete_cannot_remove_ownership_beneath_a_root_fence() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + let root = "root-fenced-standalone-run-delete"; + let run = create_run(root, AgentOrgRunStatus::Completed); + establish_conversation_delete_fence(root).expect("establish fence"); + + let error = AgentOrgRunStore::delete_by_id(&run.id) + .expect_err("standalone cleanup must not bypass root deletion ownership"); + assert!(error.starts_with(CONVERSATION_DELETING_ERROR_CODE)); + + let conn = database::db::get_connection().unwrap(); + let run_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_runs WHERE id=?1", + [&run.id], + |row| row.get(0), + ) + .unwrap(); + let mapping_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_run_sessions WHERE org_run_id=?1", + [&run.id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(run_count, 1); + assert_eq!(mapping_count, 1); + assert!(is_conversation_deleting_with_connection(&conn, root).unwrap()); +} + +#[test] +fn unfenced_live_or_paused_run_can_finish_materializing_workers() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + for status in [ + AgentOrgRunStatus::Starting, + AgentOrgRunStatus::Running, + AgentOrgRunStatus::Paused, + ] { + let root = format!("root-{}-materialization", status.as_str()); + let run = create_run(&root, status); + let worker_id = format!("{}-worker", status.as_str()); + let worker = UnifiedSessionRecord { + session_id: worker_id.clone(), + name: "Materializing worker".to_string(), + status: SessionStatus::Pending.as_str().to_string(), + session_type: crate::session::persistence::session_type::ORG_MEMBER.to_string(), + agent_definition_id: Some("agent-worker".to_string()), + org_member_id: Some("member-worker".to_string()), + parent_session_id: Some(root), + created_at: chrono::Utc::now().to_rfc3339(), + updated_at: chrono::Utc::now().to_rfc3339(), + ..Default::default() + }; + + AgentOrgRunStore::materialize_rust_worker_sessions(&run.id, &[worker]) + .expect("unfenced launch topology may finish materializing"); + let conn = database::db::get_connection().unwrap(); + let mapping_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_run_sessions + WHERE org_run_id=?1 AND session_id=?2 AND role='worker'", + params![run.id, worker_id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(mapping_count, 1, "status={}", status.as_str()); + } +} + +#[test] +fn worker_materialization_rejects_terminal_phases() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + + for status in [ + AgentOrgRunStatus::Completed, + AgentOrgRunStatus::Failed, + AgentOrgRunStatus::Cancelled, + AgentOrgRunStatus::Abandoned, + ] { + let root = format!("root-materialization-{}", status.as_str()); + let run = create_run(&root, status); + let worker_id = format!("worker-materialization-{}", status.as_str()); + let worker = UnifiedSessionRecord { + session_id: worker_id.clone(), + name: "Blocked worker".to_string(), + status: SessionStatus::Pending.as_str().to_string(), + session_type: crate::session::persistence::session_type::ORG_MEMBER.to_string(), + agent_definition_id: Some("agent-worker".to_string()), + org_member_id: Some("member-worker".to_string()), + parent_session_id: Some(root), + created_at: chrono::Utc::now().to_rfc3339(), + updated_at: chrono::Utc::now().to_rfc3339(), + ..Default::default() + }; + + let error = AgentOrgRunStore::materialize_rust_worker_sessions(&run.id, &[worker]) + .expect_err("non-materializing phase must reject Workers"); + assert!(error.starts_with("agent_org_run_not_materializable:")); + + let conn = database::db::get_connection().unwrap(); + let session_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_sessions WHERE session_id=?1", + [&worker_id], + |row| row.get(0), + ) + .unwrap(); + let mapping_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_run_sessions + WHERE org_run_id=?1 AND role='worker'", + [&run.id], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(session_count, 0, "status={}", status.as_str()); + assert_eq!(mapping_count, 0, "status={}", status.as_str()); + } +} + +#[test] +fn shared_run_writable_predicate_and_resume_observe_fence() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + let writable_root = "root-run-writable"; + let writable_run = create_run(writable_root, AgentOrgRunStatus::Running); + let conn = database::db::get_connection().unwrap(); + assert!(is_run_writable_with_connection(&conn, &writable_run.id).unwrap()); + + let paused_root = "root-resume-fenced"; + let paused_run = create_run(paused_root, AgentOrgRunStatus::Paused); + establish_conversation_delete_fence(paused_root).unwrap(); + let resume_error = AgentOrgRunStore::mark_resumed(&paused_run.id) + .expect_err("fenced paused Run must not resume"); + assert!(resume_error.starts_with(CONVERSATION_DELETING_ERROR_CODE)); + AgentOrgRunStore::mark_failed(&paused_run.id, "late materialization failure").unwrap(); + assert_eq!( + load_status(&paused_run.id), + AgentOrgRunStatus::Cancelled.as_str() + ); + + let failed_root = "root-paused-materialization-failed"; + let failed_run = create_run(failed_root, AgentOrgRunStatus::Paused); + AgentOrgRunStore::mark_failed(&failed_run.id, "materialization failed").unwrap(); + assert_eq!( + load_status(&failed_run.id), + AgentOrgRunStatus::Failed.as_str() + ); + + let conn = database::db::get_connection().unwrap(); + assert!(!is_run_writable_with_connection(&conn, &paused_run.id).unwrap()); + assert!(remove_conversation_delete_fence_with_connection(&conn, paused_root).unwrap()); + assert!(!is_conversation_deleting_with_connection(&conn, paused_root).unwrap()); +} + +#[tokio::test] +async fn ordinary_session_admission_has_zero_agent_org_queries_and_leases() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + reset_submission_metrics(); + let record = UnifiedSessionRecord { + session_id: "ordinary-sde-admission".to_string(), + session_type: crate::session::persistence::session_type::CODING.to_string(), + ..Default::default() + }; + let resolved = submission_scope_for_loaded_session(&record).expect("classify ordinary SDE"); + assert_eq!(resolved, AgentOrgSubmissionScope::Ordinary); + let scope = std::sync::Arc::new(AgentOrgSubmissionPolicy::new(resolved)); + + assert!(admit_agent_org_submission(&scope, &record.session_id) + .await + .expect("ordinary admission") + .is_none()); + assert_eq!(submission_metrics(), (0, 0)); +} + +#[tokio::test] +async fn ordinary_session_admission_ignores_agent_org_schema_failure() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + let conn = database::db::get_connection().expect("test sqlite connection"); + conn.execute("DROP TABLE agent_org_conversation_delete_fences", []) + .expect("inject Agent Org schema failure"); + drop(conn); + reset_submission_metrics(); + let scope = std::sync::Arc::new(AgentOrgSubmissionPolicy::new( + AgentOrgSubmissionScope::Ordinary, + )); + + assert!( + admit_agent_org_submission(&scope, "ordinary-schema-failure") + .await + .expect("ordinary SDE must not depend on Agent Org schema") + .is_none() + ); + assert_eq!(submission_metrics(), (0, 0)); +} + +#[tokio::test] +async fn cold_unknown_resolves_exact_mapping_once_then_stays_ordinary() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + reset_submission_metrics(); + let scope = std::sync::Arc::new(AgentOrgSubmissionPolicy::new( + AgentOrgSubmissionScope::Unknown, + )); + + for _ in 0..2 { + assert!(admit_agent_org_submission(&scope, "cold-unknown-ordinary") + .await + .expect("resolve cold unknown") + .is_none()); + } + assert_eq!(scope.snapshot(), AgentOrgSubmissionScope::Ordinary); + assert_eq!(submission_metrics(), (1, 0)); +} + +#[tokio::test] +async fn concurrent_cold_unknown_uses_one_exact_mapping_query() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + reset_submission_metrics(); + let scope = std::sync::Arc::new(AgentOrgSubmissionPolicy::new( + AgentOrgSubmissionScope::Unknown, + )); + let admissions = (0..8).map(|_| { + let scope = std::sync::Arc::clone(&scope); + tokio::spawn(async move { + admit_agent_org_submission(&scope, "cold-unknown-concurrent") + .await + .expect("ordinary admission") + }) + }); + for admission in admissions { + assert!(admission.await.expect("admission task").is_none()); + } + assert_eq!(submission_metrics(), (1, 0)); +} + +#[tokio::test] +async fn agent_org_admission_lease_balances_and_fence_rejects() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + let root = "root-submission-admission"; + let run = create_run(root, AgentOrgRunStatus::Running); + let scope = std::sync::Arc::new(AgentOrgSubmissionPolicy::new( + AgentOrgSubmissionScope::Run { + run_id: run.id.clone(), + }, + )); + reset_submission_metrics(); + + let lease = admit_agent_org_submission(&scope, root) + .await + .expect("admit writable Agent Org") + .expect("Agent Org acquires lease"); + assert!(agent_org_submission_in_progress(root)); + drop(lease); + assert!(!agent_org_submission_in_progress(root)); + assert_eq!(submission_metrics(), (1, 2)); + + establish_conversation_delete_fence(root).expect("fence root"); + let error = admit_agent_org_submission(&scope, root) + .await + .expect_err("fenced Agent Org rejects submission"); + assert!(error.starts_with(CONVERSATION_DELETING_ERROR_CODE)); + assert!(!agent_org_submission_in_progress(root)); + assert_eq!(submission_metrics(), (2, 4)); +} + +#[tokio::test] +async fn install_recheck_rejects_fence_created_after_initial_admission() { + let _sandbox = test_helpers::test_env::sandbox(); + ensure_schemas(); + let root = "root-late-install-fence"; + let run = create_run(root, AgentOrgRunStatus::Running); + let scope = std::sync::Arc::new(AgentOrgSubmissionPolicy::new( + AgentOrgSubmissionScope::Run { + run_id: run.id.clone(), + }, + )); + + let lease = admit_agent_org_submission(&scope, root) + .await + .expect("initial runtime-build admission") + .expect("Agent Org runtime build holds lease"); + establish_conversation_delete_fence(root).expect("fence during slow runtime build"); + let error = recheck_agent_org_submission(&scope) + .await + .expect_err("the sole install boundary must reject the late fence"); + assert!( + error.starts_with(CONVERSATION_DELETING_ERROR_CODE), + "{error}" + ); + assert!(agent_org_submission_in_progress(root)); + drop(lease); + assert!(!agent_org_submission_in_progress(root)); +} diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/migration.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/migration.rs index 05d98223d..cd2f8788c 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/migration.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/migration.rs @@ -91,7 +91,12 @@ pub(super) fn init_schema(conn: &Connection) -> SqliteResult &'static str { match self { + Self::Starting => "starting", Self::Running => "running", Self::Paused => "paused", Self::Completed => "completed", @@ -89,6 +114,7 @@ impl AgentOrgRunStatus { pub fn parse(value: &str) -> Option { match value { + "starting" => Some(Self::Starting), "running" => Some(Self::Running), "paused" => Some(Self::Paused), "completed" => Some(Self::Completed), diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/store.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/store.rs index b9a6eb4e4..7412c3e02 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/store.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/store.rs @@ -12,6 +12,7 @@ use crate::session::persistence::{ use crate::session::SessionStatus; use database::db::{get_connection, with_sessions_writer}; +use super::ensure_conversation_writable_with_connection; use super::finality::load_and_assess; use super::helpers::{ context_for_run_record, flatten_members, insert_run, load_by_id, row_to_run, @@ -184,6 +185,9 @@ impl AgentOrgRunStore { let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; + if let Some(root_session_id) = run.root_session_id.as_deref() { + ensure_conversation_writable_with_connection(&tx, root_session_id)?; + } Self::validate_lineage_with_connection(&tx, run)?; if let Some(session) = coordinator_session { upsert_session_with_connection(&tx, session).map_err(|err| err.to_string())?; @@ -332,16 +336,36 @@ impl AgentOrgRunStore { let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; - let expected_root: String = tx + let (expected_root, status_raw): (Option, String) = tx .query_row( - "SELECT root_session_id FROM agent_org_runs WHERE id=?1", + "SELECT root_session_id, status FROM agent_org_runs WHERE id=?1", [org_run_id], - |row| row.get::<_, Option>(0), + |row| Ok((row.get::<_, Option>(0)?, row.get(1)?)), ) .optional() .map_err(|err| err.to_string())? - .flatten() + .ok_or_else(|| format!("Agent Org run {org_run_id} was not found"))?; + let expected_root = expected_root .ok_or_else(|| format!("Agent Org run {org_run_id} has no root session"))?; + // Pausing can win the race with detached launch materialization. + // A paused Run may finish persisting its topology, but no work is + // dispatched until resume. Check the fence first so deletion still + // produces its stable error after changing the Run to cancelled. + ensure_conversation_writable_with_connection(&tx, &expected_root)?; + let status = AgentOrgRunStatus::parse(&status_raw).ok_or_else(|| { + format!("unknown Agent Org run status {status_raw:?} for run {org_run_id}") + })?; + if !matches!( + status, + AgentOrgRunStatus::Starting + | AgentOrgRunStatus::Running + | AgentOrgRunStatus::Paused + ) { + return Err(format!( + "agent_org_run_not_materializable: run {org_run_id} has status {}", + status.as_str() + )); + } let roster = Self::snapshot_member_agent_ids_with_connection(&tx, org_run_id)? .ok_or_else(|| format!("Agent Org run {org_run_id} has no launch snapshot"))?; for session in sessions { @@ -481,8 +505,23 @@ impl AgentOrgRunStore { let paused = validate_status(AgentOrgRunStatus::Paused.as_str())?; let now = chrono::Utc::now().to_rfc3339(); let changed = with_sessions_writer(|| -> Result { - let conn = get_connection().map_err(|err| err.to_string())?; - let rows_changed = conn + let mut conn = get_connection().map_err(|err| err.to_string())?; + let tx = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|err| err.to_string())?; + let root_session_id = tx + .query_row( + "SELECT root_session_id FROM agent_org_runs WHERE id=?1", + [run_id], + |row| row.get::<_, Option>(0), + ) + .optional() + .map_err(|err| err.to_string())? + .flatten(); + if let Some(root_session_id) = root_session_id.as_deref() { + ensure_conversation_writable_with_connection(&tx, root_session_id)?; + } + let rows_changed = tx .execute( "UPDATE agent_org_runs SET status = ?1, @@ -492,6 +531,7 @@ impl AgentOrgRunStore { params![running.as_str(), now, run_id, paused.as_str()], ) .map_err(|err| err.to_string())?; + tx.commit().map_err(|err| err.to_string())?; Ok(rows_changed > 0) })?; if changed { @@ -500,66 +540,42 @@ impl AgentOrgRunStore { Ok(changed) } - /// Establish the durable fence for a user-requested hierarchy deletion. - /// - /// `paused` remains resumable, so deletion must not use it as the final - /// stop signal. Moving a live run to `cancelled` prevents resume and wake - /// paths from starting new work while the caller drains Rust runtimes. - pub(crate) fn cancel_for_delete_with_connection( - conn: &Connection, - run_id: &str, - ) -> Result { - let now = chrono::Utc::now().to_rfc3339(); - let changed = conn - .execute( - "UPDATE agent_org_runs - SET status='cancelled', - updated_at=?2, - completed_at=COALESCE(completed_at, ?2) - WHERE id=?1 - AND status IN ('running', 'paused')", - params![run_id, &now], - ) - .map_err(|err| err.to_string())? - > 0; - conn.execute( - "UPDATE agent_org_plan_approvals - SET status='cancelled', decision_by='system', resolved_at=?2 - WHERE org_run_id=?1 AND status='pending'", - params![run_id, &now], - ) - .map_err(|err| err.to_string())?; - Ok(changed) - } - pub fn mark_failed(run_id: &str, error_message: &str) -> Result<(), String> { let status = validate_status(AgentOrgRunStatus::Failed.as_str())?; let now = chrono::Utc::now().to_rfc3339(); - with_sessions_writer(|| -> Result<(), String> { + let changed = with_sessions_writer(|| -> Result { let mut conn = get_connection().map_err(|err| err.to_string())?; let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; - tx.execute( - "UPDATE agent_org_runs + let changed = tx + .execute( + "UPDATE agent_org_runs SET status = ?1, last_error = ?2, updated_at = ?3, completed_at = ?3 - WHERE id = ?4", - params![status.as_str(), error_message, now, run_id], - ) - .map_err(|err| err.to_string())?; - tx.execute( - "UPDATE agent_org_plan_approvals - SET status='cancelled', decision_by='system', resolved_at=?2 - WHERE org_run_id=?1 AND status='pending'", - params![run_id, &now], - ) - .map_err(|err| err.to_string())?; - tx.commit().map_err(|err| err.to_string()) + WHERE id = ?4 + AND status IN ('starting', 'running', 'paused')", + params![status.as_str(), error_message, now, run_id], + ) + .map_err(|err| err.to_string())? + > 0; + if changed { + tx.execute( + "UPDATE agent_org_plan_approvals + SET status='cancelled', decision_by='system', resolved_at=?2 + WHERE org_run_id=?1 AND status='pending'", + params![run_id, &now], + ) + .map_err(|err| err.to_string())?; + } + tx.commit().map_err(|err| err.to_string())?; + Ok(changed) })?; - crate::coordination::agent_org_run_events::notify_agent_org_run_changed(run_id); + if changed { + crate::coordination::agent_org_run_events::notify_agent_org_run_changed(run_id); + } Ok(()) } @@ -801,6 +817,8 @@ impl AgentOrgRunStore { resolution: AgentOrgRunResolution, access: AgentOrgRunResolutionAccess, ) -> Result, String> { + #[cfg(any(test, debug_assertions))] + super::deletion::record_submission_query(); let Some(run) = Self::resolve_run_for_session(session_id, resolution, access)? else { return Ok(None); }; @@ -1013,7 +1031,9 @@ impl AgentOrgRunStore { .filter(|run| { matches!( run.status, - AgentOrgRunStatus::Running | AgentOrgRunStatus::Paused + AgentOrgRunStatus::Starting + | AgentOrgRunStatus::Running + | AgentOrgRunStatus::Paused ) }) .collect::>(); @@ -1293,6 +1313,11 @@ impl AgentOrgRunStore { FROM agent_org_runs WHERE root_session_id IS NOT NULL AND status = ?1 + AND NOT EXISTS ( + SELECT 1 + FROM agent_org_conversation_delete_fences fence + WHERE fence.root_session_id=agent_org_runs.root_session_id + ) ORDER BY updated_at DESC LIMIT ?2", ) @@ -1351,6 +1376,18 @@ impl AgentOrgRunStore { let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; + let root_session_id = tx + .query_row( + "SELECT root_session_id FROM agent_org_runs WHERE id=?1", + [run_id], + |row| row.get::<_, Option>(0), + ) + .optional() + .map_err(|err| err.to_string())? + .flatten(); + if let Some(root_session_id) = root_session_id { + ensure_conversation_writable_with_connection(&tx, &root_session_id)?; + } let outcome = Self::delete_by_id_with_connection(&tx, run_id)?; tx.commit().map_err(|err| err.to_string())?; Ok(outcome) diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_tasks/mod.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_tasks/mod.rs index 5c275ada9..6cd258ab9 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_org_tasks/mod.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_tasks/mod.rs @@ -13,7 +13,7 @@ use database::db::{get_connection, with_sessions_writer}; use rusqlite::{params, Connection, Result as SqliteResult}; use crate::coordination::agent_org_runs::{ - recovery_dispatch_recipient_is_available, AgentOrgRunStore, + is_run_writable_with_connection, recovery_dispatch_recipient_is_available, AgentOrgRunStore, }; pub(super) mod graph; @@ -819,16 +819,7 @@ pub(crate) fn enqueue_task_assignments_if_still_ready_for_recovery( let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; - let running: bool = tx - .query_row( - "SELECT EXISTS( - SELECT 1 FROM agent_org_runs WHERE id=?1 AND status='running' - )", - params![org_run_id], - |row| row.get(0), - ) - .map_err(|err| err.to_string())?; - if !running { + if !is_run_writable_with_connection(&tx, org_run_id)? { tx.commit().map_err(|err| err.to_string())?; return Ok(Vec::new()); } diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/inspect.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/inspect.rs index 36b29ad67..385a1099c 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/inspect.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/inspect.rs @@ -777,9 +777,7 @@ pub(super) fn inspect_stalled_run_with_connection( conn: &Connection, run_id: &str, ) -> Result { - if AgentOrgRunStore::get_run_status_with_connection(conn, run_id)? - != Some(AgentOrgRunStatus::Running) - { + if !crate::coordination::agent_org_runs::is_run_writable_with_connection(conn, run_id)? { return Ok(StallRecoveryPlan::default()); } diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/recover.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/recover.rs index 3f6d6db36..8fcdcae4f 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/recover.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/recover.rs @@ -12,6 +12,7 @@ use super::inspect::{ PendingMaterializationDisposition, }; use super::*; +use crate::coordination::agent_org_runs::is_run_writable_with_connection; pub fn spawn(app_handle: AppHandle) { tauri::async_runtime::spawn(async move { @@ -123,14 +124,43 @@ pub fn recover_stalled_run( execute_stall_recovery_plan(run_id, plan, wake_hook.as_ref()) } +fn run_is_writable(run_id: &str) -> Result { + let conn = get_connection().map_err(|err| err.to_string())?; + is_run_writable_with_connection(&conn, run_id) +} + +/// Serialize the final fire-and-forget callback with Root fencing. The +/// production hook only schedules async work, so holding the writer gate here +/// is brief; downstream wake promotion still performs its own durable check. +pub(super) fn dispatch_wakes_if_run_writable<'a>( + run_id: &str, + member_ids: impl IntoIterator, + wake_hook: &dyn InboxWakeHook, +) -> Result<(), String> { + let member_ids = member_ids.into_iter().collect::>(); + with_sessions_writer(|| -> Result<(), String> { + let conn = get_connection().map_err(|err| err.to_string())?; + if is_run_writable_with_connection(&conn, run_id)? { + for member_id in member_ids { + wake_hook.wake_member(member_id, run_id); + } + } + Ok(()) + }) +} + /// Execute an advisory analyzer plan through a caller-supplied Wake hook. /// Keeping orchestration here makes the full reconcile → revalidate → persist /// → wake ordering directly testable without constructing a Tauri runtime. -fn execute_stall_recovery_plan( +pub(super) fn execute_stall_recovery_plan( run_id: &str, plan: StallRecoveryPlan, wake_hook: &dyn InboxWakeHook, ) -> Result { + if !run_is_writable(run_id)? { + return Ok(plan); + } + // Reconcile first: when the run actually closes there is nothing // left to wake or repair. When reconciliation declines (e.g. the // coordinator root session is still open), fall through and deliver @@ -184,20 +214,20 @@ fn execute_stall_recovery_plan( // Members without a derived action were selected only because the // analyzer observed unread durable input. Recheck that input rather than // waking from the stale plan alone. - if AgentOrgRunStore::get_run_status(run_id)? == Some(AgentOrgRunStatus::Running) { - for member_id in &plan.wake_member_ids { - if !action_member_ids.contains(member_id.as_str()) - && has_unread_for_member(run_id, member_id)? - { - wake_member_ids.insert(member_id.clone()); - } + for member_id in &plan.wake_member_ids { + if !action_member_ids.contains(member_id.as_str()) + && has_unread_for_member(run_id, member_id)? + { + wake_member_ids.insert(member_id.clone()); } } if !wake_member_ids.is_empty() { - for member_id in &wake_member_ids { - wake_hook.wake_member(member_id, run_id); - } + dispatch_wakes_if_run_writable( + run_id, + wake_member_ids.iter().map(String::as_str), + wake_hook, + )?; } if let Some(reason) = plan.coordinator_repair_reason.as_deref() { @@ -214,7 +244,11 @@ fn execute_stall_recovery_plan( plan.coordinator_repair_inbox_fingerprint.as_deref(), )? { CoordinatorNoticeDispatch::Inserted | CoordinatorNoticeDispatch::ExistingUnread => { - wake_hook.wake_member(COORDINATOR_MEMBER_ID, run_id); + dispatch_wakes_if_run_writable( + run_id, + std::iter::once(COORDINATOR_MEMBER_ID), + wake_hook, + )?; } CoordinatorNoticeDispatch::Deferred => { tracing::debug!( @@ -252,6 +286,10 @@ fn clear_coordinator_notice_budget_if_recovered(run_id: &str) -> Result<(), Stri let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; + if !is_run_writable_with_connection(&tx, run_id)? { + tx.commit().map_err(|err| err.to_string())?; + return Ok(()); + } if !inspect_stalled_run_with_connection(&tx, run_id)?.coordinator_repair_active { tx.execute( "DELETE FROM agent_org_recovery_attempts @@ -291,16 +329,7 @@ fn insert_member_continuation_if_tasks_current( let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; - let running: bool = tx - .query_row( - "SELECT EXISTS( - SELECT 1 FROM agent_org_runs WHERE id=?1 AND status='running' - )", - params![run_id], - |row| row.get(0), - ) - .map_err(|err| err.to_string())?; - if !running { + if !is_run_writable_with_connection(&tx, run_id)? { tx.commit().map_err(|err| err.to_string())?; return Ok(false); } @@ -420,6 +449,10 @@ fn insert_coordinator_stall_notice( let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; + if !is_run_writable_with_connection(&tx, run_id)? { + tx.commit().map_err(|err| err.to_string())?; + return Ok(CoordinatorNoticeDispatch::Stale); + } let current_plan = inspect_stalled_run_with_connection(&tx, run_id)?; if current_plan.coordinator_repair_fingerprint.as_deref() != Some(reason_fingerprint) { tx.commit().map_err(|err| err.to_string())?; diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/reservation.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/reservation.rs index 740af8c24..834fb20b6 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/reservation.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/reservation.rs @@ -9,6 +9,7 @@ use super::budget::{ RecoveryAttemptSnapshot, }; use super::*; +use crate::coordination::agent_org_runs::is_run_writable_with_connection; /// Provisional durable claim for one scheduler dispatch. /// @@ -39,6 +40,10 @@ pub(crate) fn reserve_member_rewake_dispatch( let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; + if !is_run_writable_with_connection(&tx, run_id)? { + tx.commit().map_err(|err| err.to_string())?; + return Ok(MemberRewakeReservationOutcome::Deferred); + } if !matches!( budget_disposition_with_connection(&tx, run_id, MEMBER_REWAKE, member_id, fingerprint,)?, BudgetDisposition::Allowed @@ -96,8 +101,15 @@ pub(crate) fn commit_member_rewake_reservation( reservation: &MemberRewakeReservation, ) -> Result<(), String> { with_sessions_writer(|| -> Result<(), String> { - let conn = get_connection().map_err(|err| err.to_string())?; - conn.execute( + let mut conn = get_connection().map_err(|err| err.to_string())?; + let tx = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|err| err.to_string())?; + if !is_run_writable_with_connection(&tx, &reservation.run_id)? { + tx.commit().map_err(|err| err.to_string())?; + return Ok(()); + } + tx.execute( "UPDATE agent_org_recovery_attempts SET reservation_token=NULL WHERE org_run_id=?1 AND action_kind=?2 AND target_key=?3 @@ -110,7 +122,7 @@ pub(crate) fn commit_member_rewake_reservation( ], ) .map_err(|err| err.to_string())?; - Ok(()) + tx.commit().map_err(|err| err.to_string()) }) } @@ -122,6 +134,10 @@ pub(crate) fn refund_member_rewake_reservation( let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; + if !is_run_writable_with_connection(&tx, &reservation.run_id)? { + tx.commit().map_err(|err| err.to_string())?; + return Ok(false); + } let owns_current: bool = tx .query_row( "SELECT EXISTS( diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/tests.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/tests.rs index a9d20e13e..76af6aa59 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/tests.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_watchdog/tests.rs @@ -2,9 +2,58 @@ use super::budget::{ budget_disposition, coordinator_notice_allowed, rewake_budget_exhausted, BudgetDisposition, }; use super::inspect::is_wakeable_status; -use super::recover::{recover_listed_runs, run_best_effort_cleanup}; +use super::recover::{ + dispatch_wakes_if_run_writable, execute_stall_recovery_plan, recover_listed_runs, + run_best_effort_cleanup, +}; use super::*; use crate::coordination::agent_org_runs::{AgentOrgRunEntryMode, AgentOrgRunRecord}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Barrier}; + +#[derive(Default)] +struct RecordingWakeHook { + calls: AtomicUsize, +} + +impl InboxWakeHook for RecordingWakeHook { + fn wake_member(&self, _member_id: &str, _org_run_id: &str) { + self.calls.fetch_add(1, Ordering::SeqCst); + } +} + +struct BlockingWakeHook { + calls: AtomicUsize, + entered: Barrier, + release: Barrier, +} + +impl InboxWakeHook for BlockingWakeHook { + fn wake_member(&self, _member_id: &str, _org_run_id: &str) { + self.calls.fetch_add(1, Ordering::SeqCst); + self.entered.wait(); + self.release.wait(); + } +} + +fn seed_running_run(conn: &Connection, run_id: &str, root_session_id: &str) { + crate::coordination::agent_org_runs::init_schema(conn).expect("run schema"); + crate::coordination::agent_org_plan_approvals::init_schema(conn).expect("plan approval schema"); + init_schema(conn).expect("watchdog schema"); + let now = Utc::now().to_rfc3339(); + conn.execute( + "INSERT INTO agent_org_runs ( + id, org_id, coordinator_agent_id, root_session_id, + org_snapshot_json, entry_mode, status, work_item_id, + project_slug, routine_fire_id, summary, last_error, + created_at, updated_at, completed_at + ) VALUES (?1, 'org', 'coordinator-agent', ?2, + NULL, 'standalone_session', 'running', NULL, + NULL, NULL, NULL, NULL, ?3, ?3, NULL)", + params![run_id, root_session_id, &now], + ) + .expect("seed running run"); +} fn fake_run(id: &str) -> AgentOrgRunRecord { let now = Utc::now().to_rfc3339(); @@ -40,8 +89,8 @@ fn wakeable_status_includes_idle_and_terminal_but_not_running() { fn member_rewake_reservation_is_atomic_and_refundable() { let _sandbox = test_helpers::test_env::sandbox(); let conn = get_connection().expect("db"); - init_schema(&conn).expect("schema"); let run_id = format!("run-{}", uuid::Uuid::new_v4()); + seed_running_run(&conn, &run_id, "root-reserved"); let member_id = "member-reserved"; let fingerprint = "unread-42"; @@ -68,8 +117,8 @@ fn member_rewake_reservation_is_atomic_and_refundable() { fn stale_rewake_refund_cannot_undo_newer_input() { let _sandbox = test_helpers::test_env::sandbox(); let conn = get_connection().expect("db"); - init_schema(&conn).expect("schema"); let run_id = format!("run-{}", uuid::Uuid::new_v4()); + seed_running_run(&conn, &run_id, "root-new-input"); let member_id = "member-new-input"; let old = match reserve_member_rewake_dispatch(&run_id, member_id, "unread-1") .expect("reserve old fingerprint") @@ -97,6 +146,122 @@ fn stale_rewake_refund_cannot_undo_newer_input() { ); } +#[test] +fn fenced_running_run_is_excluded_and_stale_recovery_is_a_noop() { + let _sandbox = test_helpers::test_env::sandbox(); + let run_id = format!("run-{}", uuid::Uuid::new_v4()); + let root_session_id = "root-fenced-watchdog"; + let conn = get_connection().expect("db"); + seed_running_run(&conn, &run_id, root_session_id); + drop(conn); + + crate::coordination::agent_org_runs::establish_conversation_delete_fence(root_session_id) + .expect("establish durable root fence"); + let conn = get_connection().expect("db"); + conn.execute( + "UPDATE agent_org_runs SET status='running' WHERE id=?1", + params![&run_id], + ) + .expect("keep run running to isolate root-fence behavior"); + + assert!( + AgentOrgRunStore::list_running_runs(usize::MAX) + .expect("list writable running runs") + .into_iter() + .all(|run| run.id != run_id), + "watchdog scan must exclude a fenced root" + ); + assert!( + inspect_stalled_run(&run_id) + .expect("inspect fenced run") + .is_noop(), + "direct inspection must not derive recovery work for a fenced root" + ); + + let stale_plan = StallRecoveryPlan { + wake_member_ids: vec!["member-stale".to_string()], + ..StallRecoveryPlan::default() + }; + let wake_hook = RecordingWakeHook::default(); + assert_eq!( + execute_stall_recovery_plan(&run_id, stale_plan.clone(), &wake_hook) + .expect("execute stale plan"), + stale_plan + ); + assert_eq!(wake_hook.calls.load(Ordering::SeqCst), 0); + assert!(matches!( + reserve_member_rewake_dispatch(&run_id, "member-stale", "unread-stale") + .expect("fenced reservation is deferred"), + MemberRewakeReservationOutcome::Deferred + )); + let recovery_attempts: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_recovery_attempts WHERE org_run_id=?1", + params![&run_id], + |row| row.get(0), + ) + .expect("count recovery attempts"); + assert_eq!(recovery_attempts, 0); +} + +#[test] +fn wake_callback_and_root_fence_have_one_serial_order() { + let _sandbox = test_helpers::test_env::sandbox(); + let run_id = format!("run-{}", uuid::Uuid::new_v4()); + let root_session_id = "root-watchdog-callback-race"; + let conn = get_connection().expect("db"); + seed_running_run(&conn, &run_id, root_session_id); + drop(conn); + + let hook = Arc::new(BlockingWakeHook { + calls: AtomicUsize::new(0), + entered: Barrier::new(2), + release: Barrier::new(2), + }); + let dispatch_hook = Arc::clone(&hook); + let dispatch_run_id = run_id.clone(); + let dispatch = std::thread::spawn(move || { + dispatch_wakes_if_run_writable( + &dispatch_run_id, + std::iter::once("member-race"), + dispatch_hook.as_ref(), + ) + }); + hook.entered.wait(); + + let fence_started = Arc::new(Barrier::new(2)); + let fence_started_in_thread = Arc::clone(&fence_started); + let fence = std::thread::spawn(move || { + fence_started_in_thread.wait(); + crate::coordination::agent_org_runs::establish_conversation_delete_fence(root_session_id) + }); + fence_started.wait(); + let conn = get_connection().expect("read before callback release"); + let fenced: bool = conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM agent_org_conversation_delete_fences WHERE root_session_id=?1)", + [root_session_id], + |row| row.get(0), + ) + .unwrap(); + assert!(!fenced, "fence must wait for the in-flight callback gate"); + drop(conn); + + hook.release.wait(); + dispatch.join().unwrap().unwrap(); + fence.join().unwrap().unwrap(); + assert_eq!(hook.calls.load(Ordering::SeqCst), 1); + let conn = get_connection().expect("read committed fence"); + let fenced: bool = conn + .query_row( + "SELECT EXISTS(SELECT 1 FROM agent_org_conversation_delete_fences WHERE root_session_id=?1)", + [root_session_id], + |row| row.get(0), + ) + .unwrap(); + assert!(fenced); +} + #[test] fn one_failed_run_does_not_skip_later_runs() { let first = fake_run("run-first"); diff --git a/src-tauri/crates/agent-core/src/core/session/gateway_pipeline.rs b/src-tauri/crates/agent-core/src/core/session/gateway_pipeline.rs index ba769e8f5..4847237ef 100644 --- a/src-tauri/crates/agent-core/src/core/session/gateway_pipeline.rs +++ b/src-tauri/crates/agent-core/src/core/session/gateway_pipeline.rs @@ -24,6 +24,12 @@ pub async fn process_gateway_message( ide_context: Option<&IdeContext>, app_handle: Option, ) -> Result, String> { + let submission_scope = session.agent_org_submission_scope(); + let _submission = crate::coordination::agent_org_runs::admit_agent_org_submission( + &submission_scope, + &session.id, + ) + .await?; let preview: String = crate::utils::safe_truncate_chars_to_string(&msg.content, 80); info!( "Processing message from {}:{}: {}...", diff --git a/src-tauri/crates/agent-core/src/core/session/launch/mod.rs b/src-tauri/crates/agent-core/src/core/session/launch/mod.rs index 0be21edaa..a2782a61b 100644 --- a/src-tauri/crates/agent-core/src/core/session/launch/mod.rs +++ b/src-tauri/crates/agent-core/src/core/session/launch/mod.rs @@ -216,6 +216,7 @@ async fn generate_title_before_first_turn( /// coupling this decoupling removes. The title task persists the name and /// emits `session:renamed` when it finishes; the first turn proceeds /// concurrently regardless of when (or whether) the title resolves. +#[allow(clippy::too_many_arguments)] fn spawn_session_title_generation( state: AgentAppState, session_id: String, @@ -224,8 +225,27 @@ fn spawn_session_title_generation( model: Option, native_harness_type: Option, content: String, + agent_org_run_id: Option, ) { + let title_submission = agent_org_run_id + .as_ref() + .map(|_| crate::coordination::agent_org_runs::AgentOrgSubmissionLease::begin(&session_id)); tokio::spawn(async move { + let _title_submission = title_submission; + if let Some(run_id) = agent_org_run_id.as_deref() { + if let Err(error) = + crate::coordination::agent_org_runs::recheck_known_agent_org_submission(run_id) + .await + { + tracing::debug!( + session_id = %session_id, + run_id = %run_id, + %error, + "[session_title] skipped title generation for deleting Agent Org" + ); + return; + } + } generate_title_before_first_turn( &state, &session_id, @@ -645,6 +665,7 @@ pub(crate) async fn launch_rust_agent_run( model_for_send.clone(), native_harness_type_for_send, content_for_send.clone(), + agent_org_run_id_for_background.clone(), ); let send_result = send_initial_turn( &state_for_background, @@ -766,6 +787,7 @@ pub(crate) async fn launch_rust_agent_run( model_for_send.clone(), native_harness_type_for_send, content_for_send.clone(), + agent_org_run_id_for_send.clone(), ); // A plain (non-org) launch's first message IS the user's real diff --git a/src-tauri/crates/agent-core/src/core/session/scheduler.rs b/src-tauri/crates/agent-core/src/core/session/scheduler.rs index 29e6fed79..3dfa0482d 100644 --- a/src-tauri/crates/agent-core/src/core/session/scheduler.rs +++ b/src-tauri/crates/agent-core/src/core/session/scheduler.rs @@ -48,6 +48,10 @@ use super::turn::streaming::{ broadcast_agent_error_structured, classify_streaming_error_message, StreamingError, }; use crate::bus::broadcast_event; +use crate::coordination::agent_org_runs::{ + admit_agent_org_submission, AgentOrgSubmissionPolicy, AgentOrgSubmissionScope, + SharedAgentOrgSubmissionScope, +}; // ============================================ // Scheduled Message @@ -185,6 +189,7 @@ struct SchedulerInner { pub struct DialogScheduler { /// Session this scheduler belongs to. session_id: String, + submission_scope: SharedAgentOrgSubmissionScope, /// Channel capacity. capacity: usize, /// Lazily initialized sender. `None` until first `enqueue()`. @@ -209,8 +214,23 @@ impl DialogScheduler { /// Once full, `enqueue` returns an error so the caller can surface /// "session queue full" to the user. pub fn new(session_id: impl Into, capacity: usize) -> Self { + Self::new_with_submission_scope( + session_id, + capacity, + Arc::new(AgentOrgSubmissionPolicy::new( + AgentOrgSubmissionScope::Ordinary, + )), + ) + } + + pub(crate) fn new_with_submission_scope( + session_id: impl Into, + capacity: usize, + submission_scope: SharedAgentOrgSubmissionScope, + ) -> Self { Self { session_id: session_id.into(), + submission_scope, capacity, inner: TokioMutex::new(None), pending: Arc::new(AtomicUsize::new(0)), @@ -258,6 +278,8 @@ impl DialogScheduler { /// **Note**: This method is `async` because lazy initialization requires /// holding a lock to spawn the worker on first use. pub async fn enqueue(&self, mut msg: ScheduledMessage) -> Result { + let _submission = + admit_agent_org_submission(&self.submission_scope, &self.session_id).await?; let tx = self.ensure_initialized().await; let message_id = msg.message_id.clone(); @@ -822,4 +844,56 @@ mod tests { assert_eq!(executed.load(Ordering::SeqCst), 1); assert_eq!(scheduler.pending_count(), 0); } + + #[tokio::test] + async fn fenced_agent_org_rejects_turn_and_maintenance_enqueue() { + let _sandbox = test_helpers::test_env::sandbox(); + let conn = database::db::get_connection().expect("test database"); + crate::foundation::persistence::test_schema::ensure_agent_sessions_schema(&conn); + crate::coordination::init_agent_org_schemas(&conn).expect("Agent Org schemas"); + let now = chrono::Utc::now().to_rfc3339(); + conn.execute( + "INSERT INTO agent_org_runs ( + id, org_id, coordinator_agent_id, root_session_id, entry_mode, + status, created_at, updated_at + ) VALUES ('scheduler-fenced-run', 'org', 'coordinator', + 'scheduler-fenced-root', 'standalone_session', 'running', ?1, ?1)", + [&now], + ) + .expect("seed Run"); + conn.execute( + "INSERT INTO agent_org_conversation_delete_fences ( + root_session_id, created_at, updated_at + ) VALUES ('scheduler-fenced-root', ?1, ?1)", + [&now], + ) + .expect("seed fence"); + drop(conn); + + let scope = Arc::new(AgentOrgSubmissionPolicy::new( + AgentOrgSubmissionScope::Run { + run_id: "scheduler-fenced-run".to_string(), + }, + )); + let scheduler = + DialogScheduler::new_with_submission_scope("scheduler-fenced-root", 8, scope); + for kind in [ScheduledKind::Turn, ScheduledKind::Maintenance] { + let error = scheduler + .enqueue(ScheduledMessage { + kind, + message_id: format!("fenced-{kind:?}"), + generation: 0, + client_message_id: None, + turn_intent_id: String::new(), + org_run_id: Some("scheduler-fenced-run".to_string()), + content: String::new(), + execute: Box::new(|| Box::pin(async { Ok(String::new()) })), + }) + .await + .expect_err("fenced enqueue must fail"); + assert!(error.starts_with("conversation_deleting:"), "{error}"); + } + assert_eq!(scheduler.pending_count(), 0); + assert!(!scheduler.is_processing()); + } } diff --git a/src-tauri/crates/agent-core/src/core/session/turn/entry.rs b/src-tauri/crates/agent-core/src/core/session/turn/entry.rs index efa0ea0f3..6fcd24cbc 100644 --- a/src-tauri/crates/agent-core/src/core/session/turn/entry.rs +++ b/src-tauri/crates/agent-core/src/core/session/turn/entry.rs @@ -129,6 +129,61 @@ mod tests { "expected newline tail to become the user task" ); } + + #[tokio::test] + async fn process_message_rejects_fenced_agent_org_before_runtime_access() { + let _sandbox = test_helpers::test_env::sandbox(); + let conn = database::db::get_connection().expect("test database"); + crate::foundation::persistence::test_schema::ensure_agent_sessions_schema(&conn); + crate::coordination::init_agent_org_schemas(&conn).expect("Agent Org schemas"); + let now = chrono::Utc::now().to_rfc3339(); + conn.execute( + "INSERT INTO agent_org_runs ( + id, org_id, coordinator_agent_id, root_session_id, entry_mode, + status, created_at, updated_at + ) VALUES ('process-fenced-run', 'org', 'coordinator', + 'process-fenced-root', 'standalone_session', 'running', ?1, ?1)", + [&now], + ) + .expect("seed Run"); + conn.execute( + "INSERT INTO agent_org_conversation_delete_fences ( + root_session_id, created_at, updated_at + ) VALUES ('process-fenced-root', ?1, ?1)", + [&now], + ) + .expect("seed fence"); + drop(conn); + + let session = std::sync::Arc::new(crate::state::AgentSession::new( + "process-fenced-root".to_string(), + crate::definitions::sde_agent(), + )); + session.set_agent_org_submission_scope( + crate::coordination::agent_org_runs::AgentOrgSubmissionScope::Run { + run_id: "process-fenced-run".to_string(), + }, + ); + let error = super::process_message( + session, + super::TurnInput { + content: "blocked".to_string(), + display_text: None, + agent_mode: None, + images: None, + ide_context: None, + is_resume: false, + channel: None, + chat_id: None, + turn_id: None, + turn_intent_id: "process-fenced-intent".to_string(), + }, + None, + ) + .await + .expect_err("fenced process_message must fail before runtime access"); + assert!(error.starts_with("conversation_deleting:"), "{error}"); + } } // ============================================ @@ -149,6 +204,12 @@ pub async fn process_message( input: TurnInput, app_handle: Option, ) -> Result { + let submission_scope = session.agent_org_submission_scope(); + let _submission = crate::coordination::agent_org_runs::admit_agent_org_submission( + &submission_scope, + &session.id, + ) + .await?; let runtime = session .runtime .read() diff --git a/src-tauri/crates/agent-core/src/core/session/turn/processor/inbox_drain/tests.rs b/src-tauri/crates/agent-core/src/core/session/turn/processor/inbox_drain/tests.rs index 1a9cef2b0..30d773d5e 100644 --- a/src-tauri/crates/agent-core/src/core/session/turn/processor/inbox_drain/tests.rs +++ b/src-tauri/crates/agent-core/src/core/session/turn/processor/inbox_drain/tests.rs @@ -315,8 +315,8 @@ fn shared_agent_id_member_session_drains_only_its_member_inbox() { #[test] fn user_intervention_pauses_member_inbox_drain_without_marking_read() { let _sandbox = test_helpers::test_env::sandbox(); - let run_id = format!("run-{}", uuid::Uuid::new_v4()); - let ctx = ctx_for_with_member(&run_id, "worker-1", "Worker 1"); + let ctx = running_ctx_for_members(&[("member-worker-1", "worker-1", "Worker 1")]); + let run_id = ctx.run_id.clone(); AgentInboxStore::insert(InsertInboxParams { recipient_agent_id: "worker-1".into(), @@ -359,8 +359,8 @@ fn user_intervention_pauses_member_inbox_drain_without_marking_read() { #[test] fn return_to_work_restores_member_inbox_drain() { let _sandbox = test_helpers::test_env::sandbox(); - let run_id = format!("run-{}", uuid::Uuid::new_v4()); - let ctx = ctx_for_with_member(&run_id, "worker-1", "Worker 1"); + let ctx = running_ctx_for_members(&[("member-worker-1", "worker-1", "Worker 1")]); + let run_id = ctx.run_id.clone(); AgentInboxStore::insert(InsertInboxParams { recipient_agent_id: "worker-1".into(), diff --git a/src-tauri/crates/agent-core/src/init/mod.rs b/src-tauri/crates/agent-core/src/init/mod.rs index 6dbd53d09..86af6743e 100644 --- a/src-tauri/crates/agent-core/src/init/mod.rs +++ b/src-tauri/crates/agent-core/src/init/mod.rs @@ -239,7 +239,28 @@ fn load_agent_org_context( state: &AgentAppState, session_id: &str, run_hint: Option<&str>, + cached_context: Option<&crate::coordination::agent_org_runs::AgentOrgRunContext>, + cached_scope: Option, + persisted_session: Option<&crate::session::persistence::UnifiedSessionRecord>, ) -> Result, String> { + if run_hint.is_none() { + if let Some(context) = cached_context { + return Ok(Some(context.clone())); + } + if cached_scope + == Some(crate::coordination::agent_org_runs::AgentOrgSubmissionScope::Ordinary) + { + return Ok(None); + } + if persisted_session.is_some_and(|record| { + record.session_type != crate::session::persistence::session_type::ORG_MEMBER + && record.org_member_id.is_none() + }) { + // A canonical ordinary row is conclusive. Do not make SDE/OS + // depend on Agent Org schema availability or fence I/O. + return Ok(None); + } + } let Some(handle) = state.app_handle.as_ref() else { tracing::debug!( session_id = %session_id, @@ -328,7 +349,43 @@ async fn ensure_session_initialized( _ if !resolved.selected_model_id.is_empty() => Some(resolved.selected_model_id.clone()), _ => None, }; - let agent_org_context = load_agent_org_context(state, session_id, agent_org_run_hint)?; + let cached_session = state.get_session(session_id).await; + let cached_runtime = match cached_session.as_ref() { + Some(session) => session.get_runtime().await, + None => None, + }; + let mut persisted_session = if cached_runtime.is_none() { + crate::session::persistence::get_session(session_id) + .map_err(|err| format!("failed to load session ownership for {session_id}: {err}"))? + } else { + None + }; + let agent_org_context = load_agent_org_context( + state, + session_id, + agent_org_run_hint, + cached_runtime + .as_ref() + .and_then(|runtime| runtime.agent_org_context.as_ref()), + cached_session + .as_ref() + .map(|session| session.agent_org_submission_scope().snapshot()), + persisted_session.as_ref(), + )?; + let submission_run_id = agent_org_run_hint.map(str::to_string).or_else(|| { + agent_org_context + .as_ref() + .map(|context| context.run_id.clone()) + }); + let submission_scope = match submission_run_id.as_ref() { + Some(run_id) => crate::coordination::agent_org_runs::AgentOrgSubmissionScope::Run { + run_id: run_id.clone(), + }, + None => crate::coordination::agent_org_runs::AgentOrgSubmissionScope::Ordinary, + }; + if let Some(session) = state.get_session(session_id).await { + session.set_agent_org_submission_scope(submission_scope.clone()); + } // Fast path: re-entrant init for an already-running session. if let Some(existing) = fast_path::try_reuse_existing( @@ -346,6 +403,16 @@ async fn ensure_session_initialized( return Ok(existing); } + let _runtime_init_submission = match submission_run_id.as_deref() { + Some(run_id) => Some( + crate::coordination::agent_org_runs::admit_known_agent_org_submission( + session_id, run_id, + ) + .await?, + ), + None => None, + }; + // Slow path: we're about to (re)build a runtime — model is now required. let model = requested_model.ok_or("model is required: not provided by caller and not set in config")?; @@ -413,6 +480,7 @@ async fn ensure_session_initialized( .get_session(session_id) .await .ok_or_else(|| format!("Session {} missing after registration", session_id))?; + session_handle.set_agent_org_submission_scope(submission_scope); // Capability derivation — single pass over `resolved` for all gates. let cap_flags = capabilities::CapabilityFlags::from_resolved(&resolved); @@ -469,9 +537,11 @@ async fn ensure_session_initialized( controller.config() }; - let session_record = crate::session::persistence::get_session(session_id) - .ok() - .flatten(); + if persisted_session.is_none() { + persisted_session = crate::session::persistence::get_session(session_id) + .map_err(|err| format!("failed to load session metadata for {session_id}: {err}"))?; + } + let session_record = persisted_session; let agent_org_current_member_id = match agent_org_context.as_ref() { Some(context) if context.root_session_id.as_deref() == Some(session_id) => { Some(crate::coordination::agent_org_runs::COORDINATOR_MEMBER_ID.to_string()) @@ -644,7 +714,7 @@ async fn ensure_session_initialized( agent_definition_id, }, ) - .await; + .await?; runtime_assemble::mark_running_for_gateway(state, cap_flags.has_gateway, &account_id).await; runtime_assemble::register_in_file_registry(session_id, &log_prefix, &model, &workspace_root); diff --git a/src-tauri/crates/agent-core/src/init/runtime_assemble.rs b/src-tauri/crates/agent-core/src/init/runtime_assemble.rs index e0adbc9cf..dfd50895e 100644 --- a/src-tauri/crates/agent-core/src/init/runtime_assemble.rs +++ b/src-tauri/crates/agent-core/src/init/runtime_assemble.rs @@ -120,7 +120,13 @@ pub(super) struct AssembleParams { pub(super) async fn install_runtime( session_handle: &AgentSession, params: AssembleParams, -) -> Arc { +) -> Result, String> { + // The slow build may have overlapped deletion after its first check. + // Re-read the durable fence at the sole runtime installation boundary. + crate::coordination::agent_org_runs::recheck_agent_org_submission( + &session_handle.agent_org_submission_scope(), + ) + .await?; let runtime = Arc::new(SessionRuntime { provider: params.provider, tool_registry: params.final_registry, @@ -141,7 +147,7 @@ pub(super) async fn install_runtime( agent_definition_id: params.agent_definition_id, }); session_handle.set_runtime(Arc::clone(&runtime)).await; - runtime + Ok(runtime) } /// Side-effect: mark the app as "running" + remember the active account. diff --git a/src-tauri/crates/agent-core/src/state/commands/session/compaction.rs b/src-tauri/crates/agent-core/src/state/commands/session/compaction.rs index b533e56bf..d2359cb84 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/compaction.rs +++ b/src-tauri/crates/agent-core/src/state/commands/session/compaction.rs @@ -129,6 +129,7 @@ pub async fn agent_session_manual_compact( state.inner(), &session_id, super::identity::IdentityOverrides::default(), + None, ) .await { diff --git a/src-tauri/crates/agent-core/src/state/commands/session/identity.rs b/src-tauri/crates/agent-core/src/state/commands/session/identity.rs index 0db3a7563..1df847a1c 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/identity.rs +++ b/src-tauri/crates/agent-core/src/state/commands/session/identity.rs @@ -32,6 +32,7 @@ pub(super) struct SessionIdentity { pub(super) account_id: Option, pub(super) native_harness_type: Option, pub(super) workspace_root: PathBuf, + pub(super) agent_org_run_id: Option, } /// Caller-supplied overrides. Fields that are `None` are resolved from @@ -72,6 +73,7 @@ pub(super) async fn resolve_session_identity( state: &AgentAppState, session_id: &str, overrides: IdentityOverrides, + agent_org_run_hint: Option<&str>, ) -> Result { let personal_ws = crate::definitions::prefix_lookup::uses_personal_workspace(session_id); @@ -206,11 +208,56 @@ pub(super) async fn resolve_session_identity( } }; + // Agent Org ownership follows the same precedence as runtime admission: + // reliable runtime context, then the already-loaded persisted type, and + // only a cold unknown falls back to PR1's exact mapping. Ordinary rows + // return without touching Agent Org tables. + let agent_org_run_id = if let Some(run_id) = agent_org_run_hint { + Some(run_id.to_string()) + } else if let Some(runtime) = cached_runtime.as_ref() { + runtime + .agent_org_context + .as_ref() + .map(|context| context.run_id.clone()) + } else { + let needs_exact_mapping = db_record + .as_ref() + .map(|record| { + record.session_type == session_persistence::session_type::ORG_MEMBER + || record.org_member_id.is_some() + }) + .unwrap_or(true); + if needs_exact_mapping { + let sid = session_id.to_string(); + let record = db_record.clone(); + let scope = tokio::task::spawn_blocking(move || match record { + Some(record) => { + crate::coordination::agent_org_runs::submission_scope_for_loaded_session( + &record, + ) + } + None => crate::coordination::agent_org_runs::exact_submission_scope(&sid), + }) + .await + .map_err(|err| format!("Agent Org ownership task failed: {err}"))??; + match scope { + crate::coordination::agent_org_runs::AgentOrgSubmissionScope::Run { run_id } => { + Some(run_id) + } + crate::coordination::agent_org_runs::AgentOrgSubmissionScope::Ordinary + | crate::coordination::agent_org_runs::AgentOrgSubmissionScope::Unknown => None, + } + } else { + None + } + }; + Ok(SessionIdentity { model, account_id, workspace_root, native_harness_type, + agent_org_run_id, }) } diff --git a/src-tauri/crates/agent-core/src/state/commands/session/message/org_wake.rs b/src-tauri/crates/agent-core/src/state/commands/session/message/org_wake.rs index a05fca57b..aecb14de2 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/message/org_wake.rs +++ b/src-tauri/crates/agent-core/src/state/commands/session/message/org_wake.rs @@ -14,9 +14,13 @@ pub(super) fn promote_agent_org_wake_session_to_running( run_id: &str, session_id: &str, ) -> Result { - use crate::coordination::agent_org_runs::AgentOrgRunStatus; + use crate::coordination::agent_org_runs::{is_run_writable_with_connection, AgentOrgRunStatus}; use crate::session::SessionStatus; + if !is_run_writable_with_connection(conn, run_id)? { + return Ok(0); + } + let wakeable = SessionStatus::AGENT_ORG_WAKEABLE; let now = chrono::Utc::now().to_rfc3339(); conn.execute( @@ -71,6 +75,10 @@ pub(super) fn promote_agent_org_direct_session_to_running( run_id: &str, session_id: &str, ) -> Result { + if !crate::coordination::agent_org_runs::is_run_writable_with_connection(conn, run_id)? { + return Ok(0); + } + conn.execute( "UPDATE agent_sessions SET status=?1, updated_at=?2 diff --git a/src-tauri/crates/agent-core/src/state/commands/session/message/send.rs b/src-tauri/crates/agent-core/src/state/commands/session/message/send.rs index 27a1906ff..19c04118c 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/message/send.rs +++ b/src-tauri/crates/agent-core/src/state/commands/session/message/send.rs @@ -84,6 +84,21 @@ pub(crate) async fn send_message_impl( intent_org_run_id: Option, source: TurnIntentBridgeSource, ) -> Result { + let explicit_run_id = org_wake_run_id + .as_ref() + .or(intent_org_run_id.as_ref()) + .cloned(); + let explicit_submission = match explicit_run_id.as_deref() { + Some(run_id) => Some( + crate::coordination::agent_org_runs::admit_known_agent_org_submission( + &session_id, + run_id, + ) + .await?, + ), + None => None, + }; + // Canonical user-intent id: callers that already mint one at the // submit boundary pass it through; legacy / internal callers that // don't (mobile remote, wake hook, plan-approval re-entry) get a @@ -106,7 +121,8 @@ pub(crate) async fn send_message_impl( ); // ── 1. Resolve session identity (unified — single code path) ───────── - let identity = resolve_session_identity(state, &session_id, overrides).await?; + let identity = + resolve_session_identity(state, &session_id, overrides, explicit_run_id.as_deref()).await?; // Goal loop: a real user submission becomes (or replaces) the // session's standing goal and resets the continuation counter. @@ -124,6 +140,18 @@ pub(crate) async fn send_message_impl( let effective_account_id = identity.account_id; let effective_workspace_root = identity.workspace_root; let effective_native_harness_type = identity.native_harness_type; + let identity_agent_org_run_id = identity.agent_org_run_id; + let submission_run_id = explicit_run_id.or(identity_agent_org_run_id); + let _inferred_submission = match (explicit_submission.is_none(), submission_run_id.as_deref()) { + (true, Some(run_id)) => Some( + crate::coordination::agent_org_runs::admit_known_agent_org_submission( + &session_id, + run_id, + ) + .await?, + ), + _ => None, + }; // ── 2. Ensure session is initialized (lazy runtime creation) ───────── let launch_spec = crate::init::launch_spec::AgentLaunchSpec::from_session_sources( @@ -135,7 +163,7 @@ pub(crate) async fn send_message_impl( effective_native_harness_type, ) .await? - .with_agent_org_run_hint(intent_org_run_id.clone()); + .with_agent_org_run_hint(submission_run_id); let runtime = crate::init::init_session(state, launch_spec).await?; diff --git a/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/lifecycle.rs b/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/lifecycle.rs index 8c567f979..059a58692 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/lifecycle.rs +++ b/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/lifecycle.rs @@ -15,7 +15,8 @@ use crate::coordination::agent_inbox::{ AgentInboxStore, AgentMessage, InsertInboxParams, SYSTEM_SENDER_ID, }; use crate::coordination::agent_org_runs::{ - AgentOrgRunContext, AgentOrgRunStore, COORDINATOR_MEMBER_ID, + ensure_conversation_writable_with_connection, AgentOrgRunContext, AgentOrgRunStore, + COORDINATOR_MEMBER_ID, }; use crate::state::control_flow::CancelReason; use crate::state::AgentAppState; @@ -185,16 +186,23 @@ pub(super) fn resume_agent_org_context_sync( let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; - let status: Option = tx + let run: Option<(Option, String)> = tx .query_row( - "SELECT status FROM agent_org_runs WHERE id=?1", + "SELECT root_session_id, status FROM agent_org_runs WHERE id=?1", params![&context.run_id], - |row| row.get(0), + |row| Ok((row.get(0)?, row.get(1)?)), ) .optional() .map_err(|err| err.to_string())?; - let transitioned = status.as_deref() == Some("paused"); - let run_is_running = transitioned || status.as_deref() == Some("running"); + if let Some(root_session_id) = run + .as_ref() + .and_then(|(root_session_id, _)| root_session_id.as_deref()) + { + ensure_conversation_writable_with_connection(&tx, root_session_id)?; + } + let status = run.as_ref().map(|(_, status)| status.as_str()); + let transitioned = status == Some("paused"); + let run_is_running = transitioned || status == Some("running"); if transitioned { tx.execute( "UPDATE agent_org_runs diff --git a/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/run_view.rs b/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/run_view.rs index 7bf19b50d..ad7103b45 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/run_view.rs +++ b/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/run_view.rs @@ -206,7 +206,7 @@ fn build_agent_org_run_view( .facts .run_status .ok_or_else(|| format!("Agent Org run {} no longer exists", context.run_id))?; - let run_status = run_status_value.as_str().to_string(); + let run_status = project_run_status(run_status_value).to_string(); let task_page = AgentOrgTaskStore::list_summary_page_with_connection( &tx, @@ -335,6 +335,9 @@ pub(super) fn project_run_phase( pending_plan_approvals: &[AgentOrgPlanApprovalSummary], ) -> AgentOrgRunPhase { match run_status { + // Starting is intentionally projected through the existing wire shape; + // it has not begun dispatching work yet. + AgentOrgRunStatus::Starting => AgentOrgRunPhase::Coordinating, AgentOrgRunStatus::Paused => AgentOrgRunPhase::Paused, AgentOrgRunStatus::Completed => AgentOrgRunPhase::Completed, AgentOrgRunStatus::Failed => AgentOrgRunPhase::Failed, @@ -379,6 +382,16 @@ pub(super) fn project_run_phase( } } +/// `starting` is an internal persistence phase added for deletion safety. The +/// existing frontend wire contract has no such status, so project it as the +/// pre-existing non-terminal `running` value until materialization completes. +pub(super) fn project_run_status(status: AgentOrgRunStatus) -> &'static str { + match status { + AgentOrgRunStatus::Starting => AgentOrgRunStatus::Running.as_str(), + status => status.as_str(), + } +} + pub(super) fn tasks_for_context( context: &AgentOrgRunContext, tasks: Vec, diff --git a/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/tests.rs b/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/tests.rs index b5bf8d3a4..f830aef09 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/tests.rs +++ b/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/tests.rs @@ -24,6 +24,14 @@ use crate::coordination::agent_org_runs::{ use crate::coordination::agent_org_tasks::{Task, TaskExecutionMode, TaskStatus, TaskSummary}; use crate::definitions::orgs::HierarchyMode; +#[test] +fn starting_run_status_projects_through_the_existing_wire_contract() { + assert_eq!( + project_run_status(AgentOrgRunStatus::Starting), + AgentOrgRunStatus::Running.as_str() + ); +} + fn context_with_shared_member_agent_id() -> AgentOrgRunContext { AgentOrgRunContext { run_id: "run-shared-agent".to_string(), @@ -327,7 +335,7 @@ fn resume_wake_requires_unread_inbox() { #[test] fn terminal_group_message_writes_neither_inbox_nor_intervention_clear() { let _sandbox = test_helpers::test_env::sandbox(); - let context = prepare_command_run("completed"); + let context = prepare_command_run("running"); AgentMemberInterventionStore::enter(EnterMemberInterventionParams { org_run_id: context.run_id.clone(), member_id: "member-planner".to_string(), @@ -337,6 +345,13 @@ fn terminal_group_message_writes_neither_inbox_nor_intervention_clear() { ttl_secs: 60, }) .expect("enter intervention"); + let conn = get_connection().expect("db connection"); + conn.execute( + "UPDATE agent_org_runs SET status='completed' WHERE id=?1", + [&context.run_id], + ) + .expect("complete run after entering intervention"); + drop(conn); let error = persist_group_chat_message( &context, @@ -525,6 +540,44 @@ fn paused_resume_and_coordinator_seed_commit_or_rollback_together() { assert_eq!(inbox_count_for_member(&context, COORDINATOR_MEMBER_ID), 1); } +#[test] +fn resume_refuses_persistently_fenced_root_even_if_run_is_paused() { + let _sandbox = test_helpers::test_env::sandbox(); + let context = prepare_command_run("paused"); + let conn = get_connection().expect("db connection"); + crate::coordination::agent_org_plan_approvals::init_schema(&conn) + .expect("plan approval schema"); + drop(conn); + crate::coordination::agent_org_runs::establish_conversation_delete_fence( + context + .root_session_id + .as_deref() + .expect("command fixture root"), + ) + .expect("establish durable root fence"); + let conn = get_connection().expect("db connection"); + conn.execute( + "UPDATE agent_org_runs SET status='paused' WHERE id=?1", + params![&context.run_id], + ) + .expect("keep run paused to isolate root-fence behavior"); + drop(conn); + + let error = resume_agent_org_context_sync(&context, true) + .expect_err("a fenced conversation cannot be resumed"); + assert!(error.starts_with("conversation_deleting:"), "{error}"); + let conn = get_connection().expect("db connection"); + let status: String = conn + .query_row( + "SELECT status FROM agent_org_runs WHERE id=?1", + params![&context.run_id], + |row| row.get(0), + ) + .expect("load fenced run status"); + assert_eq!(status, "paused"); + assert_eq!(inbox_count_for_member(&context, COORDINATOR_MEMBER_ID), 0); +} + #[test] fn explicit_resume_of_running_run_repairs_unread_without_duplicate_seed() { let _sandbox = test_helpers::test_env::sandbox(); @@ -648,10 +701,7 @@ fn return_to_work_rolls_back_intervention_clear_when_boundary_capture_fails() { #[test] fn group_chat_target_clear_exits_direct_intervention() { let _sandbox = test_helpers::test_env::sandbox(); - let conn = get_connection().expect("db connection"); - crate::coordination::agent_member_interventions::init_schema(&conn) - .expect("intervention schema"); - let context = context_with_shared_member_agent_id(); + let context = prepare_command_run("running"); AgentMemberInterventionStore::enter(EnterMemberInterventionParams { org_run_id: context.run_id.clone(), diff --git a/src-tauri/crates/agent-core/src/state/commands/session/persistence.rs b/src-tauri/crates/agent-core/src/state/commands/session/persistence.rs index be391207f..aa0f73733 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/persistence.rs +++ b/src-tauri/crates/agent-core/src/state/commands/session/persistence.rs @@ -104,46 +104,34 @@ pub async fn agent_delete_session( }); }; - let (plan, quiesced_runtime_session_ids) = if matches!( - plan.run_status, - crate::coordination::agent_org_runs::AgentOrgRunStatus::Running - | crate::coordination::agent_org_runs::AgentOrgRunStatus::Paused - | crate::coordination::agent_org_runs::AgentOrgRunStatus::Cancelled - ) { - let fenced_plan = - tokio::task::spawn_blocking(move || establish_agent_org_delete_fence(&plan)) - .await - .map_err(|err| format!("Agent Org deletion fence worker failed: {err}"))??; - let quiesced_runtime_session_ids = if fenced_plan.run_status - == crate::coordination::agent_org_runs::AgentOrgRunStatus::Cancelled - { - stop_agent_org_runtime_sessions(&state, &fenced_plan).await? - } else { - ensure_agent_org_runtime_sessions_idle(&state, &fenced_plan).await?; - HashSet::new() - }; - let root_session_id = fenced_plan.root_session_id.clone(); - let current_plan = tokio::task::spawn_blocking(move || { - let conn = get_connection().map_err(|err| err.to_string())?; - load_agent_org_session_delete_plan(&conn, &root_session_id)?.ok_or_else(|| { - format!( - "Refusing to delete Agent Org root {root_session_id}: ownership disappeared while stopping" - ) - }) - }) + let fenced_plan = tokio::task::spawn_blocking(move || establish_agent_org_delete_fence(&plan)) .await - .map_err(|err| format!("Agent Org post-stop planning worker failed: {err}"))??; - if !agent_org_delete_topology_matches(&fenced_plan, ¤t_plan) { - return Err(format!( - "Refusing to delete Agent Org run {}: session hierarchy changed while stopping", - fenced_plan.run_id - )); - } - (current_plan, quiesced_runtime_session_ids) + .map_err(|err| format!("Agent Org deletion fence worker failed: {err}"))??; + let quiesced_runtime_session_ids = if fenced_plan.run_status + == crate::coordination::agent_org_runs::AgentOrgRunStatus::Cancelled + { + stop_agent_org_runtime_sessions(&state, &fenced_plan).await? } else { - ensure_agent_org_runtime_sessions_idle(&state, &plan).await?; - (plan, HashSet::new()) + wait_for_agent_org_runtime_sessions_idle(&state, &fenced_plan).await?; + HashSet::new() }; + let root_session_id = fenced_plan.root_session_id.clone(); + let plan = tokio::task::spawn_blocking(move || { + let conn = get_connection().map_err(|err| err.to_string())?; + load_agent_org_session_delete_plan(&conn, &root_session_id)?.ok_or_else(|| { + format!( + "Refusing to delete Agent Org root {root_session_id}: ownership disappeared while stopping" + ) + }) + }) + .await + .map_err(|err| format!("Agent Org post-stop planning worker failed: {err}"))??; + if !agent_org_delete_topology_matches(&fenced_plan, &plan) { + return Err(format!( + "Refusing to delete Agent Org run {}: session hierarchy changed while stopping", + fenced_plan.run_id + )); + } validate_agent_org_delete_ready(&plan, &quiesced_runtime_session_ids)?; ensure_agent_org_runtime_sessions_idle(&state, &plan).await?; @@ -418,12 +406,16 @@ fn establish_agent_org_delete_fence( )); } + let cancelled_run_ids = + crate::coordination::agent_org_runs::establish_conversation_delete_fence_with_connection( + &tx, + ¤t_plan.root_session_id, + )?; let changed = match current_plan.run_status { - crate::coordination::agent_org_runs::AgentOrgRunStatus::Running + crate::coordination::agent_org_runs::AgentOrgRunStatus::Starting + | crate::coordination::agent_org_runs::AgentOrgRunStatus::Running | crate::coordination::agent_org_runs::AgentOrgRunStatus::Paused => { - let changed = - AgentOrgRunStore::cancel_for_delete_with_connection(&tx, ¤t_plan.run_id)?; - if !changed { + if !cancelled_run_ids.contains(¤t_plan.run_id) { return Err(format!( "Refusing to delete Agent Org run {}: run status changed before cancellation", current_plan.run_id @@ -519,6 +511,16 @@ async fn agent_org_runtime_blockers( blockers } +fn agent_org_submission_blockers(plan: &AgentOrgSessionDeletePlan) -> Vec { + plan.sessions + .iter() + .filter(|node| { + crate::coordination::agent_org_runs::agent_org_submission_in_progress(&node.session_id) + }) + .map(|node| format!("{}(submission_in_progress=true)", node.session_id)) + .collect() +} + async fn stop_agent_org_runtime_sessions( state: &AgentAppState, plan: &AgentOrgSessionDeletePlan, @@ -531,21 +533,22 @@ async fn stop_agent_org_runtime_sessions_with_timeout( plan: &AgentOrgSessionDeletePlan, timeout: Duration, ) -> Result, String> { - let runtime_sessions = agent_org_runtime_sessions(state, plan).await; - let runtime_session_ids = runtime_sessions - .iter() - .map(|(session_id, _)| session_id.clone()) - .collect::>(); - - for (_, session) in &runtime_sessions { - session - .cancel_active_turn(CancelReason::AgentOrgDelete) - .await; - } - + let mut runtime_session_ids = HashSet::new(); let deadline = tokio::time::Instant::now() + timeout; loop { - let blockers = agent_org_runtime_blockers(&runtime_sessions).await; + // Initialization that held a submission lease may install its runtime + // after fencing began. Refresh the registry each pass so that late + // runtime is cancelled before its lease can let deletion continue. + let runtime_sessions = agent_org_runtime_sessions(state, plan).await; + for (session_id, session) in &runtime_sessions { + if runtime_session_ids.insert(session_id.clone()) { + session + .cancel_active_turn(CancelReason::AgentOrgDelete) + .await; + } + } + let mut blockers = agent_org_runtime_blockers(&runtime_sessions).await; + blockers.extend(agent_org_submission_blockers(plan)); if blockers.is_empty() { return Ok(runtime_session_ids); } @@ -560,12 +563,36 @@ async fn stop_agent_org_runtime_sessions_with_timeout( } } +async fn wait_for_agent_org_runtime_sessions_idle( + state: &AgentAppState, + plan: &AgentOrgSessionDeletePlan, +) -> Result<(), String> { + let deadline = tokio::time::Instant::now() + AGENT_ORG_DELETE_STOP_TIMEOUT; + loop { + let runtime_sessions = agent_org_runtime_sessions(state, plan).await; + let mut blockers = agent_org_runtime_blockers(&runtime_sessions).await; + blockers.extend(agent_org_submission_blockers(plan)); + if blockers.is_empty() { + return Ok(()); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!( + "Timed out waiting for Agent Org run {} submissions before deletion: {}", + plan.run_id, + blockers.join(", ") + )); + } + tokio::time::sleep(AGENT_ORG_DELETE_STOP_POLL_INTERVAL).await; + } +} + async fn ensure_agent_org_runtime_sessions_idle( state: &AgentAppState, plan: &AgentOrgSessionDeletePlan, ) -> Result<(), String> { let runtime_sessions = agent_org_runtime_sessions(state, plan).await; - let blockers = agent_org_runtime_blockers(&runtime_sessions).await; + let mut blockers = agent_org_runtime_blockers(&runtime_sessions).await; + blockers.extend(agent_org_submission_blockers(plan)); if blockers.is_empty() { Ok(()) } else { @@ -617,6 +644,10 @@ fn delete_agent_org_session_hierarchy( expected_plan.run_id )); } + crate::coordination::agent_org_runs::remove_conversation_delete_fence_with_connection( + &tx, + &expected_plan.root_session_id, + )?; ensure_agent_org_hierarchy_absent(&tx, expected_plan)?; tx.commit().map_err(|err| err.to_string())?; Ok::<_, String>(outcome) @@ -1811,4 +1842,41 @@ mod tests { .await .expect("maintenance finishes after the timeout assertion"); } + + #[tokio::test] + async fn session_hierarchy_delete_waits_for_agent_org_submission_lease() { + let _sandbox = test_helpers::test_env::sandbox(); + let root = "hierarchy-submission-lease-root"; + let state = AgentAppState::new(); + let plan = AgentOrgSessionDeletePlan { + run_id: "hierarchy-submission-lease-run".to_string(), + root_session_id: root.to_string(), + run_status: crate::coordination::agent_org_runs::AgentOrgRunStatus::Cancelled, + sessions: vec![AgentOrgSessionDeleteNode { + session_id: root.to_string(), + parent_session_id: None, + status: SessionStatus::Pending, + depth: 0, + }], + }; + let lease = crate::coordination::agent_org_runs::AgentOrgSubmissionLease::begin(root); + + let error = stop_agent_org_runtime_sessions_with_timeout( + &state, + &plan, + std::time::Duration::from_millis(25), + ) + .await + .expect_err("deletion must wait for an initializing submission"); + assert!(error.contains("submission_in_progress=true"), "{error}"); + + drop(lease); + stop_agent_org_runtime_sessions_with_timeout( + &state, + &plan, + std::time::Duration::from_millis(25), + ) + .await + .expect("deletion continues after the submission exits"); + } } diff --git a/src-tauri/crates/agent-core/src/state/session_runtime.rs b/src-tauri/crates/agent-core/src/state/session_runtime.rs index a9dae6a07..31eb7421f 100644 --- a/src-tauri/crates/agent-core/src/state/session_runtime.rs +++ b/src-tauri/crates/agent-core/src/state/session_runtime.rs @@ -4,6 +4,9 @@ use std::sync::atomic::{AtomicBool, AtomicI64, Ordering}; use std::sync::Arc; use std::time::Instant; +use crate::coordination::agent_org_runs::{ + AgentOrgSubmissionPolicy, AgentOrgSubmissionScope, SharedAgentOrgSubmissionScope, +}; use crate::definitions::AgentDefinition; use crate::definitions::SessionMode; use crate::interaction::mode_switch::ModeSwitchManager; @@ -106,6 +109,9 @@ pub struct AgentSession { pub id: String, /// The resolved agent definition (inheritance applied). pub definition: AgentDefinition, + /// Exact Agent Org ownership for runtime/dispatch admission. Durable + /// deletion-fence state is never cached here. + submission_scope: SharedAgentOrgSubmissionScope, // ── Runtime Resources ───────────────────────────────────────────────── /// Runtime resources (provider, tools, policy). Set after initialization. @@ -253,6 +259,9 @@ impl AgentSession { .unwrap_or(false); let session_id_for_scheduler = id.clone(); + let submission_scope = Arc::new(AgentOrgSubmissionPolicy::new( + AgentOrgSubmissionScope::Ordinary, + )); // Shared cancel flag: session + any manager that needs cancel-aware // waits (see *Manager::with_cancel_flag). Must be built before the @@ -291,6 +300,7 @@ impl AgentSession { Self { id, definition, + submission_scope: Arc::clone(&submission_scope), runtime: tokio::sync::RwLock::new(None), compaction: tokio::sync::Mutex::new(CompactionState::default()), last_context_tokens: Arc::new(AtomicI64::new(0)), @@ -310,7 +320,11 @@ impl AgentSession { last_active_at: tokio::sync::Mutex::new(Instant::now()), active_turn: tokio::sync::Mutex::new(None), active_turn_generation: Arc::new(parking_lot::RwLock::new(None)), - scheduler: DialogScheduler::new(session_id_for_scheduler, 32), + scheduler: DialogScheduler::new_with_submission_scope( + session_id_for_scheduler, + 32, + submission_scope, + ), steering_queue: Arc::new(tokio::sync::Mutex::new(Vec::new())), em_state: Arc::new(tokio::sync::Mutex::new(ExtractMemoriesState::default())), ad_state: Arc::new(tokio::sync::Mutex::new(AutoDreamState::default())), @@ -330,10 +344,18 @@ impl AgentSession { } /// Attach (or replace) the runtime after initialization completes. - pub async fn set_runtime(&self, runtime: Arc) { + pub(crate) async fn set_runtime(&self, runtime: Arc) { *self.runtime.write().await = Some(runtime); } + pub(crate) fn set_agent_org_submission_scope(&self, scope: AgentOrgSubmissionScope) { + self.submission_scope.store(scope); + } + + pub(crate) fn agent_org_submission_scope(&self) -> SharedAgentOrgSubmissionScope { + Arc::clone(&self.submission_scope) + } + /// Return the current runtime, if initialized. pub async fn get_runtime(&self) -> Option> { self.runtime.read().await.clone() diff --git a/src-tauri/crates/e2e-test/src/agent_org.rs b/src-tauri/crates/e2e-test/src/agent_org.rs index f5886d227..a5d98a62f 100644 --- a/src-tauri/crates/e2e-test/src/agent_org.rs +++ b/src-tauri/crates/e2e-test/src/agent_org.rs @@ -37,6 +37,7 @@ const SEED_CLI_MEMBER_RUN_PATH: &str = "/agent/test/agent-org/stale-workers/seed const SEED_RUST_MEMBER_RUN_PATH: &str = "/agent/test/agent-org/stale-workers/seed-run"; const SESSION_DELETE_SNAPSHOT_PATH: &str = "/agent/test/agent-org/session-delete/snapshot"; const SESSION_DELETE_ATTEMPT_PATH: &str = "/agent/test/agent-org/session-delete/attempt"; +const SUBMISSION_CONTROL_PATH: &str = "/agent/test/agent-org/submission-control"; const TASKS_SEED_PATH: &str = "/agent/test/agent-org/tasks/seed"; const TASKS_LIST_PATH: &str = "/agent/test/agent-org/tasks/list"; const PAUSE_RUN_PATH: &str = "/agent/test/agent-org/run/pause"; @@ -3701,3 +3702,193 @@ pub async fn app_restart_transitions_running_runs_to_paused(cfg: &Config) -> boo } passed && cleanup_ok } + +pub async fn ordinary_sde_agent_org_isolation_production_command(cfg: &Config) -> bool { + let label = "ordinary-sde-agent-org-isolation-production-command"; + if let Err(error) = post_agent_org_json( + cfg, + SUBMISSION_CONTROL_PATH, + serde_json::json!({ "action": "reset_metrics" }), + ) + .await + { + return harness::print_error(label, &error); + } + + let suffix = unique_run_id("ordinary-isolation"); + let session_id = format!("ordinary-sde-{suffix}"); + let workspace = tmp_agent_org_workspace("ordinary-isolation"); + let fake_model = format!("e2e-fake-provider-{suffix}"); + let options = harness::SdeMessageOpts { + model_override: Some(&fake_model), + ..Default::default() + }; + let response = match harness::send_sde_message_with_opts( + cfg, + "Reply with an ordinary SDE isolation marker.", + &session_id, + "build", + &workspace, + &options, + ) + .await + { + Ok(response) => response, + Err(error) => return harness::print_error(label, &error), + }; + let compact = post_agent_org_json( + cfg, + SUBMISSION_CONTROL_PATH, + serde_json::json!({ "action": "manual_compact", "session_id": session_id }), + ) + .await + .unwrap_or_default(); + let channel = post_agent_org_json( + cfg, + SUBMISSION_CONTROL_PATH, + serde_json::json!({ + "action": "channel_message", + "session_id": session_id, + "model": fake_model, + "account_id": cfg.account_id, + }), + ) + .await + .unwrap_or_default(); + let gateway = post_agent_org_json( + cfg, + SUBMISSION_CONTROL_PATH, + serde_json::json!({ "action": "gateway_message", "session_id": session_id }), + ) + .await + .unwrap_or_default(); + let metrics = match post_agent_org_json( + cfg, + SUBMISSION_CONTROL_PATH, + serde_json::json!({ "action": "snapshot_metrics" }), + ) + .await + { + Ok(response) => response, + Err(error) => return harness::print_error(label, &error), + }; + + harness::print_result( + label, + &metrics.to_string(), + &[ + ( + "ordinary SDE completed its production lazy-init/message path", + response.content.contains("E2E_FAKE_PROVIDER_REPLY"), + ), + ( + "ordinary Manual Compact path remained available", + compact.get("ok").and_then(serde_json::Value::as_bool) == Some(true), + ), + ( + "ordinary Channel path remained available", + channel.get("ok").and_then(serde_json::Value::as_bool) == Some(true), + ), + ( + "ordinary Gateway path remained available", + gateway.get("ok").and_then(serde_json::Value::as_bool) == Some(true), + ), + ( + "ordinary SDE issued zero Agent Org admission queries", + metrics.get("queries").and_then(serde_json::Value::as_u64) == Some(0), + ), + ( + "ordinary SDE performed zero Agent Org lease mutations", + metrics + .get("lease_mutations") + .and_then(serde_json::Value::as_u64) + == Some(0), + ), + ], + ) +} + +pub async fn agent_org_runtime_submission_fence_production_command(cfg: &Config) -> bool { + let label = "agent-org-runtime-submission-fence-production-command"; + let fixture = unique_run_id("submission-fence"); + let root_session_id = format!("root-{fixture}"); + let worker_session_id = format!("worker-{fixture}"); + let seed = match post_agent_org_json( + cfg, + SEED_RUST_MEMBER_RUN_PATH, + serde_json::json!({ + "org_id": fixture, + "root_session_id": root_session_id, + "run_status": "running", + "root_status": "idle", + "workers": [{ + "member_id": "m-submission-fence", + "agent_definition_id": "builtin:explore", + "session_id": worker_session_id, + "status": "pending" + }] + }), + ) + .await + { + Ok(response) if response.get("ok").and_then(serde_json::Value::as_bool) == Some(true) => { + response + } + Ok(response) => return harness::print_error(label, &response.to_string()), + Err(error) => return harness::print_error(label, &error), + }; + let run_id = seed + .get("org_run_id") + .and_then(serde_json::Value::as_str) + .unwrap_or_default() + .to_string(); + let fenced = post_agent_org_json( + cfg, + SUBMISSION_CONTROL_PATH, + serde_json::json!({ + "action": "fence_fixture", + "root_session_id": root_session_id, + }), + ) + .await + .unwrap_or_default(); + let wake = post_agent_org_json( + cfg, + SESSION_RETURN_TO_WORK_PATH, + serde_json::json!({ "session_id": worker_session_id }), + ) + .await + .unwrap_or_default(); + let state = post_agent_org_json( + cfg, + DURABLE_INVARIANTS_PATH, + serde_json::json!({ "org_run_id": run_id, "root_session_id": root_session_id }), + ) + .await + .unwrap_or_default(); + let cleanup = post_agent_org_json( + cfg, + SESSION_DELETE_ATTEMPT_PATH, + serde_json::json!({ "session_id": root_session_id }), + ) + .await + .unwrap_or_default(); + + harness::print_result( + label, + &serde_json::json!({ "seed": seed, "fence": fenced, "wake": wake, "state": state, "cleanup": cleanup }).to_string(), + &[ + ("fixture fence committed", fenced.get("ok").and_then(serde_json::Value::as_bool) == Some(true)), + ( + "production Wake rejected the fenced Agent Org session", + wake.get("ok").and_then(serde_json::Value::as_bool) == Some(false) + && wake + .get("error") + .and_then(serde_json::Value::as_str) + .is_some_and(|error| error.starts_with("agent_org_run_not_writable:")), + ), + ("fencing cancelled the live Run", state.get("runStatus").and_then(serde_json::Value::as_str) == Some("cancelled")), + ("production deletion removed the disposable fixture and fence", cleanup.get("ok").and_then(serde_json::Value::as_bool) == Some(true)), + ], + ) +} diff --git a/src-tauri/crates/e2e-test/src/main.rs b/src-tauri/crates/e2e-test/src/main.rs index 71bdf2e0a..98f491096 100644 --- a/src-tauri/crates/e2e-test/src/main.rs +++ b/src-tauri/crates/e2e-test/src/main.rs @@ -668,6 +668,16 @@ fn all_scenarios() -> Vec { "agent-org-launch-materializes-member-sessions", agent_org::launch_materializes_member_sessions_in_run_view ), + scenario!( + "agent-org", + "agent-org-runtime-submission-fence-production-command", + agent_org::agent_org_runtime_submission_fence_production_command + ), + scenario!( + "agent-org", + "ordinary-sde-agent-org-isolation-production-command", + agent_org::ordinary_sde_agent_org_isolation_production_command + ), scenario!( "agent-org", "agent-org-production-return-to-work-drains-visible-input", diff --git a/src-tauri/src/api/agent/mod.rs b/src-tauri/src/api/agent/mod.rs index 33c70d839..e651bdad8 100644 --- a/src-tauri/src/api/agent/mod.rs +++ b/src-tauri/src/api/agent/mod.rs @@ -34,12 +34,135 @@ pub use public::AgentStatusResponse; #[cfg(debug_assertions)] use axum::routing::post; +#[cfg(debug_assertions)] +use axum::Json; use axum::{routing::get, Router}; // ============================================ // Router // ============================================ +#[cfg(debug_assertions)] +async fn test_agent_org_submission_control( + Json(body): Json, +) -> Json { + use agent_core::coordination::agent_org_runs::{ + debug_establish_e2e_conversation_delete_fence, reset_submission_metrics, submission_metrics, + }; + + match body.get("action").and_then(serde_json::Value::as_str) { + Some("reset_metrics") => { + reset_submission_metrics(); + Json(serde_json::json!({ "ok": true })) + } + Some("snapshot_metrics") => { + let (queries, lease_mutations) = submission_metrics(); + Json(serde_json::json!({ + "ok": true, + "queries": queries, + "lease_mutations": lease_mutations, + })) + } + Some("manual_compact" | "channel_message" | "gateway_message") => { + use tauri::Manager; + let Some(session_id) = body.get("session_id").and_then(serde_json::Value::as_str) + else { + return Json(serde_json::json!({ "ok": false, "error": "session_id is required" })); + }; + if !session_id.starts_with("ordinary-sde-e2e-agent-org-fixture:ordinary-isolation-") { + return Json( + serde_json::json!({ "ok": false, "error": "fixture session required" }), + ); + } + let Some(handle) = crate::api::get_app_handle() else { + return Json( + serde_json::json!({ "ok": false, "error": "AppHandle not initialized" }), + ); + }; + let state = handle.state::(); + if body.get("action").and_then(serde_json::Value::as_str) == Some("manual_compact") { + return match agent_core::state::commands::session::agent_session_manual_compact( + state, + session_id.to_string(), + None, + ) + .await + { + Ok(result) => Json(serde_json::json!({ "ok": true, "result": result })), + Err(error) => Json(serde_json::json!({ "ok": false, "error": error })), + }; + } + if body.get("action").and_then(serde_json::Value::as_str) == Some("gateway_message") { + let Some(session) = state.get_session(session_id).await else { + return Json( + serde_json::json!({ "ok": false, "error": "fixture session not initialized" }), + ); + }; + let mut message = agent_core::bus::InboundMessage::new( + "e2e:ordinary-isolation", + "fixture-sender", + "fixture-chat", + "ordinary gateway isolation marker", + ); + message.session_key_override = Some(session_id.to_string()); + return match agent_core::session::gateway_pipeline::process_gateway_message( + message, + session, + None, + Some(handle.clone()), + ) + .await + { + Ok(result) => Json(serde_json::json!({ "ok": true, "result": result })), + Err(error) => Json(serde_json::json!({ "ok": false, "error": error })), + }; + } + match agent_core::state::commands::session::channel::channel_process_message( + state, + "ordinary channel isolation marker".to_string(), + Some(session_id.to_string()), + body.get("model") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + body.get("account_id") + .and_then(serde_json::Value::as_str) + .map(str::to_string), + None, + None, + None, + None, + ) + .await + { + Ok(result) => Json(serde_json::json!({ "ok": true, "result": result })), + Err(error) => Json(serde_json::json!({ "ok": false, "error": error })), + } + } + Some("fence_fixture") => { + let Some(root_session_id) = body + .get("root_session_id") + .and_then(serde_json::Value::as_str) + .map(str::to_string) + else { + return Json(serde_json::json!({ + "ok": false, + "error": "root_session_id is required", + })); + }; + match tokio::task::spawn_blocking(move || { + debug_establish_e2e_conversation_delete_fence(&root_session_id) + }) + .await + { + Ok(Ok(())) => Json(serde_json::json!({ "ok": true })), + Ok(Err(error)) => Json(serde_json::json!({ "ok": false, "error": error })), + Err(error) => Json(serde_json::json!({ "ok": false, "error": error.to_string() })), + } + } + _ => Json(serde_json::json!({ "ok": false, "error": "unknown action" })), + } +} + /// Create the agent API routes. pub fn create_routes() -> Router { let router = Router::new() @@ -730,6 +853,10 @@ pub fn create_routes() -> Router { "/test/agent-org/session-delete/attempt", post(test::agent_org::test_agent_org_session_delete_attempt), ) + .route( + "/test/agent-org/submission-control", + post(test_agent_org_submission_control), + ) .route( "/test/agent-org/stale-workers/seed-cli-member", post(test::agent_org::test_agent_org_seed_cli_member_run),