From 0802e0340dbdc1a57fc2e2daebb2a07877389cd7 Mon Sep 17 00:00:00 2001 From: Shibo Sheng Date: Thu, 20 Aug 2026 01:09:52 +0800 Subject: [PATCH 1/2] feat(agent-org): add transactional turn context and member FIFO Persist a typed context for every Agent Org turn and allocate Member dispatches from a single per-run/member FIFO in the same immediate transaction as the base Turn Intent. Wire Starting and Coordinator Root admission, keep ordinary SDE turns context-free, and fail closed before persisting legacy Member group/inbox work that lacks typed authority. Use one strict current-schema manifest and isolated legacy cleanup without adding a second runtime or scheduler. Refs: #758 Pre-commit hook ran. Total eslint: 5, total circular: 0 --- .../agent_org_runs/store/starting.rs | 14 +- .../core/coordination/agent_org_runs/tests.rs | 19 + .../coordination/agent_org_turn_contexts.rs | 1035 +++++++++++++++++ .../agent_org_turn_contexts/tests.rs | 583 ++++++++++ .../agent-core/src/core/coordination/mod.rs | 9 + .../src/core/coordination/schema.rs | 84 +- .../state/commands/session/message/send.rs | 64 +- .../commands/session/org_tasks/group_chat.rs | 29 +- .../state/commands/session/org_tasks/tests.rs | 24 + .../session-persistence/src/turn_intents.rs | 145 +-- src-tauri/src/api/agent/test/agent_org.rs | 6 +- src-tauri/src/setup/hooks.rs | 2 +- 12 files changed, 1873 insertions(+), 141 deletions(-) create mode 100644 src-tauri/crates/agent-core/src/core/coordination/agent_org_turn_contexts.rs create mode 100644 src-tauri/crates/agent-core/src/core/coordination/agent_org_turn_contexts/tests.rs diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/store/starting.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/store/starting.rs index 87b16164fe..bf3fd6f39f 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/store/starting.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/store/starting.rs @@ -396,14 +396,16 @@ impl AgentOrgRunStore { "initial input is not durably materialized for Starting run {run_id}" )); } - crate::foundation::session_bridge::upsert_turn_intent_with_connection( - &transaction, + let admission = crate::coordination::agent_org_turn_contexts::AgentOrgTurnAdmission::starting_coordinator( + run_id, &root_session_id, &input.turn_intent_id, - Some(&input.message_id), - Some(run_id), - crate::foundation::session_bridge::TurnIntentBridgeSource::AgentOrg, - crate::foundation::session_bridge::TurnIntentBridgeStatus::Queued, + Some(input.message_id.clone()), + expected_generation, + ); + crate::coordination::agent_org_turn_contexts::accept_with_connection( + &transaction, + &admission, )?; transaction .execute( diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/tests.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/tests.rs index a40a3e4185..86bca916bd 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/tests.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_runs/tests.rs @@ -361,6 +361,16 @@ fn starting_finish_requires_exact_member_and_input_durability_then_is_idempotent ) .expect("load durable initial Turn Intent"); assert_eq!(turn_status, "queued"); + let context: (i64, String, String, Option) = conn + .query_row( + "SELECT COUNT(*), turn_kind, source_kind, member_dispatch_sequence + FROM agent_org_runtime_turn_contexts + WHERE session_id='starting-root' AND turn_intent_id='starting-turn'", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .expect("load initial Coordinator context"); + assert_eq!(context, (1, "coordinator".into(), "root_turn".into(), None)); } #[test] @@ -386,6 +396,15 @@ fn starting_without_initial_work_finishes_idle() { .expect("run exists") .idled_at .is_some()); + let conn = database::db::get_connection().expect("db"); + let context_count: i64 = conn + .query_row( + "SELECT COUNT(*) FROM agent_org_runtime_turn_contexts WHERE org_run_id=?1", + [&run.id], + |row| row.get(0), + ) + .expect("count no-work contexts"); + assert_eq!(context_count, 0); } #[test] diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_turn_contexts.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_turn_contexts.rs new file mode 100644 index 0000000000..7bf443fe06 --- /dev/null +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_turn_contexts.rs @@ -0,0 +1,1035 @@ +//! Canonical Agent Org companion context and per-Member dispatch ordering. +//! +//! `session_turn_intents` remains the generic lifecycle owner. This module +//! atomically attaches the Agent Org-only execution identity and, for Member +//! turns, allocates the one FIFO sequence shared by every typed source. + +use rusqlite::{params, Connection, OptionalExtension}; + +use crate::definitions::orgs::{validate_launch_snapshot, AgentOrgLaunchSnapshot}; +use crate::foundation::session_bridge::{TurnIntentBridgeSource, TurnIntentBridgeStatus}; + +use super::agent_org_runs::{AgentOrgRunStatus, COORDINATOR_MEMBER_ID}; + +pub(crate) const TURN_CONTEXT_INVARIANT_PREFIX: &str = "agent_org_turn_context_invalid:"; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum AgentOrgTurnKind { + Coordinator, + TaskExecution, + UserDirectedWork, +} + +impl AgentOrgTurnKind { + const fn as_str(self) -> &'static str { + match self { + Self::Coordinator => "coordinator", + Self::TaskExecution => "task_execution", + Self::UserDirectedWork => "user_directed_work", + } + } + + fn parse(value: &str) -> Option { + Some(match value { + "coordinator" => Self::Coordinator, + "task_execution" => Self::TaskExecution, + "user_directed_work" => Self::UserDirectedWork, + _ => return None, + }) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum AgentOrgTurnSourceKind { + RootTurn, + Task, + DirectMember, + GroupMention, + MemberInbox, +} + +impl AgentOrgTurnSourceKind { + const fn as_str(self) -> &'static str { + match self { + Self::RootTurn => "root_turn", + Self::Task => "task", + Self::DirectMember => "direct_member", + Self::GroupMention => "group_mention", + Self::MemberInbox => "member_inbox", + } + } + + fn parse(value: &str) -> Option { + Some(match value { + "root_turn" => Self::RootTurn, + "task" => Self::Task, + "direct_member" => Self::DirectMember, + "group_mention" => Self::GroupMention, + "member_inbox" => Self::MemberInbox, + _ => return None, + }) + } +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct AgentOrgTurnContext { + pub context_id: i64, + pub session_id: String, + pub turn_intent_id: String, + pub org_run_id: String, + pub participant_id: String, + pub turn_kind: AgentOrgTurnKind, + pub task_id: Option, + pub owner_member_id: Option, + pub dispatch_member_id: Option, + pub member_dispatch_sequence: Option, + pub source_kind: AgentOrgTurnSourceKind, + pub source_id: String, + pub root_authority_turn_id: Option, + pub actor_version: Option, + pub activation_generation: Option, + pub created_at: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum AdmissionKind { + Coordinator { + expected_generation: Option, + }, + TaskExecution { + task_id: String, + owner_member_id: String, + activation_generation: i64, + }, + UserDirectedWork { + dispatch_member_id: String, + source: UserDirectedSource, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum UserDirectedSource { + DirectMember { source_event_id: String }, + GroupMention { source_inbox_id: i64 }, + MemberInbox { source_inbox_id: i64 }, +} + +/// Closed construction surface for all Agent Org Turn kinds. Product entry +/// points receive only the constructors they can prove from canonical data; +/// no caller can submit a free-form `turn_kind` string. +#[derive(Debug, Clone)] +pub(crate) struct AgentOrgTurnAdmission { + org_run_id: String, + session_id: String, + turn_intent_id: String, + client_message_id: Option, + base_source: TurnIntentBridgeSource, + kind: AdmissionKind, +} + +impl AgentOrgTurnAdmission { + pub(crate) fn coordinator( + org_run_id: impl Into, + session_id: impl Into, + turn_intent_id: impl Into, + client_message_id: Option, + base_source: TurnIntentBridgeSource, + ) -> Self { + Self { + org_run_id: org_run_id.into(), + session_id: session_id.into(), + turn_intent_id: turn_intent_id.into(), + client_message_id, + base_source, + kind: AdmissionKind::Coordinator { + expected_generation: None, + }, + } + } + + pub(crate) fn starting_coordinator( + org_run_id: impl Into, + session_id: impl Into, + turn_intent_id: impl Into, + client_message_id: Option, + expected_generation: i64, + ) -> Self { + let mut request = Self::coordinator( + org_run_id, + session_id, + turn_intent_id, + client_message_id, + TurnIntentBridgeSource::AgentOrg, + ); + request.kind = AdmissionKind::Coordinator { + expected_generation: Some(expected_generation), + }; + request + } + + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn task_execution( + org_run_id: impl Into, + session_id: impl Into, + turn_intent_id: impl Into, + client_message_id: Option, + task_id: impl Into, + owner_member_id: impl Into, + activation_generation: i64, + ) -> Self { + Self { + org_run_id: org_run_id.into(), + session_id: session_id.into(), + turn_intent_id: turn_intent_id.into(), + client_message_id, + base_source: TurnIntentBridgeSource::AgentOrg, + kind: AdmissionKind::TaskExecution { + task_id: task_id.into(), + owner_member_id: owner_member_id.into(), + activation_generation, + }, + } + } + + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn direct_member( + org_run_id: impl Into, + session_id: impl Into, + turn_intent_id: impl Into, + client_message_id: Option, + dispatch_member_id: impl Into, + source_event_id: impl Into, + ) -> Self { + Self { + org_run_id: org_run_id.into(), + session_id: session_id.into(), + turn_intent_id: turn_intent_id.into(), + client_message_id, + base_source: TurnIntentBridgeSource::AgentOrg, + kind: AdmissionKind::UserDirectedWork { + dispatch_member_id: dispatch_member_id.into(), + source: UserDirectedSource::DirectMember { + source_event_id: source_event_id.into(), + }, + }, + } + } + + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn group_mention( + org_run_id: impl Into, + session_id: impl Into, + turn_intent_id: impl Into, + client_message_id: Option, + dispatch_member_id: impl Into, + source_inbox_id: i64, + ) -> Self { + Self { + org_run_id: org_run_id.into(), + session_id: session_id.into(), + turn_intent_id: turn_intent_id.into(), + client_message_id, + base_source: TurnIntentBridgeSource::AgentOrg, + kind: AdmissionKind::UserDirectedWork { + dispatch_member_id: dispatch_member_id.into(), + source: UserDirectedSource::GroupMention { source_inbox_id }, + }, + } + } + + #[cfg_attr(not(test), allow(dead_code))] + pub(crate) fn member_inbox( + org_run_id: impl Into, + session_id: impl Into, + turn_intent_id: impl Into, + client_message_id: Option, + dispatch_member_id: impl Into, + source_inbox_id: i64, + ) -> Self { + Self { + org_run_id: org_run_id.into(), + session_id: session_id.into(), + turn_intent_id: turn_intent_id.into(), + client_message_id, + base_source: TurnIntentBridgeSource::AgentOrg, + kind: AdmissionKind::UserDirectedWork { + dispatch_member_id: dispatch_member_id.into(), + source: UserDirectedSource::MemberInbox { source_inbox_id }, + }, + } + } +} + +pub(super) fn create_schema(conn: &Connection) -> rusqlite::Result<()> { + conn.execute_batch( + "CREATE TABLE IF NOT EXISTS agent_org_runtime_member_dispatch_allocators ( + org_run_id TEXT NOT NULL, + member_id TEXT NOT NULL CHECK(length(trim(member_id)) > 0), + next_sequence INTEGER NOT NULL CHECK(next_sequence >= 1), + PRIMARY KEY(org_run_id, member_id), + FOREIGN KEY(org_run_id) REFERENCES agent_org_runtime_runs(id) ON DELETE CASCADE + ); + + CREATE TABLE IF NOT EXISTS agent_org_runtime_turn_contexts ( + context_id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL CHECK(length(trim(session_id)) > 0), + turn_intent_id TEXT NOT NULL CHECK(length(trim(turn_intent_id)) > 0), + org_run_id TEXT NOT NULL, + participant_id TEXT NOT NULL CHECK(length(trim(participant_id)) > 0), + turn_kind TEXT NOT NULL, + task_id TEXT, + owner_member_id TEXT, + dispatch_member_id TEXT, + member_dispatch_sequence INTEGER, + source_kind TEXT NOT NULL, + source_id TEXT NOT NULL CHECK(length(trim(source_id)) > 0), + root_authority_turn_id TEXT, + actor_version INTEGER, + activation_generation INTEGER, + created_at TEXT NOT NULL, + UNIQUE(session_id, turn_intent_id), + FOREIGN KEY(session_id, turn_intent_id) + REFERENCES session_turn_intents(session_id, turn_intent_id) + ON DELETE CASCADE, + FOREIGN KEY(org_run_id) + REFERENCES agent_org_runtime_runs(id) ON DELETE CASCADE, + CHECK( + (turn_kind='coordinator' + AND participant_id='coordinator' + AND task_id IS NULL AND owner_member_id IS NULL + AND dispatch_member_id IS NULL AND member_dispatch_sequence IS NULL + AND source_kind='root_turn' AND source_id=turn_intent_id + AND root_authority_turn_id IS NULL AND actor_version IS NULL + AND activation_generation IS NOT NULL AND activation_generation >= 1) + OR + (turn_kind='task_execution' + AND task_id IS NOT NULL AND length(trim(task_id)) > 0 + AND owner_member_id=participant_id + AND dispatch_member_id=participant_id + AND member_dispatch_sequence IS NOT NULL AND member_dispatch_sequence >= 1 + AND source_kind='task' AND source_id=task_id + AND root_authority_turn_id IS NULL AND actor_version IS NULL + AND activation_generation IS NOT NULL AND activation_generation >= 1) + OR + (turn_kind='user_directed_work' + AND task_id IS NULL AND owner_member_id IS NULL + AND dispatch_member_id=participant_id + AND member_dispatch_sequence IS NOT NULL AND member_dispatch_sequence >= 1 + AND actor_version IS NOT NULL AND actor_version >= 1 + AND activation_generation IS NULL + AND ( + (source_kind='direct_member' + AND root_authority_turn_id=turn_intent_id) + OR + (source_kind='group_mention' + AND root_authority_turn_id=turn_intent_id) + OR + (source_kind='member_inbox' + AND root_authority_turn_id IS NULL) + )) + ) + ); + CREATE UNIQUE INDEX IF NOT EXISTS idx_agent_org_runtime_turn_contexts_member_sequence + ON agent_org_runtime_turn_contexts( + org_run_id, dispatch_member_id, member_dispatch_sequence + ) + WHERE dispatch_member_id IS NOT NULL; + CREATE INDEX IF NOT EXISTS idx_agent_org_runtime_turn_contexts_source + ON agent_org_runtime_turn_contexts( + org_run_id, source_kind, source_id, context_id + );", + ) +} + +/// Open the canonical writer transaction and accept exactly one typed turn. +pub(crate) fn accept(request: &AgentOrgTurnAdmission) -> Result { + database::db::with_sessions_writer(|| { + let mut connection = database::db::get_connection().map_err(|error| error.to_string())?; + let transaction = connection + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|error| error.to_string())?; + let context = accept_with_connection(&transaction, request)?; + transaction.commit().map_err(|error| error.to_string())?; + Ok(context) + }) +} + +/// Connection-scoped admission for lifecycle owners that already hold an +/// IMMEDIATE transaction (notably Starting completion). +pub(crate) fn accept_with_connection( + conn: &Connection, + request: &AgentOrgTurnAdmission, +) -> Result { + validate_non_empty(request)?; + + let base = read_base(conn, &request.session_id, &request.turn_intent_id)?; + let existing = read_context_optional(conn, &request.session_id, &request.turn_intent_id)?; + match (base, existing) { + (Some(base), Some(context)) => { + ensure_base_matches(request, &base)?; + ensure_context_matches(request, &context)?; + return Ok(context); + } + (Some(_), None) => { + return Err(invariant_error(format!( + "base Turn exists without companion context for {}/{}", + request.session_id, request.turn_intent_id + ))) + } + (None, Some(_)) => { + return Err(invariant_error(format!( + "companion context exists without base Turn for {}/{}", + request.session_id, request.turn_intent_id + ))) + } + (None, None) => {} + } + + let canonical = resolve_canonical_admission(conn, request)?; + let sequence = match canonical.dispatch_member_id.as_deref() { + Some(member_id) => Some(allocate_member_sequence( + conn, + &request.org_run_id, + member_id, + )?), + None => None, + }; + + crate::foundation::session_bridge::upsert_turn_intent_with_connection( + conn, + &request.session_id, + &request.turn_intent_id, + request.client_message_id.as_deref(), + Some(&request.org_run_id), + request.base_source, + TurnIntentBridgeStatus::Queued, + )?; + + let now = chrono::Utc::now().to_rfc3339(); + conn.execute( + "INSERT INTO agent_org_runtime_turn_contexts ( + session_id, turn_intent_id, org_run_id, participant_id, turn_kind, + task_id, owner_member_id, dispatch_member_id, member_dispatch_sequence, + source_kind, source_id, root_authority_turn_id, actor_version, + activation_generation, created_at + ) VALUES ( + ?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15 + )", + params![ + &request.session_id, + &request.turn_intent_id, + &request.org_run_id, + &canonical.participant_id, + canonical.turn_kind.as_str(), + canonical.task_id.as_deref(), + canonical.owner_member_id.as_deref(), + canonical.dispatch_member_id.as_deref(), + sequence, + canonical.source_kind.as_str(), + &canonical.source_id, + canonical.root_authority_turn_id.as_deref(), + canonical.actor_version, + canonical.activation_generation, + &now, + ], + ) + .map_err(|error| invariant_error(format!("failed to insert companion context: {error}")))?; + + require_context_with_connection(conn, &request.session_id, &request.turn_intent_id) +} + +#[derive(Debug)] +struct CanonicalAdmission { + participant_id: String, + turn_kind: AgentOrgTurnKind, + task_id: Option, + owner_member_id: Option, + dispatch_member_id: Option, + source_kind: AgentOrgTurnSourceKind, + source_id: String, + root_authority_turn_id: Option, + actor_version: Option, + activation_generation: Option, +} + +fn resolve_canonical_admission( + conn: &Connection, + request: &AgentOrgTurnAdmission, +) -> Result { + let run: Option<(Option, Option, i64, String)> = conn + .query_row( + "SELECT root_session_id, org_snapshot_json, activation_generation, status + FROM agent_org_runtime_runs WHERE id=?1", + [&request.org_run_id], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .optional() + .map_err(|error| error.to_string())?; + let Some((root_session_id, snapshot_json, generation, status_raw)) = run else { + return Err(invariant_error(format!( + "run {} does not exist", + request.org_run_id + ))); + }; + let status = AgentOrgRunStatus::parse(&status_raw) + .ok_or_else(|| invariant_error(format!("unknown run status {status_raw:?}")))?; + let snapshot_json = snapshot_json.ok_or_else(|| { + invariant_error(format!( + "run {} has no immutable launch snapshot", + request.org_run_id + )) + })?; + let snapshot: AgentOrgLaunchSnapshot = serde_json::from_str(&snapshot_json) + .map_err(|error| invariant_error(format!("invalid launch snapshot JSON: {error}")))?; + validate_launch_snapshot(&snapshot) + .map_err(|error| invariant_error(format!("invalid launch snapshot: {error}")))?; + + match &request.kind { + AdmissionKind::Coordinator { + expected_generation, + } => { + let root_session_id = root_session_id.ok_or_else(|| { + invariant_error(format!("run {} has no canonical Root", request.org_run_id)) + })?; + if root_session_id != request.session_id { + return Err(invariant_error(format!( + "session {} is not canonical Root {}", + request.session_id, root_session_id + ))); + } + if let Some(expected) = expected_generation { + if generation != *expected || status != AgentOrgRunStatus::Starting { + return Err(invariant_error(format!( + "Starting authority mismatch: expected generation {expected}, current generation {generation}, status {status}" + ))); + } + } else if status != AgentOrgRunStatus::Running { + return Err(invariant_error(format!( + "Coordinator Turn requires a running Team, found {status}" + ))); + } + resolve_materialization_version( + conn, + request, + COORDINATOR_MEMBER_ID, + &snapshot.coordinator_agent_id, + )?; + Ok(CanonicalAdmission { + participant_id: COORDINATOR_MEMBER_ID.to_string(), + turn_kind: AgentOrgTurnKind::Coordinator, + task_id: None, + owner_member_id: None, + dispatch_member_id: None, + source_kind: AgentOrgTurnSourceKind::RootTurn, + source_id: request.turn_intent_id.clone(), + root_authority_turn_id: None, + actor_version: None, + activation_generation: Some(generation), + }) + } + AdmissionKind::TaskExecution { + task_id, + owner_member_id, + activation_generation, + } => { + if status != AgentOrgRunStatus::Running || generation != *activation_generation { + return Err(invariant_error(format!( + "TaskExecution authority mismatch for generation {activation_generation}; current generation {generation}, status {status}" + ))); + } + let agent_id = snapshot_member_agent_id(&snapshot, owner_member_id)?; + resolve_materialization_version(conn, request, owner_member_id, agent_id)?; + let task_owner: Option> = conn + .query_row( + "SELECT owner FROM agent_org_runtime_tasks + WHERE org_run_id=?1 AND id=?2", + params![&request.org_run_id, task_id], + |row| row.get(0), + ) + .optional() + .map_err(|error| error.to_string())?; + if task_owner.flatten().as_deref() != Some(owner_member_id) { + return Err(invariant_error(format!( + "Task {task_id} is not owned by canonical Member {owner_member_id}" + ))); + } + Ok(CanonicalAdmission { + participant_id: owner_member_id.clone(), + turn_kind: AgentOrgTurnKind::TaskExecution, + task_id: Some(task_id.clone()), + owner_member_id: Some(owner_member_id.clone()), + dispatch_member_id: Some(owner_member_id.clone()), + source_kind: AgentOrgTurnSourceKind::Task, + source_id: task_id.clone(), + root_authority_turn_id: None, + actor_version: None, + activation_generation: Some(generation), + }) + } + AdmissionKind::UserDirectedWork { + dispatch_member_id, + source, + } => { + if matches!( + status, + AgentOrgRunStatus::Starting + | AgentOrgRunStatus::Failed + | AgentOrgRunStatus::Archived + ) { + return Err(invariant_error(format!( + "UserDirectedWork cannot enter Team status {status}" + ))); + } + let agent_id = snapshot_member_agent_id(&snapshot, dispatch_member_id)?; + let actor_version = + resolve_materialization_version(conn, request, dispatch_member_id, agent_id)?; + let (source_kind, source_id, root_authority_turn_id) = match source { + UserDirectedSource::DirectMember { source_event_id } => { + let source_exists: bool = conn + .query_row( + "SELECT EXISTS( + SELECT 1 FROM events WHERE id=?1 AND session_id=?2 + )", + params![source_event_id, &request.session_id], + |row| row.get(0), + ) + .map_err(|error| error.to_string())?; + if !source_exists { + return Err(invariant_error(format!( + "DirectMember source event {source_event_id} is not canonical" + ))); + } + ( + AgentOrgTurnSourceKind::DirectMember, + source_event_id.clone(), + Some(request.turn_intent_id.clone()), + ) + } + UserDirectedSource::GroupMention { source_inbox_id } => { + validate_source_inbox( + conn, + &request.org_run_id, + dispatch_member_id, + *source_inbox_id, + )?; + ( + AgentOrgTurnSourceKind::GroupMention, + source_inbox_id.to_string(), + Some(request.turn_intent_id.clone()), + ) + } + UserDirectedSource::MemberInbox { source_inbox_id } => { + validate_source_inbox( + conn, + &request.org_run_id, + dispatch_member_id, + *source_inbox_id, + )?; + ( + AgentOrgTurnSourceKind::MemberInbox, + source_inbox_id.to_string(), + None, + ) + } + }; + Ok(CanonicalAdmission { + participant_id: dispatch_member_id.clone(), + turn_kind: AgentOrgTurnKind::UserDirectedWork, + task_id: None, + owner_member_id: None, + dispatch_member_id: Some(dispatch_member_id.clone()), + source_kind, + source_id, + root_authority_turn_id, + actor_version: Some(actor_version), + activation_generation: None, + }) + } + } +} + +fn resolve_materialization_version( + conn: &Connection, + request: &AgentOrgTurnAdmission, + member_id: &str, + agent_id: &str, +) -> Result { + let version: Option = conn + .query_row( + "SELECT materialization.generation + FROM agent_org_runtime_member_materializations materialization + JOIN agent_sessions session + ON session.session_id=materialization.session_id + WHERE materialization.org_run_id=?1 + AND materialization.member_id=?2 + AND materialization.agent_id=?3 + AND materialization.session_id=?4 + AND materialization.status='succeeded' + AND session.agent_definition_id=?3 + AND session.org_member_id=?2 + AND materialization.generation=( + SELECT MAX(latest.generation) + FROM agent_org_runtime_member_materializations latest + WHERE latest.org_run_id=?1 + AND latest.member_id=?2 + AND latest.status='succeeded' + ) + LIMIT 1", + params![ + &request.org_run_id, + member_id, + agent_id, + &request.session_id, + ], + |row| row.get(0), + ) + .optional() + .map_err(|error| error.to_string())?; + version.ok_or_else(|| { + invariant_error(format!( + "session {} is not the latest canonical materialization for {}/{}", + request.session_id, request.org_run_id, member_id + )) + }) +} + +fn snapshot_member_agent_id<'a>( + snapshot: &'a AgentOrgLaunchSnapshot, + member_id: &str, +) -> Result<&'a str, String> { + snapshot + .members + .iter() + .find(|member| member.member_id == member_id) + .map(|member| member.agent_id.as_str()) + .ok_or_else(|| invariant_error(format!("unknown canonical Member {member_id}"))) +} + +fn validate_source_inbox( + conn: &Connection, + org_run_id: &str, + dispatch_member_id: &str, + source_inbox_id: i64, +) -> Result<(), String> { + let valid: bool = conn + .query_row( + "SELECT EXISTS( + SELECT 1 FROM agent_org_runtime_inbox + WHERE id=?1 AND org_run_id=?2 AND recipient_member_id=?3 + )", + params![source_inbox_id, org_run_id, dispatch_member_id], + |row| row.get(0), + ) + .map_err(|error| error.to_string())?; + if !valid { + return Err(invariant_error(format!( + "Inbox source {source_inbox_id} is not canonical for {org_run_id}/{dispatch_member_id}" + ))); + } + Ok(()) +} + +fn allocate_member_sequence( + conn: &Connection, + org_run_id: &str, + member_id: &str, +) -> Result { + conn.query_row( + "INSERT INTO agent_org_runtime_member_dispatch_allocators ( + org_run_id, member_id, next_sequence + ) VALUES (?1, ?2, 2) + ON CONFLICT(org_run_id, member_id) DO UPDATE + SET next_sequence=next_sequence + 1 + RETURNING next_sequence - 1", + params![org_run_id, member_id], + |row| row.get(0), + ) + .map_err(|error| invariant_error(format!("failed to allocate Member sequence: {error}"))) +} + +#[derive(Debug)] +struct BaseTurn { + client_message_id: Option, + org_run_id: Option, + source: String, +} + +fn read_base( + conn: &Connection, + session_id: &str, + turn_intent_id: &str, +) -> Result, String> { + conn.query_row( + "SELECT client_message_id, org_run_id, source + FROM session_turn_intents + WHERE session_id=?1 AND turn_intent_id=?2", + params![session_id, turn_intent_id], + |row| { + Ok(BaseTurn { + client_message_id: row.get(0)?, + org_run_id: row.get(1)?, + source: row.get(2)?, + }) + }, + ) + .optional() + .map_err(|error| error.to_string()) +} + +fn ensure_base_matches(request: &AgentOrgTurnAdmission, base: &BaseTurn) -> Result<(), String> { + if base.client_message_id != request.client_message_id + || base.org_run_id.as_deref() != Some(request.org_run_id.as_str()) + || base.source != request.base_source.as_str() + { + return Err(invariant_error(format!( + "base Turn replay mismatch for {}/{}", + request.session_id, request.turn_intent_id + ))); + } + Ok(()) +} + +fn ensure_context_matches( + request: &AgentOrgTurnAdmission, + context: &AgentOrgTurnContext, +) -> Result<(), String> { + let common_matches = context.session_id == request.session_id + && context.turn_intent_id == request.turn_intent_id + && context.org_run_id == request.org_run_id; + let kind_matches = match &request.kind { + AdmissionKind::Coordinator { + expected_generation, + } => { + context.turn_kind == AgentOrgTurnKind::Coordinator + && context.participant_id == COORDINATOR_MEMBER_ID + && context.source_kind == AgentOrgTurnSourceKind::RootTurn + && context.source_id == request.turn_intent_id + && expected_generation + .map(|generation| context.activation_generation == Some(generation)) + .unwrap_or(true) + } + AdmissionKind::TaskExecution { + task_id, + owner_member_id, + activation_generation, + } => { + context.turn_kind == AgentOrgTurnKind::TaskExecution + && context.participant_id == *owner_member_id + && context.task_id.as_deref() == Some(task_id) + && context.owner_member_id.as_deref() == Some(owner_member_id) + && context.dispatch_member_id.as_deref() == Some(owner_member_id) + && context.source_kind == AgentOrgTurnSourceKind::Task + && context.source_id == *task_id + && context.activation_generation == Some(*activation_generation) + } + AdmissionKind::UserDirectedWork { + dispatch_member_id, + source, + } => { + let source_matches = match source { + UserDirectedSource::DirectMember { source_event_id } => { + context.source_kind == AgentOrgTurnSourceKind::DirectMember + && context.source_id == *source_event_id + && context.root_authority_turn_id.as_deref() + == Some(request.turn_intent_id.as_str()) + } + UserDirectedSource::GroupMention { source_inbox_id } => { + context.source_kind == AgentOrgTurnSourceKind::GroupMention + && context.source_id == source_inbox_id.to_string() + && context.root_authority_turn_id.as_deref() + == Some(request.turn_intent_id.as_str()) + } + UserDirectedSource::MemberInbox { source_inbox_id } => { + context.source_kind == AgentOrgTurnSourceKind::MemberInbox + && context.source_id == source_inbox_id.to_string() + && context.root_authority_turn_id.is_none() + } + }; + context.turn_kind == AgentOrgTurnKind::UserDirectedWork + && context.participant_id == *dispatch_member_id + && context.dispatch_member_id.as_deref() == Some(dispatch_member_id) + && source_matches + } + }; + if !common_matches || !kind_matches { + return Err(invariant_error(format!( + "companion context replay mismatch for {}/{}", + request.session_id, request.turn_intent_id + ))); + } + Ok(()) +} + +fn validate_non_empty(request: &AgentOrgTurnAdmission) -> Result<(), String> { + if request.org_run_id.trim().is_empty() + || request.session_id.trim().is_empty() + || request.turn_intent_id.trim().is_empty() + { + return Err(invariant_error( + "run/session/turn identity must not be empty".to_string(), + )); + } + Ok(()) +} + +pub(crate) fn require_context_with_connection( + conn: &Connection, + session_id: &str, + turn_intent_id: &str, +) -> Result { + read_context_optional(conn, session_id, turn_intent_id)?.ok_or_else(|| { + invariant_error(format!( + "missing companion context for {session_id}/{turn_intent_id}" + )) + }) +} + +fn read_context_optional( + conn: &Connection, + session_id: &str, + turn_intent_id: &str, +) -> Result, String> { + conn.query_row( + "SELECT context_id, session_id, turn_intent_id, org_run_id, + participant_id, turn_kind, task_id, owner_member_id, + dispatch_member_id, member_dispatch_sequence, source_kind, + source_id, root_authority_turn_id, actor_version, + activation_generation, created_at + FROM agent_org_runtime_turn_contexts + WHERE session_id=?1 AND turn_intent_id=?2", + params![session_id, turn_intent_id], + decode_context, + ) + .optional() + .map_err(|error| invariant_error(format!("failed to decode companion context: {error}"))) +} + +fn decode_context(row: &rusqlite::Row<'_>) -> rusqlite::Result { + let kind_raw: String = row.get(5)?; + let source_raw: String = row.get(10)?; + let turn_kind = AgentOrgTurnKind::parse(&kind_raw).ok_or_else(|| { + rusqlite::Error::FromSqlConversionFailure( + 5, + rusqlite::types::Type::Text, + format!("unknown Agent Org Turn kind {kind_raw:?}").into(), + ) + })?; + let source_kind = AgentOrgTurnSourceKind::parse(&source_raw).ok_or_else(|| { + rusqlite::Error::FromSqlConversionFailure( + 10, + rusqlite::types::Type::Text, + format!("unknown Agent Org Turn source {source_raw:?}").into(), + ) + })?; + Ok(AgentOrgTurnContext { + context_id: row.get(0)?, + session_id: row.get(1)?, + turn_intent_id: row.get(2)?, + org_run_id: row.get(3)?, + participant_id: row.get(4)?, + turn_kind, + task_id: row.get(6)?, + owner_member_id: row.get(7)?, + dispatch_member_id: row.get(8)?, + member_dispatch_sequence: row.get(9)?, + source_kind, + source_id: row.get(11)?, + root_authority_turn_id: row.get(12)?, + actor_version: row.get(13)?, + activation_generation: row.get(14)?, + created_at: row.get(15)?, + }) +} + +/// Agent Org-owned restart reconciliation. Generic SDE recovery never joins +/// or queries the companion table. +pub fn reconcile_in_flight_after_restart(conn: &Connection) -> Result { + // Decode every persisted in-flight context first. Unknown discriminants or + // malformed rows stop reconciliation before any state is changed. + let mut statement = conn + .prepare( + "SELECT context.context_id, context.session_id, context.turn_intent_id, + context.org_run_id, context.participant_id, context.turn_kind, + context.task_id, context.owner_member_id, + context.dispatch_member_id, context.member_dispatch_sequence, + context.source_kind, context.source_id, + context.root_authority_turn_id, context.actor_version, + context.activation_generation, context.created_at + FROM agent_org_runtime_turn_contexts context + JOIN session_turn_intents intent + ON intent.session_id=context.session_id + AND intent.turn_intent_id=context.turn_intent_id + WHERE intent.status IN ('optimistic', 'queued', 'running')", + ) + .map_err(|error| error.to_string())?; + let rows = statement + .query_map([], decode_context) + .map_err(|error| error.to_string())?; + for row in rows { + row.map_err(|error| invariant_error(format!("recovery decode failed: {error}")))?; + } + drop(statement); + + let now = chrono::Utc::now().to_rfc3339(); + let affected = conn + .execute( + "UPDATE session_turn_intents AS intent + SET status='stale', updated_at=?1 + WHERE intent.org_run_id IS NOT NULL + AND intent.status IN ('optimistic', 'queued') + AND NOT ( + intent.status='queued' + AND EXISTS ( + SELECT 1 + FROM agent_org_runtime_initial_inputs initial + JOIN agent_org_runtime_turn_contexts context + ON context.org_run_id=initial.org_run_id + AND context.turn_intent_id=initial.turn_intent_id + JOIN agent_org_runtime_runs run + ON run.id=initial.org_run_id + WHERE initial.org_run_id=intent.org_run_id + AND initial.turn_intent_id=intent.turn_intent_id + AND initial.status IN ('queued', 'dispatched') + AND initial.message_id=intent.client_message_id + AND context.session_id=intent.session_id + AND context.turn_kind='coordinator' + AND context.source_kind='root_turn' + AND context.activation_generation=run.activation_generation + AND run.root_session_id=intent.session_id + AND run.status='running' + ) + )", + [&now], + ) + .map_err(|error| error.to_string())?; + + let missing_running: i64 = conn + .query_row( + "SELECT COUNT(*) + FROM session_turn_intents intent + LEFT JOIN agent_org_runtime_turn_contexts context + ON context.session_id=intent.session_id + AND context.turn_intent_id=intent.turn_intent_id + WHERE intent.org_run_id IS NOT NULL + AND intent.status='running' + AND context.context_id IS NULL", + [], + |row| row.get(0), + ) + .map_err(|error| error.to_string())?; + if missing_running != 0 { + tracing::error!( + missing_running, + event = "agent_org_running_turn_context_missing", + "retained contextless running Agent Org Turns as unknown/in-flight" + ); + } + Ok(affected) +} + +fn invariant_error(message: String) -> String { + format!("{TURN_CONTEXT_INVARIANT_PREFIX} {message}") +} + +#[cfg(test)] +mod tests; diff --git a/src-tauri/crates/agent-core/src/core/coordination/agent_org_turn_contexts/tests.rs b/src-tauri/crates/agent-core/src/core/coordination/agent_org_turn_contexts/tests.rs new file mode 100644 index 0000000000..8eb996ea73 --- /dev/null +++ b/src-tauri/crates/agent-core/src/core/coordination/agent_org_turn_contexts/tests.rs @@ -0,0 +1,583 @@ +use std::sync::{Arc, Barrier}; +use std::time::Duration; + +use rusqlite::{params, Connection}; + +use super::*; +use crate::definitions::orgs::{AgentOrgLaunchSnapshot, FlatOrgMember, PlanApprovalPolicy}; + +const RUN_ID: &str = "run-a"; +const ROOT_SESSION_ID: &str = "session-root"; +const MEMBER_SESSION_ID: &str = "session-member"; +const MEMBER_ID: &str = "member-a"; + +fn test_upsert_turn_intent( + conn: &Connection, + session_id: &str, + turn_intent_id: &str, + client_message_id: Option<&str>, + org_run_id: Option<&str>, + source: TurnIntentBridgeSource, + status: TurnIntentBridgeStatus, +) -> Result<(), String> { + let now = chrono::Utc::now().to_rfc3339(); + conn.execute( + "INSERT OR IGNORE INTO session_turn_intents ( + session_id, turn_intent_id, client_message_id, org_run_id, + source, status, created_at, updated_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?7)", + params![ + session_id, + turn_intent_id, + client_message_id, + org_run_id, + source.as_str(), + status.as_str(), + now, + ], + ) + .map_err(|error| error.to_string())?; + Ok(()) +} + +fn register_bridge() { + crate::foundation::session_bridge::register_upsert_turn_intent_with_connection( + test_upsert_turn_intent, + ); +} + +fn snapshot_json() -> String { + serde_json::to_string(&AgentOrgLaunchSnapshot { + schema_version: 1, + org_id: "org-a".into(), + org_name: "Team A".into(), + coordinator_role: "Lead".into(), + coordinator_agent_id: "agent-coordinator".into(), + plan_approval_policy: PlanApprovalPolicy::Coordinator, + members: vec![FlatOrgMember { + member_id: MEMBER_ID.into(), + name: "Member A".into(), + role: "Builder".into(), + agent_id: "agent-member".into(), + runtime_config: None, + }], + additional_task_graph_writer_member_ids: Vec::new(), + member_communication_links: Vec::new(), + }) + .expect("serialize launch snapshot") +} + +fn create_fixture(conn: &Connection) { + register_bridge(); + conn.execute_batch( + "PRAGMA foreign_keys=ON; + CREATE TABLE session_turn_intents ( + session_id TEXT NOT NULL, + turn_intent_id TEXT NOT NULL, + client_message_id TEXT, + org_run_id TEXT, + source TEXT NOT NULL, + status TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + PRIMARY KEY(session_id, turn_intent_id) + ); + CREATE TABLE agent_sessions ( + session_id TEXT PRIMARY KEY, + agent_definition_id TEXT, + org_member_id TEXT + ); + CREATE TABLE events (id TEXT PRIMARY KEY, session_id TEXT NOT NULL); + CREATE TABLE agent_org_runtime_runs ( + id TEXT PRIMARY KEY, + root_session_id TEXT, + org_snapshot_json TEXT, + activation_generation INTEGER NOT NULL, + status TEXT NOT NULL + ); + CREATE TABLE agent_org_runtime_member_materializations ( + org_run_id TEXT NOT NULL, + member_id TEXT NOT NULL, + agent_id TEXT NOT NULL, + generation INTEGER NOT NULL, + session_id TEXT NOT NULL, + status TEXT NOT NULL, + PRIMARY KEY(org_run_id, member_id, generation), + FOREIGN KEY(org_run_id) REFERENCES agent_org_runtime_runs(id) ON DELETE CASCADE + ); + CREATE TABLE agent_org_runtime_tasks ( + org_run_id TEXT NOT NULL, + id TEXT NOT NULL, + owner TEXT, + PRIMARY KEY(org_run_id, id), + FOREIGN KEY(org_run_id) REFERENCES agent_org_runtime_runs(id) ON DELETE CASCADE + ); + CREATE TABLE agent_org_runtime_inbox ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + org_run_id TEXT NOT NULL, + recipient_member_id TEXT, + FOREIGN KEY(org_run_id) REFERENCES agent_org_runtime_runs(id) ON DELETE CASCADE + ); + CREATE TABLE agent_org_runtime_initial_inputs ( + org_run_id TEXT PRIMARY KEY, + turn_intent_id TEXT NOT NULL, + message_id TEXT NOT NULL, + status TEXT NOT NULL, + FOREIGN KEY(org_run_id) REFERENCES agent_org_runtime_runs(id) ON DELETE CASCADE + );", + ) + .expect("create canonical fixture schema"); + create_schema(conn).expect("create Turn context schema"); + conn.execute( + "INSERT INTO agent_org_runtime_runs + (id, root_session_id, org_snapshot_json, activation_generation, status) + VALUES (?1, ?2, ?3, 1, 'running')", + params![RUN_ID, ROOT_SESSION_ID, snapshot_json()], + ) + .expect("seed run"); + conn.execute_batch( + "INSERT INTO agent_sessions VALUES + ('session-root', 'agent-coordinator', 'coordinator'), + ('session-member', 'agent-member', 'member-a'); + INSERT INTO agent_org_runtime_member_materializations + (org_run_id, member_id, agent_id, generation, session_id, status) + VALUES + ('run-a', 'coordinator', 'agent-coordinator', 1, 'session-root', 'succeeded'), + ('run-a', 'member-a', 'agent-member', 1, 'session-member', 'succeeded'); + INSERT INTO agent_org_runtime_tasks VALUES ('run-a', 'task-a', 'member-a'); + INSERT INTO events VALUES ('event-direct', 'session-member'); + INSERT INTO agent_org_runtime_inbox (org_run_id, recipient_member_id) + VALUES ('run-a', 'member-a'), ('run-a', 'member-a');", + ) + .expect("seed canonical identities and sources"); +} + +fn connection() -> Connection { + let conn = Connection::open_in_memory().expect("open in-memory database"); + create_fixture(&conn); + conn +} + +fn accept_in_transaction( + conn: &mut Connection, + request: &AgentOrgTurnAdmission, +) -> Result { + let transaction = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .map_err(|error| error.to_string())?; + let context = accept_with_connection(&transaction, request)?; + transaction.commit().map_err(|error| error.to_string())?; + Ok(context) +} + +fn task_request(turn_id: &str) -> AgentOrgTurnAdmission { + AgentOrgTurnAdmission::task_execution( + RUN_ID, + MEMBER_SESSION_ID, + turn_id, + Some(format!("message-{turn_id}")), + "task-a", + MEMBER_ID, + 1, + ) +} + +fn direct_request(turn_id: &str) -> AgentOrgTurnAdmission { + AgentOrgTurnAdmission::direct_member( + RUN_ID, + MEMBER_SESSION_ID, + turn_id, + Some(format!("message-{turn_id}")), + MEMBER_ID, + "event-direct", + ) +} + +fn group_request(turn_id: &str) -> AgentOrgTurnAdmission { + AgentOrgTurnAdmission::group_mention( + RUN_ID, + MEMBER_SESSION_ID, + turn_id, + Some(format!("message-{turn_id}")), + MEMBER_ID, + 1, + ) +} + +fn inbox_request(turn_id: &str) -> AgentOrgTurnAdmission { + AgentOrgTurnAdmission::member_inbox( + RUN_ID, + MEMBER_SESSION_ID, + turn_id, + Some(format!("message-{turn_id}")), + MEMBER_ID, + 2, + ) +} + +fn status(conn: &Connection, turn_id: &str) -> String { + conn.query_row( + "SELECT status FROM session_turn_intents WHERE turn_intent_id=?1", + [turn_id], + |row| row.get(0), + ) + .expect("read Turn status") +} + +fn row_count(conn: &Connection, table: &str) -> i64 { + conn.query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| { + row.get(0) + }) + .unwrap_or_else(|error| panic!("count {table}: {error}")) +} + +#[test] +fn coordinator_is_root_scoped_and_never_allocates_member_sequence() { + let mut conn = connection(); + let request = AgentOrgTurnAdmission::coordinator( + RUN_ID, + ROOT_SESSION_ID, + "turn-root", + Some("message-root".into()), + TurnIntentBridgeSource::UserSubmit, + ); + let first = accept_in_transaction(&mut conn, &request).expect("accept Root Turn"); + let replay = accept_in_transaction(&mut conn, &request).expect("replay Root Turn"); + + assert_eq!(first, replay); + assert_eq!(first.turn_kind, AgentOrgTurnKind::Coordinator); + assert_eq!(first.participant_id, COORDINATOR_MEMBER_ID); + assert_eq!(first.source_kind, AgentOrgTurnSourceKind::RootTurn); + assert_eq!(first.activation_generation, Some(1)); + assert_eq!(first.member_dispatch_sequence, None); + assert_eq!( + row_count(&conn, "agent_org_runtime_member_dispatch_allocators"), + 0 + ); + + for (turn_id, source) in [ + ("turn-root-queue", TurnIntentBridgeSource::Queue), + ("turn-root-force", TurnIntentBridgeSource::ForceSend), + ] { + let context = accept_in_transaction( + &mut conn, + &AgentOrgTurnAdmission::coordinator( + RUN_ID, + ROOT_SESSION_ID, + turn_id, + Some(format!("message-{turn_id}")), + source, + ), + ) + .expect("accept queued/steered Root Turn"); + assert_eq!(context.turn_kind, AgentOrgTurnKind::Coordinator); + assert_eq!(context.member_dispatch_sequence, None); + } + + let member_as_root = AgentOrgTurnAdmission::coordinator( + RUN_ID, + MEMBER_SESSION_ID, + "turn-member-untyped", + None, + TurnIntentBridgeSource::AgentOrg, + ); + let error = accept_in_transaction(&mut conn, &member_as_root) + .expect_err("legacy Member path must fail closed"); + assert!(error.contains("is not canonical Root"), "{error}"); + assert_eq!(row_count(&conn, "session_turn_intents"), 3); +} + +#[test] +fn every_member_kind_and_source_shares_one_fifo_and_replay_does_not_bump_it() { + let mut conn = connection(); + let requests = [ + task_request("turn-task"), + direct_request("turn-direct"), + group_request("turn-group"), + inbox_request("turn-inbox"), + ]; + let contexts = requests + .iter() + .map(|request| accept_in_transaction(&mut conn, request).expect("accept Member Turn")) + .collect::>(); + + assert_eq!(contexts[0].turn_kind, AgentOrgTurnKind::TaskExecution); + assert_eq!( + contexts[1].source_kind, + AgentOrgTurnSourceKind::DirectMember + ); + assert_eq!( + contexts[2].source_kind, + AgentOrgTurnSourceKind::GroupMention + ); + assert_eq!(contexts[3].source_kind, AgentOrgTurnSourceKind::MemberInbox); + assert_eq!( + contexts + .iter() + .map(|context| context.member_dispatch_sequence.unwrap()) + .collect::>(), + vec![1, 2, 3, 4] + ); + assert_eq!( + contexts[1].root_authority_turn_id.as_deref(), + Some("turn-direct") + ); + assert_eq!( + contexts[2].root_authority_turn_id.as_deref(), + Some("turn-group") + ); + assert_eq!(contexts[3].root_authority_turn_id, None); + + let replay = accept_in_transaction(&mut conn, &requests[0]).expect("exact replay"); + assert_eq!(replay.member_dispatch_sequence, Some(1)); + let next_sequence: i64 = conn + .query_row( + "SELECT next_sequence FROM agent_org_runtime_member_dispatch_allocators + WHERE org_run_id=?1 AND member_id=?2", + params![RUN_ID, MEMBER_ID], + |row| row.get(0), + ) + .expect("read allocator"); + assert_eq!(next_sequence, 5); + + let conflict = AgentOrgTurnAdmission::member_inbox( + RUN_ID, + MEMBER_SESSION_ID, + "turn-task", + Some("message-turn-task".into()), + MEMBER_ID, + 2, + ); + assert!(accept_in_transaction(&mut conn, &conflict) + .expect_err("kind-changing replay must fail") + .contains("context replay mismatch")); +} + +#[test] +fn user_directed_actor_version_is_independent_from_formal_activation_generation() { + let mut conn = connection(); + conn.execute( + "UPDATE agent_org_runtime_runs SET activation_generation=2 WHERE id=?1", + [RUN_ID], + ) + .unwrap(); + + let context = accept_in_transaction(&mut conn, &direct_request("turn-cross-activation")) + .expect("UDW keeps the stable materialized actor across a formal generation bump"); + assert_eq!(context.actor_version, Some(1)); + assert_eq!(context.activation_generation, None); + + let formal = task_request("turn-stale-formal"); + assert!(accept_in_transaction(&mut conn, &formal) + .expect_err("formal work must carry the current activation generation") + .contains("TaskExecution authority mismatch")); +} + +#[test] +fn canonical_check_and_exhaustive_decode_fail_closed() { + let conn = connection(); + test_upsert_turn_intent( + &conn, + ROOT_SESSION_ID, + "turn-invalid-shape", + None, + Some(RUN_ID), + TurnIntentBridgeSource::AgentOrg, + TurnIntentBridgeStatus::Queued, + ) + .unwrap(); + let invalid = conn.execute( + "INSERT INTO agent_org_runtime_turn_contexts ( + session_id, turn_intent_id, org_run_id, participant_id, turn_kind, + dispatch_member_id, member_dispatch_sequence, source_kind, source_id, + activation_generation, created_at + ) VALUES (?1, 'turn-invalid-shape', ?2, 'coordinator', 'coordinator', + 'member-a', 1, 'root_turn', 'turn-invalid-shape', 1, 'now')", + params![ROOT_SESSION_ID, RUN_ID], + ); + assert!(invalid.is_err(), "row-shape CHECK must reject mixed kinds"); + + test_upsert_turn_intent( + &conn, + ROOT_SESSION_ID, + "turn-unknown-kind", + None, + Some(RUN_ID), + TurnIntentBridgeSource::AgentOrg, + TurnIntentBridgeStatus::Queued, + ) + .unwrap(); + conn.execute_batch("PRAGMA ignore_check_constraints=ON;") + .expect("simulate corrupted stored discriminant"); + conn.execute( + "INSERT INTO agent_org_runtime_turn_contexts ( + session_id, turn_intent_id, org_run_id, participant_id, turn_kind, + source_kind, source_id, activation_generation, created_at + ) VALUES (?1, 'turn-unknown-kind', ?2, 'coordinator', 'future_kind', + 'root_turn', 'turn-unknown-kind', 1, 'now')", + params![ROOT_SESSION_ID, RUN_ID], + ) + .expect("seed corrupt future discriminant"); + conn.execute_batch("PRAGMA ignore_check_constraints=OFF;") + .unwrap(); + let error = require_context_with_connection(&conn, ROOT_SESSION_ID, "turn-unknown-kind") + .expect_err("unknown discriminant must not default"); + assert!(error.contains("unknown Agent Org Turn kind"), "{error}"); +} + +#[test] +fn transaction_failure_rolls_back_allocator_base_and_context() { + for failure_target in ["base", "context"] { + let mut conn = connection(); + let trigger = if failure_target == "base" { + "CREATE TRIGGER fail_admission BEFORE INSERT ON session_turn_intents + BEGIN SELECT RAISE(ABORT, 'base fault'); END;" + } else { + "CREATE TRIGGER fail_admission BEFORE INSERT ON agent_org_runtime_turn_contexts + BEGIN SELECT RAISE(ABORT, 'context fault'); END;" + }; + conn.execute_batch(trigger).expect("install fault trigger"); + let transaction = conn + .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) + .unwrap(); + let error = accept_with_connection(&transaction, &task_request("turn-fault")) + .expect_err("fault must abort admission"); + assert!(error.contains("fault"), "{error}"); + transaction.rollback().expect("rollback failed admission"); + assert_eq!(row_count(&conn, "session_turn_intents"), 0); + assert_eq!(row_count(&conn, "agent_org_runtime_turn_contexts"), 0); + assert_eq!( + row_count(&conn, "agent_org_runtime_member_dispatch_allocators"), + 0 + ); + } +} + +#[test] +fn fifty_concurrent_mixed_sources_receive_one_strict_sequence() { + const COUNT: usize = 50; + let directory = tempfile::tempdir().expect("temporary database directory"); + let path = directory.path().join("sessions.db"); + let conn = Connection::open(&path).expect("create shared database"); + conn.execute_batch("PRAGMA journal_mode=WAL;").unwrap(); + create_fixture(&conn); + drop(conn); + + let barrier = Arc::new(Barrier::new(COUNT)); + let handles = (0..COUNT) + .map(|index| { + let path = path.clone(); + let barrier = Arc::clone(&barrier); + std::thread::spawn(move || { + let mut conn = Connection::open(path).expect("open shared database"); + conn.busy_timeout(Duration::from_secs(20)).unwrap(); + conn.execute_batch("PRAGMA foreign_keys=ON;").unwrap(); + let turn_id = format!("turn-concurrent-{index}"); + let request = match index % 4 { + 0 => task_request(&turn_id), + 1 => direct_request(&turn_id), + 2 => group_request(&turn_id), + _ => inbox_request(&turn_id), + }; + barrier.wait(); + accept_in_transaction(&mut conn, &request) + .expect("concurrent admission") + .member_dispatch_sequence + .expect("Member sequence") + }) + }) + .collect::>(); + let mut sequences = handles + .into_iter() + .map(|handle| handle.join().expect("admission thread")) + .collect::>(); + sequences.sort_unstable(); + assert_eq!(sequences, (1..=COUNT as i64).collect::>()); +} + +#[test] +fn recovery_preserves_only_typed_canonical_initial_and_keeps_running_unknown() { + let mut conn = connection(); + let initial = AgentOrgTurnAdmission::coordinator( + RUN_ID, + ROOT_SESSION_ID, + "turn-initial", + Some("message-initial".into()), + TurnIntentBridgeSource::AgentOrg, + ); + accept_in_transaction(&mut conn, &initial).unwrap(); + conn.execute( + "INSERT INTO agent_org_runtime_initial_inputs + (org_run_id, turn_intent_id, message_id, status) + VALUES (?1, 'turn-initial', 'message-initial', 'queued')", + [RUN_ID], + ) + .unwrap(); + let later = AgentOrgTurnAdmission::coordinator( + RUN_ID, + ROOT_SESSION_ID, + "turn-later-root", + Some("message-later".into()), + TurnIntentBridgeSource::AgentOrg, + ); + accept_in_transaction(&mut conn, &later).unwrap(); + for (turn_id, state) in [ + ("turn-contextless-queued", "queued"), + ("turn-contextless-running", "running"), + ("turn-contextless-terminal", "completed"), + ] { + conn.execute( + "INSERT INTO session_turn_intents + (session_id, turn_intent_id, org_run_id, source, status, created_at, updated_at) + VALUES (?1, ?2, ?3, 'agent_org', ?4, 'now', 'now')", + params![ROOT_SESSION_ID, turn_id, RUN_ID, state], + ) + .unwrap(); + } + + assert_eq!(reconcile_in_flight_after_restart(&conn).unwrap(), 2); + assert_eq!(status(&conn, "turn-initial"), "queued"); + assert_eq!(status(&conn, "turn-later-root"), "stale"); + assert_eq!(status(&conn, "turn-contextless-queued"), "stale"); + assert_eq!(status(&conn, "turn-contextless-running"), "running"); + assert_eq!(status(&conn, "turn-contextless-terminal"), "completed"); + + conn.execute( + "UPDATE agent_org_runtime_initial_inputs SET message_id='wrong-message' WHERE org_run_id=?1", + [RUN_ID], + ) + .unwrap(); + assert_eq!(reconcile_in_flight_after_restart(&conn).unwrap(), 1); + assert_eq!(status(&conn, "turn-initial"), "stale"); +} + +#[test] +fn run_delete_cascades_context_and_allocator_without_touching_generic_rows() { + let mut conn = connection(); + accept_in_transaction(&mut conn, &task_request("turn-delete")).unwrap(); + conn.execute( + "INSERT INTO session_turn_intents + (session_id, turn_intent_id, source, status, created_at, updated_at) + VALUES ('sde-session', 'sde-turn', 'user_submit', 'queued', 'now', 'now')", + [], + ) + .unwrap(); + + conn.execute( + "DELETE FROM session_turn_intents WHERE org_run_id=?1", + [RUN_ID], + ) + .unwrap(); + assert_eq!(row_count(&conn, "agent_org_runtime_turn_contexts"), 0); + assert_eq!( + row_count(&conn, "agent_org_runtime_member_dispatch_allocators"), + 1 + ); + conn.execute("DELETE FROM agent_org_runtime_runs WHERE id=?1", [RUN_ID]) + .unwrap(); + assert_eq!( + row_count(&conn, "agent_org_runtime_member_dispatch_allocators"), + 0 + ); + assert_eq!(row_count(&conn, "session_turn_intents"), 1); +} diff --git a/src-tauri/crates/agent-core/src/core/coordination/mod.rs b/src-tauri/crates/agent-core/src/core/coordination/mod.rs index ef86d4e571..ea0b1f18cc 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/mod.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/mod.rs @@ -26,6 +26,7 @@ pub mod agent_org_plan_approvals; pub mod agent_org_run_events; pub mod agent_org_runs; pub mod agent_org_tasks; +pub(crate) mod agent_org_turn_contexts; pub mod agent_org_watchdog; pub mod child_done_wake; pub mod routine_scheduler; @@ -41,3 +42,11 @@ mod schema; pub fn init_agent_org_schemas(conn: &rusqlite::Connection) -> rusqlite::Result<()> { schema::initialize(conn) } + +/// Reconcile Agent Org-owned Turn lifecycle only after its companion schema +/// has been initialized and verified. +pub fn reconcile_agent_org_turns_after_restart( + conn: &rusqlite::Connection, +) -> Result { + agent_org_turn_contexts::reconcile_in_flight_after_restart(conn) +} diff --git a/src-tauri/crates/agent-core/src/core/coordination/schema.rs b/src-tauri/crates/agent-core/src/core/coordination/schema.rs index fb74292d03..fdf69cc146 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/schema.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/schema.rs @@ -12,10 +12,10 @@ use rusqlite::{ffi, Connection, Error as SqliteError, Result as SqliteResult}; use super::{ agent_inbox, agent_member_interventions, agent_org_plan_approvals, agent_org_runs, - agent_org_tasks, agent_org_watchdog, + agent_org_tasks, agent_org_turn_contexts, agent_org_watchdog, }; -const RUNTIME_TABLES: [&str; 13] = [ +const RUNTIME_TABLES_V1: [&str; 13] = [ "agent_org_runtime_runs", "agent_org_runtime_run_progress", "agent_org_runtime_member_materializations", @@ -31,6 +31,24 @@ const RUNTIME_TABLES: [&str; 13] = [ "agent_org_runtime_member_interventions", ]; +const RUNTIME_TABLES: [&str; 15] = [ + "agent_org_runtime_runs", + "agent_org_runtime_run_progress", + "agent_org_runtime_member_materializations", + "agent_org_runtime_initial_inputs", + "agent_org_runtime_plan_approvals", + "agent_org_runtime_recovery_attempts", + "agent_org_runtime_tasks", + "agent_org_runtime_task_events", + "agent_org_runtime_task_schema_migrations", + "agent_org_runtime_inbox", + "agent_org_runtime_inbox_materializations", + "agent_org_runtime_inbox_delivery_resolutions", + "agent_org_runtime_member_interventions", + "agent_org_runtime_member_dispatch_allocators", + "agent_org_runtime_turn_contexts", +]; + const LEGACY_TABLES: [&str; 13] = [ "agent_org_runs", "agent_org_run_progress", @@ -71,20 +89,26 @@ const DROP_LEGACY_SCHEMA: &str = "DROP TABLE IF EXISTS agent_inbox_materializati type SchemaManifest = BTreeMap<(String, String), (String, String)>; pub(super) fn initialize(conn: &Connection) -> SqliteResult<()> { + let expected_v1 = expected_manifest_v1()?; let expected = expected_manifest()?; let tx = database::db::begin_immediate(conn)?; let runtime_table_count = count_known_tables(&tx, &RUNTIME_TABLES)?; - let fresh = match runtime_table_count { - 0 => true, + let (fresh, upgraded_from_v1) = match runtime_table_count { + 0 => (true, false), + count if count == RUNTIME_TABLES_V1.len() => { + verify_manifest(&tx, &expected_v1)?; + (false, true) + } count if count == RUNTIME_TABLES.len() => { verify_manifest(&tx, &expected)?; - false + (false, false) } count => { return Err(schema_error(format!( - "partial Agent Org runtime schema: found {count} of {} canonical tables", - RUNTIME_TABLES.len() + "partial Agent Org runtime schema: found {count}; expected 0, {} (PR828), or {} (PR3) canonical tables", + RUNTIME_TABLES_V1.len(), + RUNTIME_TABLES.len(), ))) } }; @@ -95,6 +119,8 @@ pub(super) fn initialize(conn: &Connection) -> SqliteResult<()> { if fresh { create_runtime_schema(&tx)?; + } else if upgraded_from_v1 { + agent_org_turn_contexts::create_schema(&tx)?; } verify_manifest(&tx, &expected)?; agent_inbox::repair_dangling_materializations(&tx)?; @@ -118,13 +144,19 @@ pub(super) fn initialize(conn: &Connection) -> SqliteResult<()> { legacy_table_count, legacy_object_count, fresh, - idempotent = !fresh, + upgraded_from_v1, + idempotent = !fresh && !upgraded_from_v1, "initialized isolated Agent Org runtime schema" ); Ok(()) } fn create_runtime_schema(conn: &Connection) -> SqliteResult<()> { + create_runtime_schema_v1(conn)?; + agent_org_turn_contexts::create_schema(conn) +} + +fn create_runtime_schema_v1(conn: &Connection) -> SqliteResult<()> { agent_org_runs::create_schema(conn)?; agent_inbox::create_schema(conn)?; agent_org_tasks::create_schema(conn)?; @@ -140,6 +172,13 @@ fn expected_manifest() -> SqliteResult { read_manifest(&expected) } +fn expected_manifest_v1() -> SqliteResult { + let expected = Connection::open_in_memory()?; + expected.execute_batch("PRAGMA foreign_keys=ON;")?; + create_runtime_schema_v1(&expected)?; + read_manifest(&expected) +} + fn verify_manifest(conn: &Connection, expected: &SchemaManifest) -> SqliteResult<()> { let actual = read_manifest(conn)?; if &actual == expected { @@ -593,6 +632,33 @@ mod tests { assert_eq!(snapshot, "{\"team\":\"A\"}"); } + #[test] + fn exact_pr828_manifest_is_extended_atomically_to_pr3() { + let conn = connection(); + create_runtime_schema_v1(&conn).expect("PR828 runtime fixture"); + conn.execute( + "INSERT INTO agent_org_runtime_runs ( + id, org_id, coordinator_agent_id, org_snapshot_json, entry_mode, + status, created_at, updated_at + ) VALUES ('run-v1', 'org-v1', 'agent-v1', '{}', + 'standalone_session', 'idle', 'now', 'now')", + [], + ) + .expect("seed PR828 runtime row"); + + initialize(&conn).expect("upgrade exact PR828 manifest"); + + verify_manifest(&conn, &expected_manifest().expect("PR3 manifest")) + .expect("canonical PR3 manifest"); + assert_eq!(count_known_tables(&conn, &RUNTIME_TABLES).unwrap(), 15); + assert_eq!(row_count(&conn, "agent_org_runtime_runs"), 1); + assert_eq!( + row_count(&conn, "agent_org_runtime_member_dispatch_allocators"), + 0 + ); + assert_eq!(row_count(&conn, "agent_org_runtime_turn_contexts"), 0); + } + #[test] fn partial_or_unknown_runtime_schema_fails_closed_before_legacy_cleanup() { for mutate in ["partial", "changed", "extra_index"] { @@ -693,7 +759,7 @@ mod tests { let conn = Connection::open(path).expect("reopen shared database"); verify_manifest(&conn, &expected_manifest().expect("expected manifest")) .expect("canonical manifest after concurrent init"); - assert_eq!(count_known_tables(&conn, &RUNTIME_TABLES).unwrap(), 13); + assert_eq!(count_known_tables(&conn, &RUNTIME_TABLES).unwrap(), 15); } #[test] 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 949fa37e97..f3586fb877 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 @@ -248,6 +248,26 @@ pub(crate) async fn send_message_impl( (None, None) => preflight_org_run_id, }; + // PR3 admits only canonical Coordinator/Root turns. Member Task/direct/ + // group/inbox producers must use their typed authority constructors in + // their owning PR; a legacy Member call fails here before base intent or + // scheduler state can be written. + if let Some(run_id) = effective_intent_org_run_id.as_deref() { + let admission = + crate::coordination::agent_org_turn_contexts::AgentOrgTurnAdmission::coordinator( + run_id, + &session_id, + &effective_turn_intent_id, + client_message_id.clone(), + source, + ); + tokio::task::spawn_blocking(move || { + crate::coordination::agent_org_turn_contexts::accept(&admission) + }) + .await + .map_err(|error| format!("Agent Org Turn admission worker failed: {error}"))??; + } + // Wingman resume: reopen the bottom bar. On fresh start the frontend // sends `wingman_start` which opens the bar, but after app restart // the frontend doesn't re-send that command. Best-effort — a missing @@ -349,14 +369,16 @@ pub(crate) async fn send_message_impl( // part of accepting the control action. If the durable takeover row // cannot be written, do not inject a message that Wake may race. persist_direct_user_intervention(direct_user_intervention.clone()).await?; - crate::foundation::session_bridge::upsert_turn_intent( - &session_id, - &effective_turn_intent_id, - client_message_id.as_deref(), - effective_intent_org_run_id.as_deref(), - source, - crate::foundation::session_bridge::TurnIntentBridgeStatus::Queued, - ); + if effective_intent_org_run_id.is_none() { + crate::foundation::session_bridge::upsert_turn_intent( + &session_id, + &effective_turn_intent_id, + client_message_id.as_deref(), + None, + source, + crate::foundation::session_bridge::TurnIntentBridgeStatus::Queued, + ); + } session_handle .steering_queue .lock() @@ -571,12 +593,20 @@ pub(crate) async fn send_message_impl( let status_sid = sid.clone(); let status_wake_run_id = org_wake_run_id.clone(); let status_intent_run_id = intent_org_run_id.clone(); + let status_turn_intent_id = turn_intent_id.clone(); match tokio::task::spawn_blocking(move || { database::db::with_sessions_writer(|| -> Result { let mut conn = database::db::get_connection().map_err(|err| err.to_string())?; let tx = conn .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate) .map_err(|err| err.to_string())?; + if status_wake_run_id.is_some() || status_intent_run_id.is_some() { + crate::coordination::agent_org_turn_contexts::require_context_with_connection( + &tx, + &status_sid, + &status_turn_intent_id, + )?; + } let updated = if let Some(run_id) = status_wake_run_id.as_deref() { promote_agent_org_wake_session_to_running(&tx, run_id, &status_sid)? } else if let Some(run_id) = status_intent_run_id.as_deref() { @@ -781,14 +811,16 @@ pub(crate) async fn send_message_impl( // `running` / terminal as the turn executes; `invalidate_pending` // marks it `stale` if rewound before it ran. See `session_turn_intents` // for the state machine. - crate::foundation::session_bridge::upsert_turn_intent( - &session_id, - &effective_turn_intent_id, - msg.client_message_id.as_deref(), - effective_intent_org_run_id.as_deref(), - source, - crate::foundation::session_bridge::TurnIntentBridgeStatus::Queued, - ); + if effective_intent_org_run_id.is_none() { + crate::foundation::session_bridge::upsert_turn_intent( + &session_id, + &effective_turn_intent_id, + msg.client_message_id.as_deref(), + None, + source, + crate::foundation::session_bridge::TurnIntentBridgeStatus::Queued, + ); + } let enqueue_result = session_handle .scheduler diff --git a/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/group_chat.rs b/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/group_chat.rs index 5812298034..9ba30233c2 100644 --- a/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/group_chat.rs +++ b/src-tauri/crates/agent-core/src/state/commands/session/org_tasks/group_chat.rs @@ -15,6 +15,7 @@ use crate::coordination::agent_inbox::{ use crate::coordination::agent_org_runs::{ AgentOrgRunContext, AgentOrgRunStatus, COORDINATOR_MEMBER_ID, }; +use crate::coordination::agent_org_turn_contexts::TURN_CONTEXT_INVARIANT_PREFIX; use crate::state::AgentAppState; use super::context::session_org_read_context; @@ -350,7 +351,7 @@ async fn agent_org_send_group_chat_message_impl_with_display( )?; } let row = tokio::task::spawn_blocking(move || { - persist_group_chat_message( + persist_pr3_group_chat_message( &durable_context, &durable_target_agent_id, &durable_target_member_id, @@ -398,6 +399,32 @@ async fn agent_org_send_group_chat_message_impl_with_display( }) } +/// PR3 wires only the canonical Coordinator/Root producer. Persisting a user +/// message for a Member would leave a durable Inbox source that the legacy +/// wake loop can never admit with typed Member authority, causing an endless +/// retry instead of the required fail-closed result. Reject at the public +/// command boundary before the Inbox or intervention transaction starts. +pub(super) fn persist_pr3_group_chat_message( + context: &AgentOrgRunContext, + target_agent_id: &str, + target_member_id: &str, + content: &str, + display_text: Option<&str>, +) -> Result { + if target_member_id != COORDINATOR_MEMBER_ID { + return Err(format!( + "{TURN_CONTEXT_INVARIANT_PREFIX} PR3 does not admit legacy Member group/inbox producer {target_member_id:?} without typed authority" + )); + } + persist_group_chat_message( + context, + target_agent_id, + target_member_id, + content, + display_text, + ) +} + /// Persist the user's Group Chat message and clear the target member's direct /// intervention as one state transition. The Run status is re-read inside the /// same IMMEDIATE transaction so a stale Run View can never write into a Run 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 8c9939f5ae..9ac1fcbcc7 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 @@ -94,6 +94,30 @@ fn inbox_count_for_member(context: &AgentOrgRunContext, member_id: &str) -> usiz usize::try_from(count).expect("non-negative inbox count") } +#[test] +fn pr3_group_chat_rejects_legacy_member_before_inbox_write() { + let _sandbox = test_helpers::test_env::sandbox(); + let context = prepare_command_run("running"); + + let error = persist_pr3_group_chat_message( + &context, + "builtin:sde", + "member-planner", + "This Member producer has no typed PR3 authority", + Some("@Planner This Member producer has no typed PR3 authority"), + ) + .expect_err("PR3 must reject the legacy Member producer"); + + assert!( + error.starts_with( + crate::coordination::agent_org_turn_contexts::TURN_CONTEXT_INVARIANT_PREFIX + ), + "{error}" + ); + assert!(error.contains("without typed authority"), "{error}"); + assert_eq!(inbox_count_for_member(&context, "member-planner"), 0); +} + fn inbox_record( sender_member_id: Option<&str>, recipient_member_id: Option<&str>, diff --git a/src-tauri/crates/session-persistence/src/turn_intents.rs b/src-tauri/crates/session-persistence/src/turn_intents.rs index 93e4fd49ad..704b81f255 100644 --- a/src-tauri/crates/session-persistence/src/turn_intents.rs +++ b/src-tauri/crates/session-persistence/src/turn_intents.rs @@ -444,35 +444,6 @@ pub fn reconcile_in_flight_after_restart(conn: &Connection) -> Result Result { - let now = Utc::now().to_rfc3339(); - let affected = conn.execute( - "UPDATE session_turn_intents - SET status = 'stale', updated_at = ?1 - WHERE org_run_id IS NOT NULL - AND status IN ('optimistic', 'queued') - AND NOT ( - status = 'queued' - AND EXISTS ( - SELECT 1 FROM agent_org_runtime_initial_inputs initial - WHERE initial.org_run_id=session_turn_intents.org_run_id - AND initial.turn_intent_id=session_turn_intents.turn_intent_id - AND initial.status IN ('queued', 'dispatched') - ) - )", - [now], - )?; - Ok(affected) -} - /// Lookup a single intent row. pub fn get_intent( conn: &Connection, @@ -905,82 +876,48 @@ mod tests { } #[test] - fn agent_org_restart_preserves_running_and_replayable_initial_input_only() { - with_temp_orgii_home(|| { - let session = "test-agent-org-restart"; - let run_id = "agent-org-restart-run"; - let conn = get_connection().expect("open sessions DB"); - agent_core::coordination::init_agent_org_schemas(&conn) - .expect("init Agent Org schemas"); - let now = Utc::now().to_rfc3339(); - conn.execute( - "INSERT INTO agent_org_runtime_runs ( - id, org_id, coordinator_agent_id, root_session_id, - entry_mode, status, has_initial_work, created_at, updated_at - ) VALUES (?1, 'restart-org', 'coordinator', ?2, - 'standalone_session', 'running', 1, ?3, ?3)", - params![run_id, session, &now], - ) - .expect("seed run"); + fn ordinary_restart_reconciliation_has_no_agent_org_schema_dependency() { + let conn = Connection::open_in_memory().expect("open SDE-only fixture"); + conn.execute_batch( + "CREATE TABLE session_turn_intents ( + session_id TEXT NOT NULL, turn_intent_id TEXT NOT NULL, + client_message_id TEXT, org_run_id TEXT, source TEXT NOT NULL, + status TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, + PRIMARY KEY(session_id, turn_intent_id) + ); + INSERT INTO session_turn_intents VALUES + ('sde', 'queued', NULL, NULL, 'user_submit', 'queued', 'now', 'now'), + ('sde', 'running', NULL, NULL, 'user_submit', 'running', 'now', 'now'), + ('org', 'owned', NULL, 'run-a', 'agent_org', 'queued', 'now', 'now');", + ) + .expect("seed fixture without Agent Org context tables"); - for (intent, status) in [ - ("optimistic-noninitial", TurnIntentStatus::Optimistic), - ("queued-noninitial", TurnIntentStatus::Queued), - ("running-final-not-committed", TurnIntentStatus::Running), - ("queued-canonical-initial", TurnIntentStatus::Queued), - ] { - upsert_initial( - session, - intent, - Some(&format!("message-{intent}")), - Some(run_id), - TurnIntentSource::AgentOrg, - status, - ) - .expect("seed Agent Org intent"); - } - conn.execute( - "INSERT INTO agent_org_runtime_initial_inputs ( - org_run_id, turn_intent_id, message_id, content, - payload_json, status, created_at, updated_at - ) VALUES (?1, 'queued-canonical-initial', 'initial-message', - 'initial input', ?2, 'queued', ?3, ?3)", - params![ - run_id, - serde_json::json!({ - "version": 1, - "images": null, - "ideContext": null, - "subAgentIds": [], - }) - .to_string(), - &now, - ], + assert_eq!(reconcile_in_flight_after_restart(&conn).unwrap(), 2); + let sde_states = conn + .prepare( + "SELECT turn_intent_id, status FROM session_turn_intents + WHERE session_id='sde' ORDER BY turn_intent_id", ) - .expect("seed canonical initial input receipt"); - - assert_eq!(reconcile_in_flight_after_restart(&conn).unwrap(), 0); - assert_eq!( - reconcile_agent_org_in_flight_after_restart(&conn).unwrap(), - 2 - ); - let rows = list_for_session(session).expect("load reconciled intents"); - let by_id = rows - .into_iter() - .map(|row| (row.turn_intent_id, row.status)) - .collect::>(); - assert_eq!(by_id["optimistic-noninitial"], TurnIntentStatus::Stale); - assert_eq!(by_id["queued-noninitial"], TurnIntentStatus::Stale); - assert_eq!( - by_id["running-final-not-committed"], - TurnIntentStatus::Running, - "unknown post-crash side effects and an uncommitted final answer must block Idle" - ); - assert_eq!( - by_id["queued-canonical-initial"], - TurnIntentStatus::Queued, - "only the stable initial input receipt is safe to replay with the same ids" - ); - }); + .unwrap() + .query_map([], |row| { + Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)) + }) + .unwrap() + .collect::>>() + .unwrap(); + assert_eq!( + sde_states, + vec![ + ("queued".into(), "stale".into()), + ("running".into(), "failed".into()) + ] + ); + assert_eq!( + get_intent(&conn, "org", "owned") + .unwrap() + .expect("Agent Org row is preserved") + .status, + TurnIntentStatus::Queued + ); } } diff --git a/src-tauri/src/api/agent/test/agent_org.rs b/src-tauri/src/api/agent/test/agent_org.rs index 40f6ade857..16d3fba896 100644 --- a/src-tauri/src/api/agent/test/agent_org.rs +++ b/src-tauri/src/api/agent/test/agent_org.rs @@ -3229,10 +3229,8 @@ pub async fn test_agent_org_simulate_app_restart() -> Json { session_persistence::turn_intents::reconcile_in_flight_after_restart(&conn) .map_err(|err| format!("reconcile_in_flight_after_restart failed: {err}"))?; let agent_org_intents_reconciled = - session_persistence::turn_intents::reconcile_agent_org_in_flight_after_restart(&conn) - .map_err(|err| { - format!("reconcile_agent_org_in_flight_after_restart failed: {err}") - })?; + agent_core::coordination::reconcile_agent_org_turns_after_restart(&conn) + .map_err(|err| format!("reconcile_agent_org_turns failed: {err}"))?; let terminal_sessions_reconciled = reconcile_sessions_with_terminal_turn_markers() .map_err(|err| { format!("reconcile_sessions_with_terminal_turn_markers failed: {err}") diff --git a/src-tauri/src/setup/hooks.rs b/src-tauri/src/setup/hooks.rs index 41699ce48f..210735a133 100644 --- a/src-tauri/src/setup/hooks.rs +++ b/src-tauri/src/setup/hooks.rs @@ -65,7 +65,7 @@ pub(crate) fn register_database_schemas() { } agent_core::coordination::init_agent_org_schemas(conn)?; - match session_persistence::turn_intents::reconcile_agent_org_in_flight_after_restart(conn) { + match agent_core::coordination::reconcile_agent_org_turns_after_restart(conn) { Ok(0) => {} Ok(count) => tracing::info!( "[startup] Reconciled {} Agent Org turn intent(s) from the previous process", From 4c4b53da14303fe2a2c921e84b79adac1fe752a7 Mon Sep 17 00:00:00 2001 From: Shibo Sheng Date: Thu, 20 Aug 2026 01:12:24 +0800 Subject: [PATCH 2/2] refactor(agent-org): remove unpublished schema upgrade path Accept only an empty Agent Org runtime namespace or the exact current 15-table manifest. Remove the obsolete 13-table manifest branch, upgrade-only schema builder, logging, and test. Keep strict corruption checks and supported legacy namespace cleanup unchanged. Refs: #758 Pre-commit hook ran. Total eslint: 5, total circular: 0 --- .../src/core/coordination/schema.rs | 88 ++++--------------- 1 file changed, 15 insertions(+), 73 deletions(-) diff --git a/src-tauri/crates/agent-core/src/core/coordination/schema.rs b/src-tauri/crates/agent-core/src/core/coordination/schema.rs index fdf69cc146..5d9a788013 100644 --- a/src-tauri/crates/agent-core/src/core/coordination/schema.rs +++ b/src-tauri/crates/agent-core/src/core/coordination/schema.rs @@ -15,22 +15,6 @@ use super::{ agent_org_tasks, agent_org_turn_contexts, agent_org_watchdog, }; -const RUNTIME_TABLES_V1: [&str; 13] = [ - "agent_org_runtime_runs", - "agent_org_runtime_run_progress", - "agent_org_runtime_member_materializations", - "agent_org_runtime_initial_inputs", - "agent_org_runtime_plan_approvals", - "agent_org_runtime_recovery_attempts", - "agent_org_runtime_tasks", - "agent_org_runtime_task_events", - "agent_org_runtime_task_schema_migrations", - "agent_org_runtime_inbox", - "agent_org_runtime_inbox_materializations", - "agent_org_runtime_inbox_delivery_resolutions", - "agent_org_runtime_member_interventions", -]; - const RUNTIME_TABLES: [&str; 15] = [ "agent_org_runtime_runs", "agent_org_runtime_run_progress", @@ -89,25 +73,19 @@ const DROP_LEGACY_SCHEMA: &str = "DROP TABLE IF EXISTS agent_inbox_materializati type SchemaManifest = BTreeMap<(String, String), (String, String)>; pub(super) fn initialize(conn: &Connection) -> SqliteResult<()> { - let expected_v1 = expected_manifest_v1()?; let expected = expected_manifest()?; let tx = database::db::begin_immediate(conn)?; let runtime_table_count = count_known_tables(&tx, &RUNTIME_TABLES)?; - let (fresh, upgraded_from_v1) = match runtime_table_count { - 0 => (true, false), - count if count == RUNTIME_TABLES_V1.len() => { - verify_manifest(&tx, &expected_v1)?; - (false, true) - } + let fresh = match runtime_table_count { + 0 => true, count if count == RUNTIME_TABLES.len() => { verify_manifest(&tx, &expected)?; - (false, false) + false } count => { return Err(schema_error(format!( - "partial Agent Org runtime schema: found {count}; expected 0, {} (PR828), or {} (PR3) canonical tables", - RUNTIME_TABLES_V1.len(), + "partial Agent Org runtime schema: found {count} of {} canonical tables; only an empty namespace or the complete current manifest is accepted", RUNTIME_TABLES.len(), ))) } @@ -119,8 +97,6 @@ pub(super) fn initialize(conn: &Connection) -> SqliteResult<()> { if fresh { create_runtime_schema(&tx)?; - } else if upgraded_from_v1 { - agent_org_turn_contexts::create_schema(&tx)?; } verify_manifest(&tx, &expected)?; agent_inbox::repair_dangling_materializations(&tx)?; @@ -144,25 +120,20 @@ pub(super) fn initialize(conn: &Connection) -> SqliteResult<()> { legacy_table_count, legacy_object_count, fresh, - upgraded_from_v1, - idempotent = !fresh && !upgraded_from_v1, + idempotent = !fresh, "initialized isolated Agent Org runtime schema" ); Ok(()) } fn create_runtime_schema(conn: &Connection) -> SqliteResult<()> { - create_runtime_schema_v1(conn)?; - agent_org_turn_contexts::create_schema(conn) -} - -fn create_runtime_schema_v1(conn: &Connection) -> SqliteResult<()> { agent_org_runs::create_schema(conn)?; agent_inbox::create_schema(conn)?; agent_org_tasks::create_schema(conn)?; agent_org_plan_approvals::create_schema(conn)?; agent_member_interventions::create_schema(conn)?; - agent_org_watchdog::create_schema(conn) + agent_org_watchdog::create_schema(conn)?; + agent_org_turn_contexts::create_schema(conn) } fn expected_manifest() -> SqliteResult { @@ -172,13 +143,6 @@ fn expected_manifest() -> SqliteResult { read_manifest(&expected) } -fn expected_manifest_v1() -> SqliteResult { - let expected = Connection::open_in_memory()?; - expected.execute_batch("PRAGMA foreign_keys=ON;")?; - create_runtime_schema_v1(&expected)?; - read_manifest(&expected) -} - fn verify_manifest(conn: &Connection, expected: &SchemaManifest) -> SqliteResult<()> { let actual = read_manifest(conn)?; if &actual == expected { @@ -632,33 +596,6 @@ mod tests { assert_eq!(snapshot, "{\"team\":\"A\"}"); } - #[test] - fn exact_pr828_manifest_is_extended_atomically_to_pr3() { - let conn = connection(); - create_runtime_schema_v1(&conn).expect("PR828 runtime fixture"); - conn.execute( - "INSERT INTO agent_org_runtime_runs ( - id, org_id, coordinator_agent_id, org_snapshot_json, entry_mode, - status, created_at, updated_at - ) VALUES ('run-v1', 'org-v1', 'agent-v1', '{}', - 'standalone_session', 'idle', 'now', 'now')", - [], - ) - .expect("seed PR828 runtime row"); - - initialize(&conn).expect("upgrade exact PR828 manifest"); - - verify_manifest(&conn, &expected_manifest().expect("PR3 manifest")) - .expect("canonical PR3 manifest"); - assert_eq!(count_known_tables(&conn, &RUNTIME_TABLES).unwrap(), 15); - assert_eq!(row_count(&conn, "agent_org_runtime_runs"), 1); - assert_eq!( - row_count(&conn, "agent_org_runtime_member_dispatch_allocators"), - 0 - ); - assert_eq!(row_count(&conn, "agent_org_runtime_turn_contexts"), 0); - } - #[test] fn partial_or_unknown_runtime_schema_fails_closed_before_legacy_cleanup() { for mutate in ["partial", "changed", "extra_index"] { @@ -667,9 +604,14 @@ mod tests { conn.execute_batch("CREATE TABLE agent_org_runs (sentinel TEXT); INSERT INTO agent_org_runs VALUES ('legacy');") .expect("legacy sentinel"); match mutate { - "partial" => conn - .execute_batch("DROP TABLE agent_org_runtime_initial_inputs;") - .expect("make partial schema"), + "partial" => { + conn.execute_batch( + "DROP TABLE agent_org_runtime_turn_contexts; + DROP TABLE agent_org_runtime_member_dispatch_allocators;", + ) + .expect("make partial schema"); + assert_eq!(count_known_tables(&conn, &RUNTIME_TABLES).unwrap(), 13); + } "changed" => { conn.execute_batch( "DROP TABLE agent_org_runtime_member_interventions;