Merge feat/r9-typed-slot-errors-20260829: R9 typed slot errors — honest fan-out statuses

All-failed fan-outs (task, budget, or script kinds) record Failed;
partially-failed fan-outs record Degraded via the new per-slot
FanoutSlotDropped event — no more silently lost work behind a clean
Completed. Partial-mode __taskError wire change disclosed in docs.

Verified on the merged tree: full tui lib suite 11,528 passed / 0 failed;
vm_tests 74/74; workflow lib 260/260; tools::workflow 103/103; fmt and
CI-exact clippy clean.

Signed-off-by: CodeWhale Bot <[email protected]>

crates/tui/src/tools/workflow/mod.rs

@@ -805,6 +805,141 @@ enum WorkflowRunStatus {
     Cancelled,
 }
 
+/// The per-slot outcome ledger a finished run is classified against.
+///
+/// A workflow script can return a perfectly good-looking value while every
+/// task it fanned out died — `parallel()`'s settled default resolves a failed
+/// slot to `null`, so the script never sees a throw. Terminal status is
+/// therefore decided from what actually ran, not from the script's return.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+struct SlotLedger {
+    /// Task records the driver holds for this run (children who were admitted).
+    tasks: usize,
+    /// How many of those records ended in a failed terminal state — a
+    /// subagent failure, a budget-exhausted stop, or a replay divergence.
+    /// A budget-dead child produced no result either; counting only plain
+    /// `Failed` let an all-budget fan-out classify as a clean completion.
+    failed_tasks: usize,
+    /// Requested slots refused before a child existed — driver admission
+    /// refusals plus `ProgressEvent::TaskRejected`.
+    rejected: u64,
+    /// Fan-outs (`parallel()`/`pipeline()`) that resolved with every slot
+    /// failed and nothing surviving, script throws included. Those leave no
+    /// task record at all, so only the structured
+    /// `ProgressEvent::FanoutAllSlotsFailed` count makes them visible here.
+    dead_fanouts: u32,
+    /// Individual slots a settled fan-out dropped to null while other slots
+    /// survived (`ProgressEvent::FanoutSlotDropped`). A partial loss with
+    /// surviving work: Degraded, never a clean Completed.
+    dropped_slots: u32,
+    /// Whether any child agent was ever admitted for this run.
+    any_child_ran: bool,
+    /// First retained dispatch-failure message, for the all-rejected receipt.
+    retained_detail: Option<String>,
+}
+
+impl SlotLedger {
+    /// The terminal status a `Completed` script run should actually record,
+    /// with the receipt line, or `None` when every slot came back clean.
+    fn classify(&self) -> Option<(WorkflowRunStatus, String)> {
+        let dead_fanouts = if self.dead_fanouts > 0 {
+            format!(
+                " and {} fan-out(s) lost every slot (no work survived them)",
+                self.dead_fanouts
+            )
+        } else {
+            String::new()
+        };
+        if !self.any_child_ran && self.rejected > 0 {
+            // #5035: every dispatch was rejected before a child ran.
+            let retained = self
+                .retained_detail
+                .as_ref()
+                .map(|message| format!("; retained detail: {message}"))
+                .unwrap_or_default();
+            return Some((
+                WorkflowRunStatus::Failed,
+                format!(
+                    "no child agents ran: all {} task dispatch(es) were rejected{retained}{dead_fanouts}",
+                    self.rejected
+                ),
+            ));
+        }
+        // R9: a run whose every task failed produced nothing. Reporting that
+        // as a partial success let a workflow that lost all its work read as
+        // "completed with caveats"; it is a failure with a preserved output.
+        // `failed_tasks` counts every failure-ish terminal state, so a
+        // fan-out that died entirely of budget exhaustion lands here too.
+        if self.tasks > 0 && self.failed_tasks == self.tasks {
+            let rejected = if self.rejected > 0 {
+                format!(" and {} dispatch(es) were rejected", self.rejected)
+            } else {
+                String::new()
+            };
+            return Some((
+                WorkflowRunStatus::Failed,
+                format!(
+                    "no task produced a result: all {} task(s) failed{rejected}{dead_fanouts}; \
+                     the recorded result reflects no completed work",
+                    self.tasks
+                ),
+            ));
+        }
+        // R9: a fan-out can die without a single task record — thunks that
+        // throw before (or instead of) calling `task()` resolve to null slots
+        // and leave only the dead-fan-out count behind. When no child
+        // anywhere survived, that run also produced nothing.
+        if self.dead_fanouts > 0 && self.tasks == 0 {
+            let rejected = if self.rejected > 0 {
+                format!(" and {} dispatch(es) were rejected", self.rejected)
+            } else {
+                String::new()
+            };
+            return Some((
+                WorkflowRunStatus::Failed,
+                format!(
+                    "no work survived: {} fan-out(s) lost every slot{rejected}; \
+                     the recorded result reflects no completed work",
+                    self.dead_fanouts
+                ),
+            ));
+        }
+        if self.failed_tasks > 0
+            || self.rejected > 0
+            || self.dead_fanouts > 0
+            || self.dropped_slots > 0
+        {
+            let mut parts = Vec::new();
+            if self.failed_tasks > 0 {
+                parts.push(format!(
+                    "{} of {} task(s) failed",
+                    self.failed_tasks, self.tasks
+                ));
+            }
+            if self.rejected > 0 {
+                parts.push(format!("{} dispatch(es) were rejected", self.rejected));
+            }
+            if self.dead_fanouts > 0 {
+                parts.push(format!("{} fan-out(s) lost every slot", self.dead_fanouts));
+            }
+            if self.dropped_slots > 0 {
+                parts.push(format!(
+                    "{} fan-out slot(s) dropped while others survived",
+                    self.dropped_slots
+                ));
+            }
+            return Some((
+                WorkflowRunStatus::Degraded,
+                format!(
+                    "completed with dropped slots: {}; the recorded result may be partial",
+                    parts.join(" and ")
+                ),
+            ));
+        }
+        None
+    }
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum WorkflowAction {
     Start,
@@ -1774,41 +1909,29 @@ async fn run_workflow_vm(
             // ledger — every requested task either became a child with a
             // terminal record or landed in `dispatch_failures` (driver
             // rejections and, via `ProgressEvent::TaskRejected`, VM-level
-            // rejections that previously vanished into null slots).
+            // rejections that previously vanished into null slots), and a
+            // fan-out whose every slot failed — script throws included —
+            // arrives via the dead-fan-out counter.
             if status == WorkflowRunStatus::Completed {
                 let task_records = driver.task_records_snapshot();
-                let failed_children = task_records
-                    .iter()
-                    .filter(|task| task.status == IrWorkflowRunStatus::Failed)
-                    .count();
-                let rejected = record.dispatch_failure_count;
-                if record.child_ids.is_empty() && rejected > 0 {
-                    // #5035: every dispatch was rejected before a child ran.
-                    status = WorkflowRunStatus::Failed;
-                    let retained_detail = record
+                let ledger = SlotLedger {
+                    tasks: task_records.len(),
+                    failed_tasks: task_records
+                        .iter()
+                        .filter(|task| task.failed_for_ledger())
+                        .count(),
+                    rejected: record.dispatch_failure_count,
+                    dead_fanouts: driver.dead_fanout_count(),
+                    dropped_slots: driver.dropped_slot_count(),
+                    any_child_ran: !record.child_ids.is_empty(),
+                    retained_detail: record
                         .dispatch_failures
                         .first()
-                        .map(|failure| format!("; retained detail: {}", failure.message))
-                        .unwrap_or_default();
-                    error = Some(format!(
-                        "no child agents ran: all {rejected} task dispatch(es) were rejected{retained_detail}"
-                    ));
-                } else if failed_children > 0 || rejected > 0 {
-                    status = WorkflowRunStatus::Degraded;
-                    let mut parts = Vec::new();
-                    if failed_children > 0 {
-                        parts.push(format!(
-                            "{failed_children} of {} task(s) failed",
-                            task_records.len()
-                        ));
-                    }
-                    if rejected > 0 {
-                        parts.push(format!("{rejected} dispatch(es) were rejected"));
-                    }
-                    error = Some(format!(
-                        "completed with dropped slots: {}; the recorded result may be partial",
-                        parts.join(" and ")
-                    ));
+                        .map(|failure| failure.message.clone()),
+                };
+                if let Some((slot_status, slot_error)) = ledger.classify() {
+                    status = slot_status;
+                    error = Some(slot_error);
                 }
             }
             record.status = status;
@@ -3258,6 +3381,26 @@ struct RuntimeTaskRecord {
     usage: Option<WorkflowTaskUsage>,
 }
 
+impl RuntimeTaskRecord {
+    /// Whether this record ended in a terminal state that produced no usable
+    /// result — the states the slot ledger counts as a failed task.
+    ///
+    /// `BudgetExceeded` is the named gap this exists to close: a child that
+    /// dies of budget exhaustion is a failed task for the all-failed rule,
+    /// not an invisible one. `ReplayDiverged` means the leaf's replay did
+    /// not reproduce its recorded result — no output either. `Cancelled` is
+    /// deliberately excluded: it is the run's own stop, not lost work, and
+    /// run-level cancellation is finalized before this ledger is consulted.
+    fn failed_for_ledger(&self) -> bool {
+        matches!(
+            self.status,
+            IrWorkflowRunStatus::Failed
+                | IrWorkflowRunStatus::BudgetExceeded
+                | IrWorkflowRunStatus::ReplayDiverged
+        )
+    }
+}
+
 struct SubAgentWorkflowDriver {
     run_id: String,
     owner_session_id: String,
@@ -3273,6 +3416,11 @@ struct SubAgentWorkflowDriver {
     /// an explicit `phase` option).
     current_phase: Mutex<Option<String>>,
     task_records: Arc<Mutex<HashMap<String, RuntimeTaskRecord>>>,
+    /// Fan-outs that resolved with every slot failed (`ProgressEvent::FanoutAllSlotsFailed`),
+    /// script throws included. Consulted by the terminal slot ledger: those
+    /// fan-outs can leave no task record at all.
+    dead_fanouts: AtomicU32,
+    dropped_slots: AtomicU32,
     total_budget: Option<u64>,
     last_budget_event: Arc<Mutex<Option<BudgetSnapshot>>>,
     /// Workflow-owned gates installed for this run (#4179).
@@ -3323,6 +3471,8 @@ impl SubAgentWorkflowDriver {
             child_counter: AtomicU32::new(0),
             current_phase: Mutex::new(None),
             task_records: Arc::new(Mutex::new(HashMap::new())),
+            dead_fanouts: AtomicU32::new(0),
+            dropped_slots: AtomicU32::new(0),
             total_budget,
             last_budget_event: Arc::new(Mutex::new(None)),
             gate_specs: Arc::new(gate_specs),
@@ -3885,6 +4035,28 @@ impl SubAgentWorkflowDriver {
             .unwrap_or_default()
     }
 
+    /// Count a fan-out that resolved with every slot failed. The prelude's
+    /// run-log breadcrumb already names it for operators; this counter is
+    /// what lets the terminal slot ledger act on it.
+    fn record_dead_fanout(&self) {
+        self.dead_fanouts.fetch_add(1, Ordering::SeqCst);
+    }
+
+    fn dead_fanout_count(&self) -> u32 {
+        self.dead_fanouts.load(Ordering::SeqCst)
+    }
+
+    /// Count one slot a settled fan-out dropped while others survived; the
+    /// per-slot breadcrumb already narrates it, this counter makes the loss
+    /// visible to the terminal slot ledger.
+    fn record_dropped_slot(&self) {
+        self.dropped_slots.fetch_add(1, Ordering::SeqCst);
+    }
+
+    fn dropped_slot_count(&self) -> u32 {
+        self.dropped_slots.load(Ordering::SeqCst)
+    }
+
     fn add_waiter_or_complete(&self, agent_id: String, waiter: oneshot::Sender<TaskCompletion>) {
         let mut state = self
             .completion_state
@@ -4149,6 +4321,18 @@ impl WorkflowDriver for SubAgentWorkflowDriver {
                 self.record_dispatch_failure(label, phase, message);
                 return;
             }
+            // R9: a dead fan-out is a status signal, not a narrated line —
+            // the prelude already logged the operator breadcrumb. Counted on
+            // the driver so the terminal slot ledger can refuse to record a
+            // run where nothing survived as a plain success.
+            ProgressEvent::FanoutAllSlotsFailed { .. } => {
+                self.record_dead_fanout();
+                return;
+            }
+            ProgressEvent::FanoutSlotDropped { .. } => {
+                self.record_dropped_slot();
+                return;
+            }
             ProgressEvent::Log { message } => (
                 format!("log: {message}"),
                 WorkflowUiEvent::new(&self.owner_session_id, WorkflowUiEventKind::Log { message }),
@@ -5187,6 +5371,224 @@ mod tests {
     use codewhale_workflow::{IsolationMode, leaf_is_write_capable};
     use std::sync::atomic::{AtomicUsize, Ordering};
 
+    #[test]
+    fn a_clean_slot_ledger_leaves_the_run_completed() {
+        let ledger = SlotLedger {
+            tasks: 3,
+            failed_tasks: 0,
+            rejected: 0,
+            dead_fanouts: 0,
+            dropped_slots: 0,
+            any_child_ran: true,
+            retained_detail: None,
+        };
+        assert_eq!(ledger.classify(), None);
+    }
+
+    #[test]
+    fn a_run_with_no_tasks_at_all_stays_completed() {
+        // A script that orchestrates nothing is not a dropped-slot failure.
+        assert_eq!(SlotLedger::default().classify(), None);
+    }
+
+    #[test]
+    fn some_failed_slots_degrade_the_run_without_failing_it() {
+        let (status, message) = SlotLedger {
+            tasks: 3,
+            failed_tasks: 1,
+            rejected: 0,
+            dead_fanouts: 0,
+            dropped_slots: 0,
+            any_child_ran: true,
+            retained_detail: None,
+        }
+        .classify()
+        .expect("a dropped slot must be classified");
+        assert_eq!(status, WorkflowRunStatus::Degraded);
+        assert!(message.contains("1 of 3 task(s) failed"), "{message}");
+    }
+
+    #[test]
+    fn a_run_whose_every_task_failed_cannot_report_success() {
+        // R9: `parallel()`'s settled default resolves each dead slot to
+        // `null`, so the script returns cleanly. The run must not.
+        let (status, message) = SlotLedger {
+            tasks: 4,
+            failed_tasks: 4,
+            rejected: 0,
+            dead_fanouts: 0,
+            dropped_slots: 0,
+            any_child_ran: true,
+            retained_detail: None,
+        }
+        .classify()
+        .expect("a fully-failed fan-out must be classified");
+        assert_eq!(status, WorkflowRunStatus::Failed);
+        assert!(
+            message.contains("no task produced a result: all 4 task(s) failed"),
+            "{message}"
+        );
+        assert_ne!(
+            owner_state_for_run_status(status),
+            OwnerState::Completed,
+            "a run that lost every task must never project as completed"
+        );
+    }
+
+    #[test]
+    fn a_fully_failed_fan_out_also_names_its_rejected_dispatches() {
+        let (status, message) = SlotLedger {
+            tasks: 2,
+            failed_tasks: 2,
+            rejected: 1,
+            dead_fanouts: 0,
+            dropped_slots: 0,
+            any_child_ran: true,
+            retained_detail: Some("admission cap".to_string()),
+        }
+        .classify()
+        .expect("a fully-failed fan-out must be classified");
+        assert_eq!(status, WorkflowRunStatus::Failed);
+        assert!(
+            message.contains("all 2 task(s) failed")
+                && message.contains("1 dispatch(es) were rejected"),
+            "{message}"
+        );
+    }
+
+    #[test]
+    fn every_dispatch_rejected_still_fails_before_any_child_ran() {
+        let (status, message) = SlotLedger {
+            tasks: 0,
+            failed_tasks: 0,
+            rejected: 2,
+            dead_fanouts: 0,
+            dropped_slots: 0,
+            any_child_ran: false,
+            retained_detail: Some("depth ceiling".to_string()),
+        }
+        .classify()
+        .expect("an all-rejected run must be classified");
+        assert_eq!(status, WorkflowRunStatus::Failed);
+        assert!(
+            message.contains("no child agents ran") && message.contains("depth ceiling"),
+            "{message}"
+        );
+    }
+
+    #[test]
+    fn budget_exhausted_children_count_as_failed_for_the_all_failed_rule() {
+        // R9 blocker: a fan-out whose every child died of budget exhaustion
+        // used to classify as a plain completion because `failed_tasks`
+        // counted only `Failed` records. The records below are built through
+        // the real completion mapping, not hand-set counters.
+        let completions = [
+            TaskCompletion::BudgetExhausted {
+                message: "shared pool drained".to_string(),
+            },
+            TaskCompletion::BudgetExhausted {
+                message: "shared pool drained".to_string(),
+            },
+        ];
+        let task_records: Vec<RuntimeTaskRecord> = completions
+            .iter()
+            .enumerate()
+            .map(|(index, completion)| {
+                let (status, output) = task_completion_status(completion);
+                RuntimeTaskRecord {
+                    agent_id: format!("agent_{index}"),
+                    label: None,
+                    role: None,
+                    status,
+                    output,
+                    schema_error: None,
+                    usage: None,
+                }
+            })
+            .collect();
+        let ledger = SlotLedger {
+            tasks: task_records.len(),
+            failed_tasks: task_records
+                .iter()
+                .filter(|task| task.failed_for_ledger())
+                .count(),
+            rejected: 0,
+            // The fan-out those children died in also reported itself dead.
+            dead_fanouts: 1,
+            dropped_slots: 0,
+            any_child_ran: true,
+            retained_detail: None,
+        };
+        assert_eq!(task_records[0].status, IrWorkflowRunStatus::BudgetExceeded);
+        let (status, message) = ledger
+            .classify()
+            .expect("an all-budget-exhausted run must be classified");
+        assert_eq!(status, WorkflowRunStatus::Failed);
+        assert!(
+            message.contains("all 2 task(s) failed")
+                && message.contains("1 fan-out(s) lost every slot"),
+            "{message}"
+        );
+        assert_ne!(
+            owner_state_for_run_status(status),
+            OwnerState::Completed,
+            "a run that lost every child to budget exhaustion must never project as completed"
+        );
+    }
+
+    #[test]
+    fn a_fan_out_of_script_throws_with_no_task_records_fails_the_run() {
+        // R9 blocker: thunks that throw without calling `task()` leave no
+        // task record and no dispatch failure — only the structured
+        // every-slot-failed count. That run produced nothing.
+        let (status, message) = SlotLedger {
+            tasks: 0,
+            failed_tasks: 0,
+            rejected: 0,
+            dead_fanouts: 1,
+            dropped_slots: 0,
+            any_child_ran: false,
+            retained_detail: None,
+        }
+        .classify()
+        .expect("a dead script-throw fan-out must be classified");
+        assert_eq!(status, WorkflowRunStatus::Failed);
+        assert!(
+            message.contains("no work survived") && message.contains("1 fan-out(s)"),
+            "{message}"
+        );
+    }
+
+    #[test]
+    fn a_dead_fan_out_beside_surviving_work_degrades_instead_of_failing() {
+        // Some work survived elsewhere in the run: keep the output, refuse
+        // the plain-success label, but do not claim nothing completed.
+        let (status, message) = SlotLedger {
+            tasks: 1,
+            failed_tasks: 0,
+            rejected: 0,
+            dead_fanouts: 1,
+            dropped_slots: 0,
+            any_child_ran: true,
+            retained_detail: None,
+        }
+        .classify()
+        .expect("a dead fan-out beside surviving work must be classified");
+        assert_eq!(status, WorkflowRunStatus::Degraded);
+        assert!(
+            message.contains("1 fan-out(s) lost every slot")
+                && message.contains("result may be partial"),
+            "{message}"
+        );
+    }
+
+    #[test]
+    fn an_empty_fan_out_is_not_a_dead_fan_out() {
+        // `parallel([])` declares no slots; a run that orchestrates nothing
+        // stays completed (mirrors `a_run_with_no_tasks_at_all_stays_completed`).
+        assert_eq!(SlotLedger::default().classify(), None);
+    }
+
     #[test]
     fn settled_runs_leave_a_report_artifact_under_codewhale_reports() {
         let tmp = tempfile::tempdir().expect("tempdir");
@@ -8062,6 +8464,182 @@ reviewer = "reviewer"
         );
     }
 
+    #[tokio::test]
+    #[allow(clippy::await_holding_lock)]
+    async fn a_fan_out_where_every_child_died_of_budget_exhaustion_fails_the_run() {
+        // R9 blocker: budget-exhausted children map to `BudgetExceeded` task
+        // records, and the ledger used to count only plain `Failed` records —
+        // so a fan-out that lost every child to the token ceiling read as a
+        // clean completion. This drives the real path end to end: per-task
+        // `tokenBudget` forks an isolated pool, the fake provider reports more
+        // tokens than the cap, the child terminalizes `BudgetExhausted`, and
+        // the completion pump delivers it as a `BudgetExceeded` record.
+        let _retry_guard = workflow_test_retry_guard();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let ctx = ToolContext::new(tmp.path().to_path_buf());
+        let manager = new_shared_subagent_manager(tmp.path().to_path_buf(), 2);
+        let (client, calls) = fake_chat_client("child done").await;
+        let runtime = SubAgentRuntime::new(
+            client,
+            "deepseek-v4-flash".to_string(),
+            ctx.clone(),
+            true,
+            None,
+            manager,
+        );
+        let tool = WorkflowTool::new(runtime.manager.clone(), runtime);
+
+        let result = tool
+            .execute(
+                json!({
+                    "action": "run",
+                    "script": "return await parallel([() => task({ description: 'spendy a', type: 'explore', allowedTools: [], tokenBudget: 1 }), () => task({ description: 'spendy b', type: 'explore', allowedTools: [], tokenBudget: 1 })]);"
+                }),
+                &ctx,
+            )
+            .await
+            .expect("all-budget fan-out still returns the run record");
+        let payload: Value = serde_json::from_str(&result.content).expect("json result");
+
+        assert_eq!(payload["status"], "failed", "{payload}");
+        let error = payload["error"].as_str().expect("error surfaced");
+        assert!(
+            error.contains("all 2 task(s) failed")
+                && error.contains("1 fan-out(s) lost every slot"),
+            "error should name the budget-dead children and the dead fan-out: {error}"
+        );
+        // The output is preserved — a failure with a receipt, not an erasure.
+        let slots = payload["result"].as_array().expect("run kept its output");
+        assert_eq!(slots.len(), 2, "{payload}");
+        assert!(slots.iter().all(|slot| slot.is_null()), "{slots:?}");
+        assert_eq!(payload["child_ids"].as_array().unwrap().len(), 2);
+        assert!(
+            calls.load(Ordering::SeqCst) >= 2,
+            "both children should have run and reported usage"
+        );
+    }
+
+    #[tokio::test]
+    #[allow(clippy::await_holding_lock)]
+    async fn a_fan_out_of_script_throws_cannot_report_completed() {
+        // R9 blocker: thunks that throw without calling `task()` produce
+        // kind `script`, drop to null, and previously left no host-visible
+        // trace beyond a log line — no task records, no dispatch failures,
+        // classify() = None, run recorded Completed. The structured
+        // every-slot-failed signal must carry the dead fan-out to the host.
+        let _retry_guard = workflow_test_retry_guard();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let ctx = ToolContext::new(tmp.path().to_path_buf());
+        let manager = new_shared_subagent_manager(tmp.path().to_path_buf(), 2);
+        let (client, calls) = fake_chat_client("unused").await;
+        let runtime = SubAgentRuntime::new(
+            client,
+            "deepseek-v4-flash".to_string(),
+            ctx.clone(),
+            true,
+            None,
+            manager,
+        );
+        let tool = WorkflowTool::new(runtime.manager.clone(), runtime);
+
+        let result = tool
+            .execute(
+                json!({
+                    "action": "run",
+                    "script": r#"
+                    const results = await parallel([
+                        () => { throw new Error("script slot a died"); },
+                        () => { throw new Error("script slot b died"); },
+                    ]);
+                    return { slots: results };
+                    "#
+                }),
+                &ctx,
+            )
+            .await
+            .expect("all-script-throw fan-out still returns the run record");
+        let payload: Value = serde_json::from_str(&result.content).expect("json result");
+
+        assert_eq!(payload["status"], "failed", "{payload}");
+        let error = payload["error"].as_str().expect("error surfaced");
+        assert!(
+            error.contains("no work survived") && error.contains("1 fan-out(s) lost every slot"),
+            "error should name the dead fan-out: {error}"
+        );
+        // The null slots survive on the record; the run refuses the success.
+        assert_eq!(
+            payload["result"]["slots"],
+            json!([null, null]),
+            "the settled array is preserved verbatim: {}",
+            payload["result"]
+        );
+        let progress = payload["progress"]
+            .as_array()
+            .expect("progress array")
+            .iter()
+            .filter_map(|value| value.as_str())
+            .collect::<Vec<_>>()
+            .join("\n");
+        assert!(
+            progress.contains("every slot failed (2 of 2)")
+                && progress.contains("no work survived this fan-out"),
+            "the operator breadcrumb must stay in the run log:\n{progress}"
+        );
+        assert_eq!(
+            calls.load(Ordering::SeqCst),
+            0,
+            "no provider call should be spent on thunks that never call task()"
+        );
+    }
+
+    #[tokio::test]
+    #[allow(clippy::await_holding_lock)]
+    async fn a_mixed_fan_out_with_surviving_work_stays_degraded_not_failed() {
+        // One slot runs a real child and survives; one thunk throws. Some
+        // work survived, so the run degrades with its null preserved — it is
+        // not a dead fan-out (1 of 2 failed) and must not record failed.
+        let _retry_guard = workflow_test_retry_guard();
+        let tmp = tempfile::tempdir().expect("tempdir");
+        let ctx = ToolContext::new(tmp.path().to_path_buf());
+        let manager = new_shared_subagent_manager(tmp.path().to_path_buf(), 2);
+        let (client, calls) = fake_chat_client("child done").await;
+        let runtime = SubAgentRuntime::new(
+            client,
+            "deepseek-v4-flash".to_string(),
+            ctx.clone(),
+            true,
+            None,
+            manager,
+        );
+        let tool = WorkflowTool::new(runtime.manager.clone(), runtime);
+
+        let result = tool
+            .execute(
+                json!({
+                    "action": "run",
+                    "script": "return await parallel([() => task({ description: 'healthy slot', type: 'explore', allowedTools: [] }), () => { throw new Error('script slot died'); }]);"
+                }),
+                &ctx,
+            )
+            .await
+            .expect("mixed fan-out still returns the run record");
+        let payload: Value = serde_json::from_str(&result.content).expect("json result");
+
+        assert_eq!(payload["status"], "degraded", "{payload}");
+        let error = payload["error"].as_str().expect("degradation surfaced");
+        assert!(
+            error.contains("result may be partial"),
+            "error should keep the partial-output caveat: {error}"
+        );
+        let slots = payload["result"].as_array().expect("run kept its output");
+        assert_eq!(slots.len(), 2, "{payload}");
+        assert!(slots[0].is_string() && slots[1].is_null(), "{slots:?}");
+        assert!(
+            calls.load(Ordering::SeqCst) >= 1,
+            "the healthy slot still ran"
+        );
+    }
+
     #[tokio::test]
     #[allow(clippy::await_holding_lock)]
     async fn declarative_dependency_results_are_forwarded_to_downstream_prompt() {

crates/workflow-js/src/driver.rs

@@ -235,6 +235,34 @@ pub enum ProgressEvent {
         /// The rejection already surfaced to JS.
         message: String,
     },
+    /// A `parallel()` / `pipeline()` fan-out resolved with every slot failed
+    /// and nothing surviving (R9). Emitted by the prelude beside its run-log
+    /// breadcrumb. A fan-out of thunks that throw without calling `task()`
+    /// leaves no task record and no dispatch failure behind, so a log line
+    /// was the only trace — and a log line cannot feed the status
+    /// classifier. This event is the structured twin of that breadcrumb:
+    /// drivers count it so the terminal-status ledger can refuse to record
+    /// such a run as a plain success.
+    FanoutAllSlotsFailed {
+        /// Which fan-out construct died: `parallel` or `pipeline`.
+        construct: String,
+        /// Slots that failed (always equal to `total`).
+        failed: u32,
+        /// Total slots the fan-out declared.
+        total: u32,
+    },
+    /// ONE slot of a settled fan-out dropped to null while other slots
+    /// survived. The structured twin of the per-slot breadcrumb: the ledger
+    /// counts these so a PARTIALLY failed fan-out records Degraded instead
+    /// of a clean Completed with silently lost work.
+    FanoutSlotDropped {
+        /// Which fan-out construct dropped it: `parallel` or `pipeline`.
+        construct: String,
+        /// Typed failure kind (`script`, `task`, `schema`, ...).
+        kind: String,
+        /// Zero-based slot index.
+        slot: u32,
+    },
 }
 
 /// Host-side executor for a Workflow run.

crates/workflow-js/src/error.rs

@@ -51,3 +51,79 @@ pub enum DriverError {
     #[error("driver unavailable: {0}")]
     Unavailable(String),
 }
+
+/// Why a `task()` call failed, as a stable machine kind (R9).
+///
+/// The host assigns the kind at the point the failure actually happens and
+/// ships it on the task envelope as `error_kind`; the JS prelude copies it
+/// onto the thrown `Error` as `.kind`, and `parallel()` / `pipeline()`
+/// classify slots from that field alone.
+///
+/// This exists because the classification used to be a substring match on the
+/// operator-facing message. A child whose own reply text contained the words
+/// "budget exhausted" — or a script that threw `new Error("run cancelled")` —
+/// could forge a fatal classification and abort a healthy run, and a genuine
+/// subagent failure was indistinguishable from a plain script throw. The kind
+/// is now data, not prose.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum TaskErrorKind {
+    /// The call never became a child: malformed options, an invalid
+    /// `responseSchema`, the workflow lifetime cap, or a driver admission
+    /// refusal. Nothing ran, so nothing was spent.
+    Admission,
+    /// The run's shared token pool is exhausted — either the pre-spawn gate
+    /// or a child that reported [`crate::TaskCompletion::BudgetExhausted`].
+    Budget,
+    /// The run or the child was cancelled. Always fatal to the fan-out:
+    /// cancellation is the run's own deadline, never a per-slot outcome.
+    Cancelled,
+    /// The child ran and failed (error result, timeout, tool refusal, ...).
+    Agent,
+    /// The reply did not satisfy `responseSchema` and no bounded repair
+    /// remained.
+    Schema,
+    /// The driver seam broke: unavailable, or the completion channel dropped
+    /// before a terminal outcome arrived.
+    Driver,
+}
+
+impl TaskErrorKind {
+    /// The wire spelling carried on the envelope and on JS `Error.kind`.
+    pub fn as_str(self) -> &'static str {
+        match self {
+            Self::Admission => "admission",
+            Self::Budget => "budget",
+            Self::Cancelled => "cancelled",
+            Self::Agent => "agent",
+            Self::Schema => "schema",
+            Self::Driver => "driver",
+        }
+    }
+}
+
+/// One `task()` failure: the operator-facing message plus its typed kind.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub(crate) struct TaskError {
+    pub(crate) kind: TaskErrorKind,
+    pub(crate) message: String,
+}
+
+impl TaskError {
+    pub(crate) fn new(kind: TaskErrorKind, message: impl Into<String>) -> Self {
+        Self {
+            kind,
+            message: message.into(),
+        }
+    }
+}
+
+impl From<&DriverError> for TaskErrorKind {
+    fn from(err: &DriverError) -> Self {
+        match err {
+            // A refused spawn never produced a child; it is admission, not a
+            // failure of work that ran.
+            DriverError::Rejected(_) => Self::Admission,
+            DriverError::Unavailable(_) => Self::Driver,
+        }
+    }
+}

crates/workflow-js/src/lib.rs

@@ -22,12 +22,27 @@
 //!   that fails `responseSchema` gets up to
 //!   [`SCHEMA_REPAIR_MAX_ATTEMPTS`] bounded repairs (#5583) before it throws;
 //!   the failed attempts are reported as driver receipts either way.
-//! * `parallel(thunks)` — all-settled fan-out; an ordinary failed slot becomes
-//!   `null`; schema-contract failures and run cancellation still fail the run;
-//!   at most [`PARALLEL_MAX_ITEMS`] items.
+//! * `parallel(thunks, opts)` — fan-out, at most [`PARALLEL_MAX_ITEMS`] items.
+//!   `opts.mode` is `settled` (default), `fail-fast`, or `partial`; an
+//!   unrecognized mode throws. `settled` keeps today's ergonomics — an
+//!   ordinary failed slot resolves to `null` — while schema-contract failures
+//!   and run cancellation still fail the run. `fail-fast` rejects the fan-out
+//!   with the first non-fatal slot error instead of nulling it. `partial`
+//!   resolves every non-cancellation failure to
+//!   `{ __taskError: { index, kind, message } }`.
 //! * `pipeline(items, ...stages)` — per-item stage chains with no barrier
-//!   between stages; an ordinary stage error drops that item to `null`, while
-//!   schema-contract failures and cancellation still fail the run; same cap.
+//!   between stages; same cap and the same three modes via the
+//!   `pipeline(items, { stages, mode })` overload.
+//!
+//! Both fan-outs attach a non-enumerable `errors` array to the resolved
+//! result — `[{ index, kind, message }]`, ordered by index, empty when
+//! nothing failed. The array's own contents, length, and JSON encoding are
+//! unchanged, so a dropped slot is now inspectable instead of erased.
+//!
+//! `kind` is a [`TaskErrorKind`] wire string (`admission`, `budget`,
+//! `cancelled`, `agent`, `schema`, `driver`) assigned by the host where the
+//! failure happened, or `script` for an error the script threw itself. It is
+//! read from the thrown `Error`'s `.kind`, never inferred from message text.
 //! * `log(msg)` / `phase(title)` — progress events forwarded to the driver.
 //! * `budget.total` / `budget.spent()` / `budget.remaining()` — live driver
 //!   snapshots (`total` is `null` and `remaining()` is `Infinity` when no
@@ -55,7 +70,7 @@ pub use driver::{
     BudgetSnapshot, ProgressEvent, SpawnedTask, TaskCompletion, TaskRequest, WorkflowDriver,
     normalize_profile,
 };
-pub use error::{DriverError, WorkflowJsError};
+pub use error::{DriverError, TaskErrorKind, WorkflowJsError};
 pub use schema::{SCHEMA_RAW_CARRY_CHARS, SCHEMA_RAW_PREVIEW_CHARS, SCHEMA_REPAIR_MAX_ATTEMPTS};
 pub use vm::{VmLimits, WorkflowRunCancel, WorkflowVm};
 

crates/workflow-js/src/testing.rs

@@ -31,6 +31,12 @@ pub enum FakeReply {
     BudgetExhausted(String),
     /// Refuse admission: `spawn_task` returns [`DriverError::Rejected`].
     Reject(String),
+    /// The driver seam is gone: `spawn_task` returns
+    /// [`DriverError::Unavailable`].
+    Unavailable(String),
+    /// Admit the task, then drop the completion sender without ever sending a
+    /// terminal outcome — the "driver dropped the completion channel" path.
+    DropCompletion,
     /// Admit the task but never complete it (for cancellation tests). The
     /// completion sender is held so the channel stays open.
     Never,
@@ -148,8 +154,12 @@ impl WorkflowDriver for FakeDriver {
                     None,
                 )
             });
-            if let FakeReply::Reject(message) = reply {
-                return Err(DriverError::Rejected(message));
+            match reply {
+                FakeReply::Reject(message) => return Err(DriverError::Rejected(message)),
+                FakeReply::Unavailable(message) => {
+                    return Err(DriverError::Unavailable(message));
+                }
+                _ => {}
             }
             inner.requests.push(request);
             inner.budget.spent += inner.spend_per_task;
@@ -160,6 +170,7 @@ impl WorkflowDriver for FakeDriver {
         let (tx, rx) = oneshot::channel();
         match reply {
             FakeReply::Never => self.lock().held.push(tx),
+            FakeReply::DropCompletion => drop(tx),
             reply => {
                 let completion = match reply {
                     FakeReply::Complete(text) => TaskCompletion::Completed { text },
@@ -168,7 +179,10 @@ impl WorkflowDriver for FakeDriver {
                     FakeReply::BudgetExhausted(message) => {
                         TaskCompletion::BudgetExhausted { message }
                     }
-                    FakeReply::Reject(_) | FakeReply::Never => unreachable!("handled above"),
+                    FakeReply::Reject(_)
+                    | FakeReply::Unavailable(_)
+                    | FakeReply::Never
+                    | FakeReply::DropCompletion => unreachable!("handled above"),
                 };
                 match delay {
                     None => {

crates/workflow-js/src/vm.rs

@@ -25,7 +25,7 @@ use serde::Deserialize;
 use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot, watch};
 
 use crate::driver::{ProgressEvent, TaskCompletion, TaskRequest, WorkflowDriver};
-use crate::error::WorkflowJsError;
+use crate::error::{TaskError, TaskErrorKind, WorkflowJsError};
 use crate::schema::{
     ReplyDecodeError, SCHEMA_REPAIR_MAX_ATTEMPTS, carried_raw, compile_schema, decode_reply,
     repair_prompt,
@@ -566,6 +566,42 @@ fn install_host(
         )
         .map_err(init_err)?;
 
+    // Structured twin of the prelude's "every slot failed" breadcrumb (R9):
+    // a dead fan-out of script-thrown thunks leaves no task record behind,
+    // so the host needs a typed event — not a log line — to keep the run's
+    // terminal status honest.
+    let fanout_driver = driver.clone();
+    globals
+        .set(
+            "__workflow_every_slot_failed",
+            Func::from(move |construct: String, failed: u32, total: u32| {
+                fanout_driver.progress(ProgressEvent::FanoutAllSlotsFailed {
+                    construct,
+                    failed,
+                    total,
+                });
+            }),
+        )
+        .map_err(init_err)?;
+
+    // Structured twin of the per-slot "dropped a failed slot as null"
+    // breadcrumb (R9): a PARTIALLY failed fan-out still resolves and leaves
+    // no task record for the dropped slot, so without this event the ledger
+    // cannot see the loss and records a clean Completed.
+    let fanout_driver = driver.clone();
+    globals
+        .set(
+            "__workflow_slot_dropped",
+            Func::from(move |construct: String, kind: String, slot: u32| {
+                fanout_driver.progress(ProgressEvent::FanoutSlotDropped {
+                    construct,
+                    kind,
+                    slot,
+                });
+            }),
+        )
+        .map_err(init_err)?;
+
     let phase_driver = driver.clone();
     globals
         .set(
@@ -619,8 +655,10 @@ fn init_err(err: rquickjs::Error) -> WorkflowJsError {
 }
 
 /// The `task()` host call. Everything that can go wrong is reported through
-/// the JSON envelope (`{"error": ...}`) so the prelude re-throws it as a real
-/// JS `Error` with a script-side stack.
+/// the JSON envelope (`{"error": ..., "error_kind": ...}`) so the prelude
+/// re-throws it as a real JS `Error` with a script-side stack and a typed
+/// [`TaskErrorKind`] on `.kind` (R9). The kind is assigned here, where the
+/// failure actually happened — never re-derived from the message text.
 async fn task_host(
     opts_json: String,
     driver: Arc<dyn WorkflowDriver>,
@@ -630,7 +668,9 @@ async fn task_host(
     let outcome = task_host_inner(opts_json, driver, cancel, spawned).await;
     let envelope = match outcome {
         Ok(value) => serde_json::json!({ "value": value }),
-        Err(message) => serde_json::json!({ "error": message }),
+        Err(TaskError { kind, message }) => {
+            serde_json::json!({ "error": message, "error_kind": kind.as_str() })
+        }
     };
     envelope.to_string()
 }
@@ -720,14 +760,27 @@ fn repair_request(
     request
 }
 
+/// The one cancellation error: the run's deadline fired. Fatal in every
+/// `parallel()` / `pipeline()` mode — a cancelled run must never resolve into
+/// a slot value.
+fn cancelled_task() -> TaskError {
+    TaskError::new(TaskErrorKind::Cancelled, "task(): run cancelled")
+}
+
+/// A terminal `responseSchema` failure, already receipted by [`fail_schema`].
+fn schema_task(message: String) -> TaskError {
+    TaskError::new(TaskErrorKind::Schema, message)
+}
+
 async fn task_host_inner(
     opts_json: String,
     driver: Arc<dyn WorkflowDriver>,
     cancel: CancelHandle,
     spawned: Rc<Cell<u64>>,
-) -> Result<serde_json::Value, String> {
+) -> Result<serde_json::Value, TaskError> {
+    let admission = |message: String| TaskError::new(TaskErrorKind::Admission, message);
     let request = parse_task_options(&opts_json)
-        .map_err(|message| reject_task(&driver, &opts_json, message))?;
+        .map_err(|message| admission(reject_task(&driver, &opts_json, message)))?;
     // Compile the schema before spawning so a malformed one fails fast
     // instead of burning a subagent.
     let validator = request
@@ -735,34 +788,37 @@ async fn task_host_inner(
         .as_ref()
         .map(compile_schema)
         .transpose()
-        .map_err(|message| reject_task(&driver, &opts_json, message))?;
+        .map_err(|message| admission(reject_task(&driver, &opts_json, message)))?;
 
     // Lifetime backstop (design §4.3) — checked and bumped before any await.
     if spawned.get() >= WORKFLOW_LIFETIME_CAP {
-        return Err(reject_task(
+        return Err(admission(reject_task(
             &driver,
             &opts_json,
             format!(
                 "task(): Workflow lifetime agent cap ({WORKFLOW_LIFETIME_CAP}) reached for this run"
             ),
-        ));
+        )));
     }
     // Fast-fail budget gate. The authoritative reservation lives in the
     // driver (design §5.3); this only stops obviously-doomed spawns early.
     let snapshot = driver.budget();
     if snapshot.exhausted() {
-        return Err(reject_task(
-            &driver,
-            &opts_json,
-            format!(
-                "task(): budget exhausted ({} of {} tokens spent)",
-                snapshot.spent,
-                snapshot.total.unwrap_or(0)
+        return Err(TaskError::new(
+            TaskErrorKind::Budget,
+            reject_task(
+                &driver,
+                &opts_json,
+                format!(
+                    "task(): budget exhausted ({} of {} tokens spent)",
+                    snapshot.spent,
+                    snapshot.total.unwrap_or(0)
+                ),
             ),
         ));
     }
     if cancel.is_cancelled() {
-        return Err("task(): run cancelled".to_string());
+        return Err(cancelled_task());
     }
 
     // Bounded schema repair (#5583): after a failed `responseSchema` decode,
@@ -782,23 +838,38 @@ async fn task_host_inner(
         let spawned_task = driver
             .spawn_task(current.clone())
             .await
-            .map_err(|err| err.to_string())?;
+            .map_err(|err| TaskError::new(TaskErrorKind::from(&err), err.to_string()))?;
         let task_id = spawned_task.task_id;
         let completion_rx = spawned_task.completion;
         let completion = tokio::select! {
-            _ = cancel.cancelled() => return Err("task(): run cancelled".to_string()),
-            completion = completion_rx => completion
-                .map_err(|_| "task(): driver dropped the completion channel".to_string())?,
+            _ = cancel.cancelled() => return Err(cancelled_task()),
+            completion = completion_rx => completion.map_err(|_| {
+                TaskError::new(
+                    TaskErrorKind::Driver,
+                    "task(): driver dropped the completion channel",
+                )
+            })?,
         };
 
         let text = match completion {
             TaskCompletion::Completed { text } => text,
             TaskCompletion::Failed { message } => {
-                return Err(format!("task(): subagent failed: {message}"));
+                return Err(TaskError::new(
+                    TaskErrorKind::Agent,
+                    format!("task(): subagent failed: {message}"),
+                ));
+            }
+            TaskCompletion::Cancelled => {
+                return Err(TaskError::new(
+                    TaskErrorKind::Cancelled,
+                    "task(): subagent cancelled",
+                ));
             }
-            TaskCompletion::Cancelled => return Err("task(): subagent cancelled".to_string()),
             TaskCompletion::BudgetExhausted { message } => {
-                return Err(format!("task(): budget exhausted: {message}"));
+                return Err(TaskError::new(
+                    TaskErrorKind::Budget,
+                    format!("task(): budget exhausted: {message}"),
+                ));
             }
         };
         // Without a schema the raw text is the contract; with one, the decode
@@ -812,7 +883,7 @@ async fn task_host_inner(
         };
         let (raw, raw_truncated) = carried_raw(&text);
         if attempt >= last_attempt {
-            return Err(fail_schema(
+            return Err(schema_task(fail_schema(
                 &driver,
                 task_id,
                 attempt,
@@ -820,7 +891,7 @@ async fn task_host_inner(
                 raw,
                 raw_truncated,
                 None,
-            ));
+            )));
         }
         // The attempt failed but a repair remains: record it as a receipt
         // (visible even when the repair succeeds), then re-run the admission
@@ -834,7 +905,7 @@ async fn task_host_inner(
             raw_truncated,
         });
         if spawned.get() >= WORKFLOW_LIFETIME_CAP {
-            return Err(fail_schema(
+            return Err(schema_task(fail_schema(
                 &driver,
                 task_id,
                 attempt,
@@ -844,11 +915,11 @@ async fn task_host_inner(
                 Some(format!(
                     "workflow lifetime agent cap ({WORKFLOW_LIFETIME_CAP}) reached"
                 )),
-            ));
+            )));
         }
         let snapshot = driver.budget();
         if snapshot.exhausted() {
-            return Err(fail_schema(
+            return Err(schema_task(fail_schema(
                 &driver,
                 task_id,
                 attempt,
@@ -856,15 +927,15 @@ async fn task_host_inner(
                 raw,
                 raw_truncated,
                 Some("budget exhausted".to_string()),
-            ));
+            )));
         }
         if cancel.is_cancelled() {
-            return Err("task(): run cancelled".to_string());
+            return Err(cancelled_task());
         }
         if let Some(wall) = wall_time_secs_left {
             let remaining = wall.saturating_sub(started.elapsed().as_secs());
             if remaining == 0 {
-                return Err(fail_schema(
+                return Err(schema_task(fail_schema(
                     &driver,
                     task_id,
                     attempt,
@@ -872,7 +943,7 @@ async fn task_host_inner(
                     raw,
                     raw_truncated,
                     Some("no wall-time left from wallTimeSecs".to_string()),
-                ));
+                )));
             }
             wall_time_secs_left = Some(remaining);
         }
@@ -1199,6 +1270,8 @@ const PRELUDE_TEMPLATE: &str = r#""use strict";
   // globalThis so scripts only see the documented Workflow surface (#4129).
   const hostTask = __workflow_task;
   const hostLog = __workflow_log;
+  const hostEverySlotFailed = __workflow_every_slot_failed;
+  const hostSlotDropped = __workflow_slot_dropped;
   const hostPhase = __workflow_phase;
   const hostBudgetTotal = __workflow_budget_total;
   const hostBudgetSpent = __workflow_budget_spent;
@@ -1206,27 +1279,81 @@ const PRELUDE_TEMPLATE: &str = r#""use strict";
 
   const MAX_ITEMS = __MAX_ITEMS__;
   const taskErrorText = (err) => String(err && err.message !== undefined ? err.message : err);
-  // Typed error kinds (R9): fatal-vs-recoverable used to be a substring
-  // match; now every error thrown by task() carries a stable `kind` and the
-  // classifiers resolve kind-first with the legacy text prefixes as fallback
-  // so pre-existing envelopes still classify.
-  const formatTaskErrorKind = (err) => {
-    if (err && typeof err === "object" && typeof err.kind === "string") {
-      return err.kind;
-    }
-    const text = taskErrorText(err);
-    if (text.includes("run cancelled") || text.includes("subagent cancelled")) return "cancelled";
-    if (text.includes("budget exhausted")) return "budget";
-    if (text.includes("responseSchema")) return "schema";
-    if (text.includes("spawn rejected")) return "admission";
-    if (text.includes("driver unavailable") || text.includes("driver dropped")) return "driver";
-    return "task";
-  };
+
+  // Typed slot errors (R9). Every error thrown by task() carries a
+  // host-assigned `kind` copied off the task envelope; anything else that
+  // reaches a slot was thrown by the script itself and reports as "script".
+  //
+  // The kind is read from the error object and never guessed from message
+  // text. A substring classifier let a child's own words ("...budget
+  // exhausted...", "...responseSchema...") forge a fatal classification and
+  // abort a healthy run, and it could not tell a genuine subagent failure
+  // apart from a plain `throw new Error(...)` in a stage.
+  const HOST_KINDS = ["admission", "budget", "cancelled", "agent", "schema", "driver"];
+  const SCRIPT_KIND = "script";
+  const taskErrorKind = (err) =>
+    err !== null && typeof err === "object" && HOST_KINDS.indexOf(err.kind) !== -1
+      ? err.kind
+      : SCRIPT_KIND;
+  // Fatal kinds are never absorbed into a slot value: cancellation is the
+  // run's own deadline, and a schema breach means the contract the caller
+  // explicitly asked for was not met. `mode: "partial"` opts out for schema
+  // (and only schema) by keeping it as a structured slot value instead.
   const isFatalTaskError = (err) => {
-    const kind = formatTaskErrorKind(err);
+    const kind = taskErrorKind(err);
     return kind === "cancelled" || kind === "schema";
   };
 
+  // Stamp the resolved kind onto an error that is about to be rethrown, so a
+  // script's own `catch (err) { err.kind }` reads the same vocabulary the
+  // slot classifier used. Host errors already carry theirs; this only names
+  // the script throws, which would otherwise surface as `undefined`.
+  const stampKind = (err, kind) => {
+    if (err !== null && typeof err === "object" && err.kind === undefined) {
+      try {
+        err.kind = kind;
+      } catch (_) {
+        // A frozen error keeps whatever it has; the log line still names it.
+      }
+    }
+    return err;
+  };
+
+  const SLOT_MODES = ["settled", "fail-fast", "partial"];
+  // `settled` is the default and is exactly today's behavior: a non-fatal
+  // slot failure resolves to `null` so an author need not handle every error.
+  // An unrecognized mode throws rather than silently falling back — a typo
+  // like `mode: "failfast"` used to read as `settled` and quietly keep
+  // dropping slots the author believed were now fatal.
+  const slotMode = (fn, opts) => {
+    if (opts === null || typeof opts !== "object" || opts.mode === undefined) {
+      return "settled";
+    }
+    if (SLOT_MODES.indexOf(opts.mode) === -1) {
+      throw new Error(
+        fn + "(): unknown mode " + JSON.stringify(opts.mode) +
+        "; expected one of " + SLOT_MODES.join(", ")
+      );
+    }
+    return opts.mode;
+  };
+
+  // The failure ledger for one fan-out, attached to the resolved array as a
+  // non-enumerable `errors` property. Non-enumerable and non-index, so the
+  // array's contents, length, and JSON encoding are byte-identical to before:
+  // `results.filter(Boolean)` still works, and a script that wants to know
+  // WHY a slot is null can now ask instead of guessing.
+  const attachSlotErrors = (results, errors) => {
+    errors.sort((a, b) => a.index - b.index);
+    Object.defineProperty(results, "errors", {
+      value: Object.freeze(errors.map((entry) => Object.freeze(entry))),
+      enumerable: false,
+      configurable: false,
+      writable: false,
+    });
+    return results;
+  };
+
   globalThis.task = async (opts) => {
     if (opts === null || typeof opts !== "object") {
       throw new TypeError("task(): expected an options object");
@@ -1234,7 +1361,11 @@ const PRELUDE_TEMPLATE: &str = r#""use strict";
     const envelope = JSON.parse(await hostTask(JSON.stringify(opts)));
     if (envelope.error !== undefined) {
       const err = new Error(envelope.error);
-      err.kind = envelope.error_kind || formatTaskErrorKind(err);
+      // The host always names the kind; a missing one means an envelope this
+      // prelude did not produce, which is not a typed task failure.
+      err.kind = HOST_KINDS.indexOf(envelope.error_kind) !== -1
+        ? envelope.error_kind
+        : SCRIPT_KIND;
       throw err;
     }
     return envelope.value;
@@ -1247,54 +1378,70 @@ const PRELUDE_TEMPLATE: &str = r#""use strict";
     if (thunks.length > MAX_ITEMS) {
       throw new Error("parallel(): max " + MAX_ITEMS + " items per call");
     }
-    const mode = opts !== null && typeof opts === "object" ? opts.mode : undefined;
+    const mode = slotMode("parallel", opts);
     const failFast = mode === "fail-fast";
-    // Explicitly-configured partial success (#5583): a schema-contract slot
-    // failure becomes a structured { __taskError: { kind, message } } value
-    // instead of failing the run, so a reviewer fan-out can complete and the
-    // script can inspect per-slot outcomes. Cancellation still cancels the
-    // run, and the structured value never masquerades as a success.
     const partial = mode === "partial";
-    const dropOrReject = (err) => {
-      const kind = formatTaskErrorKind(err);
+    const errors = [];
+    // Returns the slot value, or throws to reject the whole fan-out.
+    const onSlotError = (index, err) => {
+      const kind = taskErrorKind(err);
+      const message = taskErrorText(err);
+      stampKind(err, kind);
+      // Cancellation is the run's deadline in every mode, partial included.
       if (kind === "cancelled") throw err;
-      if (partial && kind === "schema") {
-        hostLog("parallel(): partial mode kept a schema slot failure as __taskError: " + taskErrorText(err));
-        return { __taskError: { kind: kind, message: taskErrorText(err) } };
+      if (partial) {
+        // Opt-in partial mode: every non-cancellation slot failure becomes a
+        // structured value the script can branch on. It never masquerades as
+        // a success -- `__taskError` is the whole point of the shape.
+        errors.push({ index: index, kind: kind, message: message });
+        hostLog(
+          "parallel(): partial mode kept a failed slot as __taskError (kind=" +
+          kind + ", slot " + index + "): " + message
+        );
+        return { __taskError: { index: index, kind: kind, message: message } };
       }
       if (isFatalTaskError(err)) throw err;
       if (failFast) {
-        if (err && typeof err === "object") {
-          err.kind = err.kind || kind;
-        }
-        hostLog("parallel(): fail-fast slot error: " + taskErrorText(err));
+        hostLog(
+          "parallel(): fail-fast slot error (kind=" + kind + ", slot " + index + "): " + message
+        );
         throw err;
       }
-      hostLog("parallel(): dropped a failed slot as null: " + taskErrorText(err));
+      errors.push({ index: index, kind: kind, message: message });
+      hostLog(
+        "parallel(): dropped a failed slot as null (kind=" + kind + ", slot " + index + "): " +
+        message
+      );
+      hostSlotDropped("parallel", kind, index);
       return null;
     };
-    return Promise.all(thunks.map((thunk) => {
+    const slots = thunks.map((thunk, index) => {
       try {
-        return Promise.resolve(typeof thunk === "function" ? thunk() : thunk).catch(dropOrReject);
+        return Promise.resolve(typeof thunk === "function" ? thunk() : thunk)
+          .catch((err) => onSlotError(index, err));
       } catch (err) {
-        const kind = formatTaskErrorKind(err);
-        if (kind === "cancelled") return Promise.reject(err);
-        if (partial && kind === "schema") {
-          hostLog("parallel(): partial mode kept a schema slot failure as __taskError: " + taskErrorText(err));
-          return { __taskError: { kind: kind, message: taskErrorText(err) } };
+        try {
+          return onSlotError(index, err);
+        } catch (rethrown) {
+          return Promise.reject(rethrown);
         }
-        if (isFatalTaskError(err)) return Promise.reject(err);
-        if (failFast) {
-          if (err && typeof err === "object") {
-            err.kind = err.kind || kind;
-          }
-          hostLog("parallel(): fail-fast slot error: " + taskErrorText(err));
-          return Promise.reject(err);
-        }
-        hostLog("parallel(): dropped a failed slot as null: " + taskErrorText(err));
-        return null;
       }
-    }));
+    });
+    return Promise.all(slots).then((results) => {
+      // A fan-out where nothing survived is a dead fan-out, not resilience.
+      // The default stays ergonomic (the array still resolves) but the run
+      // log says so in one line an operator can grep for, and the structured
+      // event below lets the host status classifier refuse to call such a
+      // run a plain success.
+      if (results.length > 0 && errors.length === results.length) {
+        hostLog(
+          "parallel(): every slot failed (" + errors.length + " of " + results.length +
+          "); no work survived this fan-out"
+        );
+        hostEverySlotFailed("parallel", errors.length, results.length);
+      }
+      return attachSlotErrors(results, errors);
+    });
   };
 
   globalThis.pipeline = (items, ...stages) => {
@@ -1312,30 +1459,57 @@ const PRELUDE_TEMPLATE: &str = r#""use strict";
       typeof stages[0] === "object" &&
       Array.isArray(stages[0].stages)
     ) {
-      mode = stages[0].mode === "fail-fast" ? "fail-fast" : "settled";
+      mode = slotMode("pipeline", stages[0]);
       stages = stages[0].stages;
     }
+    const failFast = mode === "fail-fast";
+    const partial = mode === "partial";
+    const errors = [];
     return Promise.all(items.map(async (item, index) => {
       let value = item;
       for (const stage of stages) {
         try {
           value = await stage(value, item, index);
         } catch (err) {
+          const kind = taskErrorKind(err);
+          const message = taskErrorText(err);
+          stampKind(err, kind);
+          if (kind === "cancelled") throw err;
+          if (partial) {
+            errors.push({ index: index, kind: kind, message: message });
+            hostLog(
+              "pipeline(): partial mode kept item " + index +
+              " as __taskError (kind=" + kind + "): " + message
+            );
+            return { __taskError: { index: index, kind: kind, message: message } };
+          }
           if (isFatalTaskError(err)) throw err;
-          const failFast = mode === "fail-fast";
           if (failFast) {
-            if (err && typeof err === "object") {
-              err.kind = err.kind || formatTaskErrorKind(err);
-            }
-            hostLog("pipeline(): fail-fast stage error on item " + index + ": " + taskErrorText(err));
+            hostLog(
+              "pipeline(): fail-fast stage error on item " + index +
+              " (kind=" + kind + "): " + message
+            );
             throw err;
           }
-          hostLog("pipeline(): dropped item " + index + " as null: " + taskErrorText(err));
+          errors.push({ index: index, kind: kind, message: message });
+          hostLog(
+            "pipeline(): dropped item " + index + " as null (kind=" + kind + "): " + message
+          );
+          hostSlotDropped("pipeline", kind, index);
           return null;
         }
       }
       return value;
-    }));
+    })).then((results) => {
+      if (results.length > 0 && errors.length === results.length) {
+        hostLog(
+          "pipeline(): every item failed (" + errors.length + " of " + results.length +
+          "); no work survived this pipeline"
+        );
+        hostEverySlotFailed("pipeline", errors.length, results.length);
+      }
+      return attachSlotErrors(results, errors);
+    });
   };
 
   globalThis.log = (message) => {
@@ -1355,6 +1529,7 @@ const PRELUDE_TEMPLATE: &str = r#""use strict";
   for (const name of [
     "__workflow_task",
     "__workflow_log",
+    "__workflow_every_slot_failed",
     "__workflow_phase",
     "__workflow_budget_total",
     "__workflow_budget_spent",

crates/workflow-js/tests/vm_tests.rs

@@ -602,7 +602,7 @@ async fn parallel_partial_mode_keeps_schema_failures_as_structured_slots() {
         ], { mode: "partial" });
         return results.map((slot) =>
             slot && typeof slot === "object" && slot.__taskError !== undefined
-                ? "schema:" + slot.__taskError.kind
+                ? "error:" + slot.__taskError.kind
                 : slot === null
                   ? "null"
                   : "value:" + JSON.stringify(slot)
@@ -619,8 +619,11 @@ async fn parallel_partial_mode_keeps_schema_failures_as_structured_slots() {
             "value:{\"refuted\":true}",
             // The JS-level kind is the fatal "schema"; the finer decode kind
             // (json_parse) lives on the receipt events, asserted below.
-            "schema:schema",
-            "null"
+            "error:schema",
+            // R9 behavior change: partial mode used to drop a dead subagent
+            // to `null` — indistinguishable from a slot that legitimately
+            // returned nothing. It is now a typed, inspectable failure.
+            "error:agent"
         ])
     );
     // Every failed slot still leaves its terminal receipt.
@@ -998,7 +1001,9 @@ async fn parallel_fail_fast_rejects_with_the_typed_slot_error() {
     .unwrap();
     let text = value.as_str().unwrap();
     assert!(
-        text.starts_with("task:") && text.contains("boom"),
+        // R9: a child that ran and failed is `agent`, distinct from the
+        // `script` kind a plain `throw` in a thunk produces.
+        text.starts_with("agent:") && text.contains("boom"),
         "fail-fast must reject with the typed slot error: {text}"
     );
     assert!(
@@ -1060,7 +1065,7 @@ async fn pipeline_fail_fast_rejects_instead_of_nulling_the_item() {
     )
     .await
     .unwrap();
-    assert_eq!(value, json!("task:stage boom"));
+    assert_eq!(value, json!("script:stage boom"));
 }
 
 #[tokio::test]
@@ -1558,6 +1563,7 @@ async fn sandbox_global_inventory_fails_closed_on_new_host_leaks() {
     for internal in [
         "__workflow_task",
         "__workflow_log",
+        "__workflow_every_slot_failed",
         "__workflow_phase",
         "__workflow_budget_total",
         "__workflow_budget_spent",
@@ -1930,3 +1936,493 @@ async fn vm_rejected_task_options_notify_the_driver() {
     assert_eq!(phase.as_deref(), Some("P1"));
     assert!(message.contains("bounded repo-relative paths"), "{message}");
 }
+
+// ---------------------------------------------------------------------------
+// R9: typed slot errors, inspectable settled failures, explicit modes.
+// ---------------------------------------------------------------------------
+
+/// Every way a `task()` can die gets its own kind, assigned by the host where
+/// the failure happened. Before R9 all six collapsed into two buckets
+/// ("budget"/"cancelled" if the message happened to say so, "task" otherwise),
+/// so a dead subagent and a typo'd script throw were the same thing.
+#[tokio::test]
+async fn every_task_failure_mode_carries_its_own_kind() {
+    let driver = Arc::new(FakeDriver::new());
+    driver.on("agent case", FakeReply::Fail("boom".to_string()));
+    driver.on(
+        "budget case",
+        FakeReply::BudgetExhausted("limit 10".to_string()),
+    );
+    driver.on("cancelled case", FakeReply::Cancelled);
+    driver.on(
+        "admission case",
+        FakeReply::Reject("admission cap".to_string()),
+    );
+    driver.on(
+        "driver case",
+        FakeReply::Unavailable("driver gone".to_string()),
+    );
+    driver.on("dropped case", FakeReply::DropCompletion);
+    driver.on(
+        "schema case",
+        FakeReply::Complete("not json at all".to_string()),
+    );
+
+    let value = run(
+        &driver,
+        r#"
+        const kinds = {};
+        const probe = async (name, opts) => {
+            try {
+                await task(opts);
+                kinds[name] = "none";
+            } catch (err) {
+                kinds[name] = err && err.kind;
+            }
+        };
+        await probe("agent", { description: "agent case" });
+        await probe("budget", { description: "budget case" });
+        await probe("cancelled", { description: "cancelled case" });
+        await probe("admission", { description: "admission case" });
+        await probe("driver", { description: "driver case" });
+        await probe("dropped", { description: "dropped case" });
+        await probe("schema", {
+            description: "schema case",
+            schemaRepairAttempts: 0,
+            responseSchema: { type: "object" },
+        });
+        // A malformed options object never reaches a child either.
+        await probe("bad-options", { description: "x", nosuchoption: 1 });
+        try {
+            await task("not an object");
+            kinds["not-an-object"] = "none";
+        } catch (err) {
+            kinds["not-an-object"] = String(err && err.kind);
+        }
+        return kinds;
+        "#,
+        json!(null),
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(
+        value,
+        json!({
+            "agent": "agent",
+            "budget": "budget",
+            "cancelled": "cancelled",
+            "admission": "admission",
+            "driver": "driver",
+            "dropped": "driver",
+            "schema": "schema",
+            "bad-options": "admission",
+            // A TypeError raised by the prelude's own argument check never
+            // came from the host, so it is a script error, not a task kind.
+            "not-an-object": "undefined",
+        })
+    );
+}
+
+/// The classifier reads `Error.kind`, never the message text. A child is free
+/// to say "budget exhausted" or "responseSchema" in its own failure prose;
+/// under the old substring classifier that forged a fatal kind and aborted an
+/// otherwise healthy fan-out.
+#[tokio::test]
+async fn slot_kinds_cannot_be_forged_from_child_failure_text() {
+    let driver = Arc::new(FakeDriver::new());
+    driver.on(
+        "liar",
+        FakeReply::Fail(
+            "the reviewer said the run cancelled because budget exhausted and responseSchema \
+             validation failed"
+                .to_string(),
+        ),
+    );
+
+    let value = run(
+        &driver,
+        r#"
+        const results = await parallel([
+            () => task({ description: "liar" }),
+            () => task({ description: "honest" }),
+        ]);
+        return {
+            slots: results,
+            kinds: results.errors.map((entry) => entry.kind),
+        };
+        "#,
+        json!(null),
+    )
+    .await
+    .expect("a child's prose must not cancel the run");
+
+    assert_eq!(
+        value,
+        json!({
+            "slots": [null, "done:honest"],
+            "kinds": ["agent"],
+        })
+    );
+}
+
+/// The settled default is unchanged on the wire — same slots, same length,
+/// same JSON — but the run can now ask why a slot is null instead of guessing.
+#[tokio::test]
+async fn settled_parallel_keeps_null_slots_and_attaches_an_inspectable_ledger() {
+    let driver = Arc::new(FakeDriver::new());
+    driver.on("beta", FakeReply::Fail("boom".to_string()));
+    driver.on(
+        "delta",
+        FakeReply::BudgetExhausted("pool drained".to_string()),
+    );
+
+    let value = run(
+        &driver,
+        r#"
+        const results = await parallel([
+            () => task({ description: "alpha" }),
+            () => task({ description: "beta" }),
+            () => task({ description: "gamma" }),
+            () => task({ description: "delta" }),
+        ]);
+        return {
+            slots: results,
+            length: results.length,
+            // Non-enumerable: the array still serializes as a plain array.
+            encoded: JSON.stringify(results),
+            errors: results.errors.map((entry) => ({
+                index: entry.index,
+                kind: entry.kind,
+                says: entry.message.indexOf("boom") !== -1
+                    || entry.message.indexOf("pool drained") !== -1,
+            })),
+        };
+        "#,
+        json!(null),
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(
+        value,
+        json!({
+            "slots": ["done:alpha", null, "done:gamma", null],
+            "length": 4,
+            "encoded": "[\"done:alpha\",null,\"done:gamma\",null]",
+            "errors": [
+                {"index": 1, "kind": "agent", "says": true},
+                {"index": 3, "kind": "budget", "says": true},
+            ],
+        })
+    );
+}
+
+/// A clean fan-out still gets the ledger, empty and frozen — a script can read
+/// `results.errors.length` unconditionally.
+#[tokio::test]
+async fn a_clean_fan_out_still_reports_an_empty_frozen_error_ledger() {
+    let value = run(
+        &Arc::new(FakeDriver::new()),
+        r#"
+        const results = await parallel([() => task({ description: "alpha" })]);
+        let mutated = false;
+        try {
+            results.errors = ["forged"];
+            mutated = true;
+        } catch (_) {
+            mutated = false;
+        }
+        return {
+            count: results.errors.length,
+            frozen: Object.isFrozen(results.errors),
+            mutated: mutated,
+            stillEmpty: results.errors.length,
+        };
+        "#,
+        json!(null),
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        value,
+        json!({"count": 0, "frozen": true, "mutated": false, "stillEmpty": 0})
+    );
+}
+
+/// `settled` is the spelling of today's default; naming it explicitly changes
+/// nothing.
+#[tokio::test]
+async fn explicit_settled_mode_matches_the_default() {
+    let driver = Arc::new(FakeDriver::new());
+    driver.on("beta", FakeReply::Fail("boom".to_string()));
+    let value = run(
+        &driver,
+        r#"
+        const thunks = () => [
+            () => task({ description: "alpha" }),
+            () => task({ description: "beta" }),
+        ];
+        const implicit = await parallel(thunks());
+        const explicit = await parallel(thunks(), { mode: "settled" });
+        return {
+            implicit: implicit,
+            explicit: explicit,
+            sameKinds: JSON.stringify(implicit.errors.map((e) => e.kind))
+                === JSON.stringify(explicit.errors.map((e) => e.kind)),
+        };
+        "#,
+        json!(null),
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        value,
+        json!({
+            "implicit": ["done:alpha", null],
+            "explicit": ["done:alpha", null],
+            "sameKinds": true,
+        })
+    );
+}
+
+/// A typo'd mode used to read as `settled`: the author believed slots were
+/// now fatal while they kept silently dropping. It throws instead.
+#[tokio::test]
+async fn an_unknown_mode_is_refused_rather_than_silently_settled() {
+    let driver = Arc::new(FakeDriver::new());
+    let value = run(
+        &driver,
+        r#"
+        const errs = [];
+        for (const mode of ["failfast", "all-settled", 7]) {
+            try {
+                await parallel([() => task({ description: "alpha" })], { mode });
+                errs.push("no-error");
+            } catch (err) {
+                errs.push(err.message);
+            }
+        }
+        try {
+            await pipeline([1], { stages: [(v) => v], mode: "failfast" });
+            errs.push("no-error");
+        } catch (err) {
+            errs.push(err.message);
+        }
+        return errs;
+        "#,
+        json!(null),
+    )
+    .await
+    .unwrap();
+    let messages = value.as_array().unwrap();
+    assert_eq!(messages.len(), 4, "{value}");
+    for message in messages {
+        let text = message.as_str().unwrap();
+        assert!(
+            text.contains("unknown mode") && text.contains("settled, fail-fast, partial"),
+            "{text}"
+        );
+    }
+    assert_eq!(
+        driver.spawn_count(),
+        0,
+        "a refused mode must not spawn anything"
+    );
+}
+
+/// Partial mode is the "inspect every outcome" contract: no failure is erased,
+/// and none of them can be mistaken for a value.
+#[tokio::test]
+async fn partial_mode_types_every_non_cancellation_failure() {
+    let driver = Arc::new(FakeDriver::new());
+    driver.on("dead", FakeReply::Fail("boom".to_string()));
+    driver.on("broke", FakeReply::BudgetExhausted("drained".to_string()));
+    driver.on("refused", FakeReply::Reject("admission cap".to_string()));
+
+    let value = run(
+        &driver,
+        r#"
+        const results = await parallel([
+            () => task({ description: "alive" }),
+            () => task({ description: "dead" }),
+            () => task({ description: "broke" }),
+            () => task({ description: "refused" }),
+        ], { mode: "partial" });
+        return {
+            shapes: results.map((slot) =>
+                slot && typeof slot === "object" && slot.__taskError
+                    ? slot.__taskError.kind + "@" + slot.__taskError.index
+                    : String(slot)
+            ),
+            ledger: results.errors.map((entry) => entry.kind),
+            noNulls: results.every((slot) => slot !== null),
+        };
+        "#,
+        json!(null),
+    )
+    .await
+    .expect("partial mode completes the fan-out");
+
+    assert_eq!(
+        value,
+        json!({
+            "shapes": ["done:alive", "agent@1", "budget@2", "admission@3"],
+            "ledger": ["agent", "budget", "admission"],
+            "noNulls": true,
+        })
+    );
+}
+
+/// `pipeline` speaks the same three modes and keeps the same ledger.
+#[tokio::test]
+async fn pipeline_supports_settled_fail_fast_and_partial_with_a_ledger() {
+    let driver = Arc::new(FakeDriver::new());
+    driver.on("bad-1", FakeReply::Fail("boom".to_string()));
+
+    let value = run(
+        &driver,
+        r#"
+        const stage = (item) => task({ description: item });
+        const settled = await pipeline(["ok-0", "bad-1", "ok-2"], stage);
+        const partial = await pipeline(["ok-0", "bad-1"], {
+            stages: [stage],
+            mode: "partial",
+        });
+        let failFast = "no-error";
+        try {
+            await pipeline(["ok-0", "bad-1"], { stages: [stage], mode: "fail-fast" });
+        } catch (err) {
+            failFast = err.kind + ":" + (err.message.indexOf("boom") !== -1);
+        }
+        return {
+            settled: settled,
+            settledLedger: settled.errors.map((e) => e.index + ":" + e.kind),
+            partial: partial.map((slot) =>
+                slot && typeof slot === "object" && slot.__taskError
+                    ? slot.__taskError.kind
+                    : slot
+            ),
+            failFast: failFast,
+        };
+        "#,
+        json!(null),
+    )
+    .await
+    .unwrap();
+
+    assert_eq!(
+        value,
+        json!({
+            "settled": ["done:ok-0", null, "done:ok-2"],
+            "settledLedger": ["1:agent"],
+            "partial": ["done:ok-0", "agent"],
+            "failFast": "agent:true",
+        })
+    );
+}
+
+/// A fan-out where nothing survived is a dead fan-out. The default still
+/// resolves (existing scripts keep working) but the run log says so in a line
+/// the host status classifier and an operator can both find.
+#[tokio::test]
+async fn a_fan_out_where_every_slot_failed_says_so_in_the_run_log() {
+    let driver = Arc::new(FakeDriver::new());
+    driver.on("doomed", FakeReply::Fail("boom".to_string()));
+    let value = run(
+        &driver,
+        r#"
+        const results = await parallel([
+            () => task({ description: "doomed a" }),
+            () => task({ description: "doomed b" }),
+        ]);
+        return { slots: results, failed: results.errors.length };
+        "#,
+        json!(null),
+    )
+    .await
+    .unwrap();
+    assert_eq!(value, json!({"slots": [null, null], "failed": 2}));
+    assert!(
+        driver.events().iter().any(|event| matches!(
+            event,
+            ProgressEvent::Log { message }
+                if message.contains("every slot failed (2 of 2)")
+                    && message.contains("no work survived")
+        )),
+        "a fully-failed fan-out must be named in the run log: {:?}",
+        driver.events()
+    );
+}
+
+/// Cancellation stays fatal in every mode — it is the run's deadline, not a
+/// per-slot outcome — and partial mode does not get to keep it as a value.
+#[tokio::test]
+async fn cancellation_is_fatal_in_partial_pipeline_mode() {
+    let driver = Arc::new(FakeDriver::new());
+    driver.on("hang", FakeReply::Never);
+    let cancel = WorkflowRunCancel::new();
+    let run_cancel = cancel.clone();
+    let run_driver = driver.clone();
+    let handle = tokio::spawn(async move {
+        WorkflowVm::new()
+            .run_script_with_cancel(
+                r#"
+                await pipeline(["hang"], {
+                    stages: [(item) => task({ description: item })],
+                    mode: "partial",
+                });
+                "#,
+                json!(null),
+                run_driver as Arc<dyn codewhale_workflow_js::WorkflowDriver>,
+                run_cancel,
+            )
+            .await
+    });
+
+    tokio::time::timeout(Duration::from_secs(2), async {
+        while driver.spawn_count() == 0 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("task should start");
+    cancel.cancel();
+
+    let result = handle.await.expect("VM task should join");
+    assert!(
+        matches!(result, Err(WorkflowJsError::Cancelled)),
+        "pipeline partial mode must not downgrade cancellation into a slot value: {result:?}"
+    );
+}
+
+/// The dropped-slot breadcrumb now names the kind, so the run log alone
+/// distinguishes "the agent failed" from "we never got to spawn it".
+#[tokio::test]
+async fn the_dropped_slot_breadcrumb_names_the_kind_and_the_slot() {
+    let driver = Arc::new(FakeDriver::new());
+    driver.on("beta", FakeReply::Fail("boom".to_string()));
+    run(
+        &driver,
+        r#"
+        return await parallel([
+            () => task({ description: "alpha" }),
+            () => task({ description: "beta" }),
+        ]);
+        "#,
+        json!(null),
+    )
+    .await
+    .unwrap();
+    assert!(
+        driver.events().iter().any(|event| matches!(
+            event,
+            ProgressEvent::Log { message }
+                if message.contains("dropped a failed slot as null")
+                    && message.contains("kind=agent")
+                    && message.contains("slot 1")
+        )),
+        "the breadcrumb must name the kind and the slot: {:?}",
+        driver.events()
+    );
+}

docs/AUTOMATIC_WORKFLOWS.md

@@ -122,6 +122,21 @@ approval policy. See [Sandbox](SANDBOX.md).
   before synthesizing one operator-facing summary. A `responseSchema` mismatch
   is a contract failure and intentionally fails the run instead of being
   silently converted to `null`.
+- A `null` slot is no longer anonymous. `parallel()` and `pipeline()` attach a
+  non-enumerable `errors` array to the result — `[{ index, kind, message }]`,
+  ordered by index — so a synthesizer can say *why* a slot is missing. The
+  array's own contents and JSON encoding are unchanged.
+- `kind` is one of `admission`, `budget`, `cancelled`, `agent`, `schema`,
+  `driver` (assigned by the host where the failure happened) or `script` (the
+  script threw it). Read it from the thrown `Error`'s `.kind`; it is never
+  inferred from message text, so a child's own prose cannot forge a kind.
+- `opts.mode` selects the contract: `settled` (default — today's behavior),
+  `fail-fast` (reject the whole fan-out with the first non-fatal slot error),
+  or `partial` (resolve every non-cancellation failure to
+  `{ __taskError: { index, kind, message } }`). An unrecognized mode throws
+  rather than quietly reading as `settled`.
+- A run whose every task failed is recorded as **failed**, not as a partial
+  success, even when the script itself returned a value.
 - Workflow token budgets govern admission and aggregate accounting. Once
   exhausted they reject later or descendant spawns, but children already
   running in parallel can reconcile aggregate usage above the hint because

docs/examples/dogfood-automatic/wf_a3_partial_failure_synthesis.workflow.js

@@ -2,7 +2,9 @@
  * #4131 WF-A3 — partial failure + synthesis.
  *
  * parallel() uses all-settled semantics: a failed slot becomes null; the run
- * continues so a synthesizer can still produce an operator summary.
+ * continues so a synthesizer can still produce an operator summary. The slot
+ * is null but not anonymous — `slots.errors` carries `{ index, kind, message }`
+ * for every drop, so the summary can name what was lost instead of guessing.
  *
  * Run: /workflow run docs/examples/dogfood-automatic/wf_a3_partial_failure_synthesis.workflow.js
  *
@@ -43,14 +45,23 @@ export default async function () {
 
   phase("Synthesize");
   const surviving = (slots || []).filter((s) => s != null);
+  // The typed failure ledger: which slot died, and of what.
+  const dropped = (slots.errors || []).map(
+    (e) => `slot ${e.index} (${e.kind}): ${e.message}`,
+  );
+  for (const line of dropped) {
+    log(`dropped: ${line}`);
+  }
   const summary = await task({
     description: "Synthesize from surviving parallel slots",
     label: "synthesizer",
     type: "general",
     prompt: [
       "Build one operator-facing summary from the surviving scout results.",
-      "Explicitly note which parallel slot failed or returned null.",
+      "Explicitly note which parallel slot failed, and why.",
       `slot_count=${(slots || []).length} surviving=${surviving.length}`,
+      "dropped_slots:",
+      dropped.length ? dropped.join("\n") : "(none)",
       "slots_json:",
       JSON.stringify(slots),
     ].join("\n"),
@@ -60,6 +71,7 @@ export default async function () {
     scenario: "WF-A3",
     slots,
     surviving_count: surviving.length,
+    dropped_slots: dropped,
     summary,
   };
 }