fix(tasks): tighten worker lookup and branch-cancel parsing

Author: vsumner

src/agent/channel.rs

@@ -98,6 +98,7 @@ const CHANGE_COMPARISON_VERBS: &[&str] = &[
     "proceed with ",
 ];
 const BRANCH_CANCELLED_PREFIX: &str = "Branch cancelled:";
+const BRANCH_CANCELLED_SENTENCE: &str = "Branch cancelled.";
 
 async fn recv_channel_event(
     event_rx: &mut broadcast::Receiver<ProcessEvent>,
@@ -196,7 +197,7 @@ fn truncate_working_memory_summary(summary: &str) -> String {
 fn branch_working_memory_event_summary(
     conclusion: &str,
 ) -> (crate::memory::WorkingMemoryEventType, String) {
-    if let Some(reason) = conclusion.strip_prefix(BRANCH_CANCELLED_PREFIX) {
+    if let Some(reason) = parse_branch_cancellation_reason(conclusion) {
         let reason = truncate_working_memory_summary(reason.trim());
         let summary = if reason.is_empty() {
             "Branch cancelled".to_string()
@@ -217,6 +218,17 @@ fn branch_working_memory_event_summary(
     )
 }
 
+fn parse_branch_cancellation_reason(conclusion: &str) -> Option<&str> {
+    let trimmed = conclusion.trim();
+    if let Some(rest) = trimmed.strip_prefix(BRANCH_CANCELLED_PREFIX) {
+        return Some(rest);
+    }
+    if let Some(rest) = trimmed.strip_prefix(BRANCH_CANCELLED_SENTENCE) {
+        return Some(rest);
+    }
+    None
+}
+
 fn sentence_contains_decision_marker(sentence: &str) -> bool {
     let sentence_lower = sentence.to_ascii_lowercase();
     DECISION_MARKERS
@@ -530,9 +542,9 @@ impl ChannelState {
 
         let reason = crate::summarize_first_non_empty_line(reason, crate::EVENT_SUMMARY_MAX_CHARS);
         let conclusion = if reason.is_empty() {
-            "Branch cancelled.".to_string()
+            BRANCH_CANCELLED_SENTENCE.to_string()
         } else {
-            format!("Branch cancelled: {reason}")
+            format!("{BRANCH_CANCELLED_PREFIX} {reason}")
         };
         self.process_run_logger
             .log_branch_completed(branch_id, &conclusion);
@@ -3299,7 +3311,7 @@ impl Channel {
                     // Regular branch: accumulate result for the next retrigger.
                     // The result text will be embedded directly in the retrigger
                     // message so the LLM knows exactly which process produced it.
-                    let branch_success = !conclusion.starts_with(BRANCH_CANCELLED_PREFIX);
+                    let branch_success = parse_branch_cancellation_reason(conclusion).is_none();
                     self.pending_results.push(PendingResult {
                         process_type: "branch",
                         process_id: branch_id.to_string(),
@@ -4326,6 +4338,14 @@ mod tests {
         assert_eq!(summary, "Branch cancelled: superseded by user request");
     }
 
+    #[test]
+    fn branch_working_memory_event_records_sentence_cancellation_as_error() {
+        let (event_type, summary) = branch_working_memory_event_summary("Branch cancelled.");
+
+        assert_eq!(event_type, WorkingMemoryEventType::Error);
+        assert_eq!(summary, "Branch cancelled");
+    }
+
     #[test]
     fn quiet_mode_invocation_uses_discord_mention_and_reply_metadata() {
         let message = inbound_message(

src/tasks/store.rs

@@ -428,32 +428,36 @@ impl TaskStore {
             .await
             .context("failed to open worker task update transaction")?;
 
-        let row = sqlx::query(&format!(
-            "{SELECT_COLUMNS} FROM tasks WHERE worker_id = ? ORDER BY updated_at DESC LIMIT 1"
+        let exact_row = sqlx::query(&format!(
+            "{SELECT_COLUMNS} FROM tasks WHERE worker_id = ? AND task_number = ?"
         ))
         .bind(worker_id)
+        .bind(task_number)
         .fetch_optional(&mut *tx)
         .await
-        .context("failed to fetch task by worker id for update")?;
+        .context("failed to fetch worker task by id and number for update")?;
+
+        let Some(row) = exact_row else {
+            let assigned_task_number = sqlx::query_scalar::<_, i64>(
+                "SELECT task_number FROM tasks WHERE worker_id = ? ORDER BY task_number DESC LIMIT 1",
+            )
+            .bind(worker_id)
+            .fetch_optional(&mut *tx)
+            .await
+            .context("failed to fetch any task by worker id for update")?;
 
-        let Some(row) = row else {
             tx.commit()
                 .await
-                .context("failed to commit empty worker task update transaction")?;
+                .context("failed to commit unmatched worker task update transaction")?;
+            if let Some(assigned_task_number) = assigned_task_number {
+                return Ok(WorkerTaskUpdateResult::WrongTask {
+                    assigned_task_number,
+                });
+            }
             return Ok(WorkerTaskUpdateResult::NotAssigned);
         };
 
         let current = task_from_row(row)?;
-        if current.task_number != task_number {
-            let assigned_task_number = current.task_number;
-            tx.commit()
-                .await
-                .context("failed to commit rejected worker task update transaction")?;
-            return Ok(WorkerTaskUpdateResult::WrongTask {
-                assigned_task_number,
-            });
-        }
-
         let previous_status = current.status;
         let task = Self::update_current_in_tx(&mut tx, task_number, current, input).await?;
 
@@ -928,6 +932,59 @@ mod tests {
         assert_eq!(result.task.status, TaskStatus::Done);
     }
 
+    #[tokio::test]
+    async fn update_worker_task_prefers_exact_match_with_duplicate_worker_bindings() {
+        let store = setup_store().await;
+        let first = store
+            .create(self_assigned_input("first task", TaskStatus::InProgress))
+            .await
+            .expect("first task should be created");
+        let second = store
+            .create(self_assigned_input("second task", TaskStatus::InProgress))
+            .await
+            .expect("second task should be created");
+
+        let shared_worker_id = "worker-shared";
+        store
+            .update(
+                first.task_number,
+                UpdateTaskInput {
+                    worker_id: Some(shared_worker_id.to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("first worker binding should update");
+        store
+            .update(
+                second.task_number,
+                UpdateTaskInput {
+                    worker_id: Some(shared_worker_id.to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("second worker binding should update");
+
+        let result = store
+            .update_worker_task(
+                shared_worker_id,
+                first.task_number,
+                UpdateTaskInput {
+                    metadata: Some(serde_json::json!({"target": "first"})),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("worker-scoped update should succeed");
+
+        let WorkerTaskUpdateResult::Updated(result) = result else {
+            panic!("expected exact task update despite duplicate worker bindings");
+        };
+        assert_eq!(result.task.task_number, first.task_number);
+        assert_eq!(result.task.metadata["target"], "first");
+    }
+
     #[tokio::test]
     async fn can_requeue_in_progress_and_clear_worker_binding() {
         let store = setup_store().await;