feat(agent-org): add idempotent task graph writers

src-tauri/crates/agent-core/src/core/coordination/agent_inbox/store_write.rs

@@ -10,6 +10,95 @@ use super::record::row_to_record;
 use super::{AgentInboxRecord, AgentInboxStore, InsertInboxParams};
 
 impl AgentInboxStore {
+    /// Resolve formal coordination rows that became impossible to execute
+    /// while their Member's exact Provider Turn was still running.
+    ///
+    /// Writer cancellation/reassignment deliberately does not stop an old
+    /// TaskExecution. Coordinator messages sent while that Turn is active are
+    /// therefore persisted but cannot be drained until the Member becomes
+    /// idle. If every Task owned by the Member is terminal or reassigned at
+    /// that boundary, a new formal wake has no Task authority to bind to. Keep
+    /// the original rows unread for audit, but record a lifecycle-owned
+    /// cancellation so they no longer cause an impossible wake loop or block
+    /// Team Quiescence forever.
+    ///
+    /// This is intentionally limited to the pre-UserDirectedWork formal
+    /// message classes. PR 9's user-directed Inbox rows have their own exact
+    /// source authority and must never be swept by this fallback.
+    pub(crate) fn resolve_obsolete_formal_rows_after_successful_member_turn(
+        org_run_id: &str,
+        member_id: &str,
+    ) -> Result<usize, String> {
+        const REASON: &str = "member_turn_finished_without_owned_formal_work";
+
+        let resolved = with_sessions_writer(|| -> Result<usize, 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 has_owned_open_work: bool = tx
+                .query_row(
+                    "SELECT EXISTS(
+                         SELECT 1 FROM agent_org_runtime_tasks
+                         WHERE org_run_id=?1 AND owner=?2
+                           AND status IN ('pending','in_progress')
+                     )",
+                    params![org_run_id, member_id],
+                    |row| row.get(0),
+                )
+                .map_err(|err| err.to_string())?;
+            if has_owned_open_work {
+                tx.commit().map_err(|err| err.to_string())?;
+                return Ok(0);
+            }
+
+            let now = chrono::Utc::now().to_rfc3339();
+            let resolved = tx
+                .execute(
+                    "INSERT OR IGNORE INTO agent_org_runtime_inbox_delivery_resolutions (
+                         inbox_id,org_run_id,resolution_kind,resolved_by_member_id,reason,
+                         replacement_inbox_id,replacement_task_id,created_at
+                     )
+                     SELECT inbox.id,inbox.org_run_id,'cancelled','system:lifecycle',
+                            ?3,NULL,NULL,?4
+                     FROM agent_org_runtime_inbox inbox
+                     WHERE inbox.org_run_id=?1
+                       AND inbox.recipient_member_id=?2
+                       AND inbox.read_at IS NULL
+                       AND inbox.payload_kind IN (
+                           'plain','task_assigned','plan_approval_response','shutdown_request'
+                       )
+                       AND NOT EXISTS (
+                           SELECT 1
+                           FROM agent_org_runtime_inbox_delivery_resolutions resolution
+                           WHERE resolution.inbox_id=inbox.id
+                       )",
+                    params![org_run_id, member_id, REASON, &now],
+                )
+                .map_err(|err| err.to_string())?;
+            tx.execute(
+                "DELETE FROM agent_org_runtime_inbox_materializations
+                 WHERE inbox_id IN (
+                     SELECT resolution.inbox_id
+                     FROM agent_org_runtime_inbox_delivery_resolutions resolution
+                     JOIN agent_org_runtime_inbox inbox ON inbox.id=resolution.inbox_id
+                     WHERE resolution.org_run_id=?1
+                       AND resolution.reason=?3
+                       AND inbox.recipient_member_id=?2
+                 )",
+                params![org_run_id, member_id, REASON],
+            )
+            .map_err(|err| err.to_string())?;
+            tx.commit().map_err(|err| err.to_string())?;
+            Ok(resolved)
+        })?;
+
+        if resolved > 0 {
+            crate::coordination::agent_org_run_events::notify_agent_org_run_changed(org_run_id);
+        }
+        Ok(resolved)
+    }
+
     /// Persist a message and return the inserted record. The caller is
     /// responsible for resolving display-name / broadcast targets to one or
     /// more concrete `AgentId`s before calling this — the store does not

src-tauri/crates/agent-core/src/core/coordination/agent_org_plan_approvals/artifact.rs

@@ -22,7 +22,7 @@ pub(super) struct OwnedPlanPath {
     file_name: String,
 }
 
-pub(super) struct StagedPlanArtifact {
+pub(crate) struct StagedPlanArtifact {
     owned: OwnedPlanPath,
     temp_path: PathBuf,
     target_path: PathBuf,
@@ -31,7 +31,7 @@ pub(super) struct StagedPlanArtifact {
 /// Plan artifacts are a derived filesystem projection of SQLite state. A
 /// dedicated lock preserves commit/install order without holding the global
 /// sessions writer across rename or directory fsync.
-pub(super) fn plan_artifact_install_lock() -> &'static parking_lot::Mutex<()> {
+pub(crate) fn plan_artifact_install_lock() -> &'static parking_lot::Mutex<()> {
     static LOCK: OnceLock<parking_lot::Mutex<()>> = OnceLock::new();
     LOCK.get_or_init(|| parking_lot::Mutex::new(()))
 }
@@ -254,7 +254,7 @@ pub(super) fn resolve_owned_plan_target(
     }
 }
 
-pub(super) fn stage_plan_artifact_with_connection(
+pub(crate) fn stage_plan_artifact_with_connection(
     conn: &Connection,
     source_session_id: &str,
     plan_path: &str,
@@ -355,7 +355,7 @@ fn stage_owned_plan_artifact(
 /// Install only the already-fsynced bytes. Callers invoke this after SQLite
 /// commits while holding the dedicated artifact lock so two revisions cannot
 /// install out of commit order and unrelated database writes are not blocked.
-pub(super) fn install_staged_plan_artifact(
+pub(crate) fn install_staged_plan_artifact(
     staged: Option<&StagedPlanArtifact>,
 ) -> Result<(), String> {
     let Some(staged) = staged else {

src-tauri/crates/agent-core/src/core/coordination/agent_org_plan_approvals/mod.rs

@@ -11,7 +11,7 @@ use serde::{Deserialize, Serialize};
 use crate::coordination::agent_org_tasks::TaskMutationOutcome;
 use crate::definitions::orgs::PlanApprovalPolicy;
 
-mod artifact;
+pub(crate) mod artifact;
 mod persistence;
 mod store;
 mod transitions;