Fix : agent queue resilience

backend/src/api/audit/briefing.rs

@@ -108,6 +108,7 @@ pub async fn start_briefing(
     };
 
     let discussion = Discussion {
+        awaiting_agent: false,
         id: discussion_id.clone(),
         project_id: Some(project.id.clone()),
         title,

backend/src/api/audit/full.rs

@@ -1185,6 +1185,7 @@ pub async fn full_audit(
             "Validation audit AI".to_string()
         };
         let discussion = Discussion {
+            awaiting_agent: false,
             id: discussion_id.clone(),
             project_id: Some(project_id.clone()),
             title: disc_title,

backend/src/api/disc_prompts.rs

@@ -595,6 +595,7 @@ mod tests {
 
     fn disc_with_messages(messages: Vec<DiscussionMessage>, language: &str) -> Discussion {
         Discussion {
+            awaiting_agent: false,
             id: "d-test".into(),
             project_id: None,
             title: "Test discussion".into(),

backend/src/api/disc_source.rs

@@ -94,6 +94,7 @@ pub async fn disc_create(
     let disc_id = Uuid::new_v4().to_string();
     let agent = req.agent.clone();
     let disc = Discussion {
+        awaiting_agent: false,
         id: disc_id.clone(),
         project_id: req.project_id.clone(),
         title: req.title.clone(),

backend/src/api/discussions/crud.rs

@@ -129,6 +129,7 @@ pub async fn create(
     let base_branch = req.base_branch;
 
     let discussion = Discussion {
+        awaiting_agent: false,
         id: Uuid::new_v4().to_string(),
         project_id: req.project_id,
         title: req.title,
@@ -350,6 +351,20 @@ pub async fn delete(
     State(state): State<AppState>,
     Path(id): Path<String>,
 ) -> Json<ApiResponse<()>> {
+    // Cancel a live agent on this disc BEFORE deleting it, or
+    // the running agent finishes writing into a row that no longer exists
+    // (FOREIGN KEY constraint failed) and keeps its semaphore permit. The
+    // token is disc-keyed in the registry (streaming.rs); its agent-task
+    // finally path saves the partial + releases the permit. Best-effort.
+    if let Ok(mut map) = state.cancel_registry.lock() {
+        if let Some(token) = map.remove(&id) {
+            token.cancel();
+            tracing::info!("delete discussion {}: cancelled live agent before delete", id);
+        }
+    } else {
+        tracing::warn!("delete discussion {}: cancel registry poisoned — deleting without cancel", id);
+    }
+
     // Fetch discussion to check for worktree before deleting
     let disc = state.db.with_conn({
         let did = id.clone();
@@ -463,3 +478,50 @@ pub async fn edit_last_user_message(
         Err(e) => Json(ApiResponse::err(format!("DB error: {}", e))),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::Arc;
+    use tokio::sync::RwLock;
+
+    async fn state_with_disc(disc_id: &str) -> AppState {
+        let db = Arc::new(crate::db::Database::open_in_memory().expect("in-memory DB"));
+        let disc_id = disc_id.to_string();
+        db.with_conn(move |conn| {
+            let now = chrono::Utc::now().to_rfc3339();
+            conn.execute(
+                "INSERT INTO discussions (id, title, created_at, updated_at, message_count, workspace_mode)
+                 VALUES (?1, 'T2 disc', ?2, ?2, 0, 'Direct')",
+                rusqlite::params![disc_id, now],
+            )?;
+            Ok(())
+        }).await.unwrap();
+        let cfg = Arc::new(RwLock::new(crate::core::config::default_config()));
+        AppState::new_defaults(cfg, db, crate::DEFAULT_MAX_CONCURRENT_AGENTS)
+    }
+
+    #[tokio::test]
+    async fn delete_discussion_cancels_a_live_agent_before_deleting() {
+        // Deleting a disc with a running agent must fire its
+        // disc-keyed token first, or the agent writes into a deleted row.
+        let state = state_with_disc("d-live").await;
+        let token = tokio_util::sync::CancellationToken::new();
+        state.cancel_registry.lock().unwrap().insert("d-live".into(), token.clone());
+
+        let resp = delete(State(state.clone()), Path("d-live".to_string())).await;
+        assert!(resp.0.success, "delete must succeed: {:?}", resp.0.error);
+        assert!(token.is_cancelled(), "the disc's agent token must be cancelled");
+        assert!(
+            state.cancel_registry.lock().unwrap().is_empty(),
+            "the cancelled token must be removed from the registry",
+        );
+    }
+
+    #[tokio::test]
+    async fn delete_discussion_without_a_live_agent_succeeds() {
+        let state = state_with_disc("d-idle").await;
+        let resp = delete(State(state), Path("d-idle".to_string())).await;
+        assert!(resp.0.success);
+    }
+}

backend/src/api/discussions/messaging.rs

@@ -195,6 +195,10 @@ pub async fn send_message(
         return Sse::new(stream);
     }
 
+    // The awaiting_agent marker is set INSIDE make_agent_stream,
+    // after its preflight early-returns (marking here left the flag
+    // stuck when a preflight failed, and a later boot appended a bogus
+    // interruption notice).
     make_agent_stream(state, id, target).await
 }
 
@@ -203,6 +207,9 @@ pub async fn run_agent(
     State(state): State<AppState>,
     Path(id): Path<String>,
 ) -> Sse<SseStream> {
+    // The awaiting_agent marker is set inside make_agent_stream (after
+    // preflights) so a forced run that fails preflight never leaves a stuck
+    // flag. See send_message above.
     make_agent_stream(state, id, None).await
 }
 
@@ -385,4 +392,116 @@ mod tests {
             .unwrap();
         assert_eq!(live, 0, "paused agent is not a live responder → Kronn would still answer");
     }
+
+    /// A run that dies in make_agent_stream's
+    /// preflight (here: Isolated disc whose worktree re-lock fails because
+    /// the project path isn't a git repo) must NOT leave awaiting_agent=1,
+    /// or the next boot reconcile appends a bogus interruption notice.
+    /// The marker is set after every preflight early-return, so this path
+    /// never touches it.
+    #[tokio::test]
+    async fn run_agent_preflight_failure_leaves_no_awaiting_marker() {
+        let disc = "d-relock-fail";
+        let state = make_state_with_disc(disc).await;
+        state
+            .db
+            .with_conn(move |conn| {
+                conn.execute(
+                    "UPDATE discussions SET workspace_mode = 'Isolated',
+                            workspace_path = NULL, worktree_branch = 'kronn/test-relock'
+                     WHERE id = ?1",
+                    rusqlite::params![disc],
+                )?;
+                Ok(())
+            })
+            .await
+            .unwrap();
+
+        let resp = run_agent(State(state.clone()), Path(disc.to_string())).await;
+        let body = sse_body_to_string(resp).await;
+        assert!(body.contains("error"), "re-lock preflight must fail, got: {body}");
+
+        let awaiting: i64 = state
+            .db
+            .with_conn(move |conn| {
+                Ok(conn.query_row(
+                    "SELECT awaiting_agent FROM discussions WHERE id = ?1",
+                    rusqlite::params![disc],
+                    |r| r.get(0),
+                )?)
+            })
+            .await
+            .unwrap();
+        assert_eq!(awaiting, 0, "failed preflight must not leave the disc marked as owed a run");
+    }
+
+    /// A BATCH child is pre-marked awaiting_agent=1 at enqueue
+    /// (create_batch_run) and its SSE stream has no consumer. A preflight
+    /// failure must behave like the agent-start-failed arm: persist the
+    /// error in the thread, clear the marker (no bogus boot notice) and
+    /// bump the batch counters (no run stuck at n-1/N).
+    #[tokio::test]
+    async fn batch_child_preflight_failure_persists_error_and_settles_the_batch() {
+        let disc = "d-batch-relock";
+        let state = make_state_with_disc(disc).await;
+        state
+            .db
+            .with_conn(move |conn| {
+                let now = chrono::Utc::now().to_rfc3339();
+                conn.execute(
+                    "INSERT INTO workflows (id, name, trigger_json, steps_json, created_at, updated_at)
+                     VALUES ('wf-r12', 'r12', '{}', '[]', ?1, ?1)",
+                    rusqlite::params![now],
+                )?;
+                conn.execute(
+                    "INSERT INTO workflow_runs (id, workflow_id, run_type, status, started_at, batch_total)
+                     VALUES ('run-r12', 'wf-r12', 'batch', 'Running', ?1, 1)",
+                    rusqlite::params![now],
+                )?;
+                conn.execute(
+                    "UPDATE discussions SET workflow_run_id = 'run-r12', awaiting_agent = 1,
+                            workspace_mode = 'Isolated', workspace_path = NULL,
+                            worktree_branch = 'kronn/test-relock'
+                     WHERE id = ?1",
+                    rusqlite::params![disc],
+                )?;
+                Ok(())
+            })
+            .await
+            .unwrap();
+
+        let resp = run_agent(State(state.clone()), Path(disc.to_string())).await;
+        let body = sse_body_to_string(resp).await;
+        assert!(body.contains("error"), "re-lock preflight must fail, got: {body}");
+
+        let (awaiting, batch_failed): (i64, i64) = state
+            .db
+            .with_conn(move |conn| {
+                let awaiting = conn.query_row(
+                    "SELECT awaiting_agent FROM discussions WHERE id = ?1",
+                    rusqlite::params![disc],
+                    |r| r.get(0),
+                )?;
+                let failed = conn.query_row(
+                    "SELECT batch_failed FROM workflow_runs WHERE id = 'run-r12'",
+                    [],
+                    |r| r.get(0),
+                )?;
+                Ok((awaiting, failed))
+            })
+            .await
+            .unwrap();
+        assert_eq!(awaiting, 0, "enqueue-time marker must be cleared on preflight failure");
+        assert_eq!(batch_failed, 1, "the child must count as failed so the batch can finish");
+
+        let msgs = state
+            .db
+            .with_conn(move |conn| crate::db::discussions::list_messages(conn, disc))
+            .await
+            .unwrap();
+        assert!(
+            msgs.iter().any(|m| m.role == MessageRole::System && m.content.starts_with("Erreur:")),
+            "the preflight error must be persisted in the thread (fire-and-forget child)"
+        );
+    }
 }