fix(swarm): surface errors, output contract, Windows-safe store, redact paths (HKUDS#119)

* fix(swarm): surface task error at read boundaries

run_swarm / get_swarm_status / get_run_result and the in-process
run_swarm tool each hand-maintained a field allowlist that omitted
SwarmTask.error. A misconfigured provider returned status="failed"
with no diagnosable reason, though the error was captured on disk
the whole time.

Add src/swarm/serialization.py as the single source of truth for the
per-task projection (now includes error + iterations) plus a run-level
error summary; route all three boundaries through it so the allowlist
cannot silently drift again. Additive JSON change; the only shape
normalization is _format_result missing-summary "" -> null.

(cherry picked from commit 95f2717ac32bd02829f650bc971e3cd6dd7fffa6)

* fix(swarm): enforce worker output contract

A worker that emitted only its plan ("Phase 1 — Plan"), mock data,
unparsed tool markup, or (for a data agent) made no tool call and
wrote no report was returned status="completed". runtime.py also
folded timeout/token_limit into completed. The run reported success
with a stub as final_report (P01); degraded workers were invisible
(P03).

Add WorkerStatus.incomplete and a hybrid output contract in worker.py:
content-sanity for every agent, tool-evidence only for data agents so
tool-less synthesis/editor roles are not false-rejected. Fail-fast to
incomplete (no corrective retry — left for a follow-up). runtime.py
only maps a true completed to TaskStatus.completed; anything else
fails the run, so final_report can no longer be a non-deliverable.

(cherry picked from commit 774b26824509678c953cd0200ef7a56c03f23201)

* fix(swarm): thread shell tools + unify runs root

The MCP run_swarm wrapper called start_run() without
include_shell_tools, so stdio swarm workers silently lost bash and
could not run the scripts their own execution rules mandate (P03-B).
build_filtered_registry also dropped a requested-but-unavailable
tool with no log, hiding the contradiction.

Pass the server's shell-tool policy into start_run and warn on a
dropped tool. Add swarm_runs_root() as the single source of truth
for the run store location, shared by mcp_server and the path_utils
run-dir allow-list so the two can no longer drift (P03-A: an
installed-wheel layout had put every worker run_dir outside it).

(cherry picked from commit 2975fda09b5ad44d15580d9330890f72f43c00ce)

* fix(swarm): make store atomic write Windows-safe

SwarmStore._atomic_write did tmp.replace(target) with no retry. On
Windows os.replace raises PermissionError WinError 5/32 when the
target is concurrently open by load_run on the poll path, crashing
the worker thread and leaving the run stuck pending (P13). POSIX
os.replace is atomic and never raises these.

Add a WinError-scoped, bounded retry around the rename and a matching
transient read/parse retry in load_run (the other side of the same
race). Non-transient errors re-raise immediately; off-Windows the
loop runs once, so POSIX behavior is unchanged. Pre-existing, not
introduced by the P01/P03/P04 fixes — surfaced by E2E dogfooding.

(cherry picked from commit 46a779cf436353591e7840879f8c32f763dde826)

* style(swarm): drop unused imports (ruff F401)

Remove an unused `from pathlib import Path` in the PR03 test added by
2975fda, and the pre-existing unused `from src.swarm.mailbox import
Mailbox` in runtime.py (file already in this branch's diff). Net ruff
delta for PR1-PR4 is now -1. The remaining mcp_server.py:41 E402 is
pre-existing on upstream and structural — left out of scope.

(cherry picked from commit e96f56af1ead9b5b7058c90ec0d4df738b45e473)

* fix: redact internal absolute paths in errors

Unknown preset / workspace-escape / missing run-or-task dir errors
embedded the full install path (OS username, .venvs/site-packages
topology) — CWE-209/497, reachable via MCP/API/agent/CLI (P10).

Drop the absolute path while keeping the actionable bits: preset
name + Available list, the workspace-root boundary, and the logical
run/task id (basename only). Two test_path_safety assertions that
matched the old leaking message are updated to the redacted wording
(behavior unchanged — still raises and rejects). Verified-clean
sites (caller's own input / env-var / names list) left untouched.

(cherry picked from commit 69ea45295410cd25ea3079c25b7e3c4c73390b80)

* fix(security): redact internal paths in errors

Add agent/src/tools/redaction.py: anchored prefix redaction (no
regex, idempotent, None-safe) that hides only known-internal root
prefixes while keeping the relative tail for diagnosability.

Wire it through the swarm error read boundaries (serialization,
runtime, store) and the file tools' final broad except so a leaked
absolute path cannot reach a user-facing string (CWE-209/497, P10).
ValueError branches and the protected providers/llm.py are left
untouched.

(cherry picked from commit 21dc9cdfacf17cccd31cacfbe34481f62f1390af)

* fix(security): redact swarm event error payloads

---------

Co-authored-by: Haozhe Wu <haozhe_wu@connect.hku.hk>

Author: 2 people

agent/mcp_server.py

@@ -371,15 +371,17 @@ def run_swarm(preset_name: str, variables: dict[str, str]) -> str:
     """
     import time
     from src.swarm.runtime import SwarmRuntime
-    from src.swarm.store import SwarmStore
+    from src.swarm.store import SwarmStore, swarm_runs_root
     from src.swarm.models import RunStatus
 
-    swarm_dir = AGENT_DIR / ".swarm" / "runs"
+    swarm_dir = swarm_runs_root()
     store = SwarmStore(base_dir=swarm_dir)
     runtime = SwarmRuntime(store=store)
 
     try:
-        run = runtime.start_run(preset_name, variables)
+        run = runtime.start_run(
+            preset_name, variables, include_shell_tools=_include_shell_tools
+        )
     except FileNotFoundError as exc:
         return json.dumps({"status": "error", "error": str(exc)}, ensure_ascii=False)
     except ValueError as exc:
@@ -392,16 +394,16 @@ def run_swarm(preset_name: str, variables: dict[str, str]) -> str:
         if current is None:
             return json.dumps({"status": "error", "error": "Run record lost"}, ensure_ascii=False)
         if current.status in (RunStatus.completed, RunStatus.failed, RunStatus.cancelled):
-            tasks = [
-                {"id": t.id, "agent_id": t.agent_id, "status": t.status.value, "summary": t.summary}
-                for t in current.tasks
-            ]
+            from src.swarm.serialization import run_level_error, serialize_task
+
+            tasks = [serialize_task(t) for t in current.tasks]
             return json.dumps(
                 {
                     "status": current.status.value,
                     "preset": preset_name,
                     "run_id": current.id,
                     "final_report": current.final_report,
+                    "error": run_level_error(current),
                     "tasks": tasks,
                     "total_input_tokens": current.total_input_tokens,
                     "total_output_tokens": current.total_output_tokens,
@@ -552,28 +554,23 @@ def get_market_data(
 
 
 def _get_swarm_store():
-    swarm_dir = AGENT_DIR / ".swarm" / "runs"
-    swarm_dir.mkdir(parents=True, exist_ok=True)
-    from src.swarm.store import SwarmStore
+    from src.swarm.store import SwarmStore, swarm_runs_root
 
+    swarm_dir = swarm_runs_root()
+    swarm_dir.mkdir(parents=True, exist_ok=True)
     return SwarmStore(base_dir=swarm_dir)
 
 
 def _run_to_dict(run) -> dict:
+    from src.swarm.serialization import run_level_error, serialize_task
+
     return {
         "run_id": run.id,
         "status": run.status.value,
         "preset": run.preset_name,
         "created_at": run.created_at,
-        "tasks": [
-            {
-                "id": t.id,
-                "agent_id": t.agent_id,
-                "status": t.status.value,
-                "summary": t.summary,
-            }
-            for t in run.tasks
-        ],
+        "error": run_level_error(run),
+        "tasks": [serialize_task(t) for t in run.tasks],
         "final_report": run.final_report,
         "total_input_tokens": run.total_input_tokens,
         "total_output_tokens": run.total_output_tokens,

agent/src/swarm/models.py

@@ -40,6 +40,23 @@ class RunStatus(str, Enum):
     cancelled = "cancelled"
 
 
+class WorkerStatus(str, Enum):
+    """Terminal status a worker returns.
+
+    ``incomplete`` is distinct from ``failed``: the worker ran without an
+    exception but produced no substantive deliverable (plan-only stub,
+    fabricated/mock numbers, unparsed tool markup, or a data agent that
+    made no tool call and wrote no report). It must never be folded into
+    ``completed`` (see P01/P03).
+    """
+
+    completed = "completed"
+    failed = "failed"
+    timeout = "timeout"
+    token_limit = "token_limit"
+    incomplete = "incomplete"
+
+
 class SwarmAgentSpec(BaseModel):
     """Role definition for a single agent in a Swarm.
 
@@ -203,7 +220,7 @@ class WorkerResult(BaseModel):
     """Return value after worker execution completes.
 
     Attributes:
-        status: "completed" or "failed".
+        status: WorkerStatus — completed|failed|timeout|token_limit|incomplete.
         summary: Execution summary.
         artifact_paths: List of generated artifact file paths.
         iterations: Actual ReAct iterations executed.
@@ -212,7 +229,7 @@ class WorkerResult(BaseModel):
         output_tokens: Cumulative output tokens (exact or estimated).
     """
 
-    status: str
+    status: WorkerStatus
     summary: str
     artifact_paths: list[str] = Field(default_factory=list)
     iterations: int = 0

agent/src/swarm/presets.py

@@ -39,7 +39,7 @@ def load_preset(name: str) -> dict:
     if not path.exists():
         available = [p.stem for p in PRESETS_DIR.glob("*.yaml")] if PRESETS_DIR.exists() else []
         raise FileNotFoundError(
-            f"Preset '{name}' not found at {path}. Available: {available}"
+            f"Preset {name!r} not found. Available: {available}"
         )
     return yaml.safe_load(path.read_text(encoding="utf-8"))
 

agent/src/swarm/runtime.py

@@ -21,7 +21,6 @@
 from typing import Callable
 
 from src.swarm import grounding
-from src.swarm.mailbox import Mailbox
 from src.swarm.models import (
     RunStatus,
     SwarmAgentSpec,
@@ -39,6 +38,7 @@
     topological_layers,
     validate_dag,
 )
+from src.tools.redaction import redact_internal_paths
 from src.swarm.worker import run_worker
 
 logger = logging.getLogger(__name__)
@@ -269,11 +269,12 @@ def _execute_run(
                     run.total_input_tokens += result.input_tokens
                     run.total_output_tokens += result.output_tokens
 
-                    if result.status in ("completed", "timeout", "token_limit"):
+                    if result.status == "completed":
                         task_summaries[tid] = result.summary
                         now_iso = datetime.now(timezone.utc).isoformat()
                         task_store.update_status(
-                            tid, TaskStatus.completed,
+                            tid,
+                            TaskStatus.completed,
                             summary=result.summary,
                             completed_at=now_iso,
                             artifacts=result.artifact_paths,
@@ -282,41 +283,51 @@ def _execute_run(
                         resolve_dependencies(run_dir / "tasks", tid)
                         self._emit_event(
                             run_id,
-                            self._make_event("task_completed", task_id=tid,
-                                             data={"status": result.status,
-                                                   "iterations": result.iterations,
-                                                   "input_tokens": result.input_tokens,
-                                                   "output_tokens": result.output_tokens}),
+                            self._make_event(
+                                "task_completed",
+                                task_id=tid,
+                                data={
+                                    "status": result.status,
+                                    "iterations": result.iterations,
+                                    "input_tokens": result.input_tokens,
+                                    "output_tokens": result.output_tokens,
+                                },
+                            ),
                         )
                     else:
                         all_succeeded = False
                         task_store.update_status(
-                            tid, TaskStatus.failed,
-                            error=result.error or "Unknown error",
+                            tid,
+                            TaskStatus.failed,
+                            error=redact_internal_paths(result.error)
+                            or f"worker did not complete (status={result.status})",
                             completed_at=datetime.now(timezone.utc).isoformat(),
                             worker_iterations=result.iterations,
                         )
                         self._emit_event(
                             run_id,
-                            self._make_event("task_failed", task_id=tid,
-                                             data={"error": result.error,
-                                                   "input_tokens": result.input_tokens,
-                                                   "output_tokens": result.output_tokens}),
+                            self._make_event(
+                                "task_failed",
+                                task_id=tid,
+                                data={
+                                    "error": redact_internal_paths(result.error),
+                                    "input_tokens": result.input_tokens,
+                                    "output_tokens": result.output_tokens,
+                                },
+                            ),
                         )
 
         except Exception as exc:
             logger.error("Run %s failed with exception", run_id, exc_info=True)
             all_succeeded = False
             self._emit_event(
                 run_id,
-                self._make_event("run_error", data={"error": str(exc)}),
+                self._make_event("run_error", data={"error": redact_internal_paths(str(exc))}),
             )
 
         # Finalize run
         final_status = (
-            RunStatus.cancelled if cancel_event.is_set()
-            else RunStatus.completed if all_succeeded
-            else RunStatus.failed
+            RunStatus.cancelled if cancel_event.is_set() else RunStatus.completed if all_succeeded else RunStatus.failed
         )
         run.status = final_status
         run.completed_at = datetime.now(timezone.utc).isoformat()
@@ -350,7 +361,9 @@ def _prefetch_grounding_data(self, run: SwarmRun) -> None:
         if len(symbols) > symbol_limit:
             logger.warning(
                 "grounding: limiting run %s symbols from %d to %d",
-                run.id, len(symbols), symbol_limit,
+                run.id,
+                len(symbols),
+                symbol_limit,
             )
             symbols = symbols[:symbol_limit]
 
@@ -359,7 +372,9 @@ def _prefetch_grounding_data(self, run: SwarmRun) -> None:
         except Exception:
             logger.warning(
                 "grounding: pre-fetch failed for run %s symbols=%s",
-                run.id, symbols, exc_info=True,
+                run.id,
+                symbols,
+                exc_info=True,
             )
             return
 
@@ -416,14 +431,16 @@ def _event_callback(event: SwarmEvent) -> None:
                 agent_spec = agent_map.get(task.agent_id)
                 if agent_spec is None:
                     results[tid] = WorkerResult(
-                        status="failed", summary="",
+                        status="failed",
+                        summary="",
                         error=f"Agent '{task.agent_id}' not found in preset",
                     )
                     continue
 
                 # Mark task as in_progress
                 task_store.update_status(
-                    tid, TaskStatus.in_progress,
+                    tid,
+                    TaskStatus.in_progress,
                     started_at=datetime.now(timezone.utc).isoformat(),
                 )
                 self._emit_event(
@@ -467,7 +484,8 @@ def _event_callback(event: SwarmEvent) -> None:
                     except Exception as exc:
                         logger.error("Worker for task %s raised exception", tid, exc_info=True)
                         results[tid] = WorkerResult(
-                            status="failed", summary="",
+                            status="failed",
+                            summary="",
                             error=str(exc),
                         )
             except FuturesTimeoutError:
@@ -477,10 +495,12 @@ def _event_callback(event: SwarmEvent) -> None:
                     pending.cancel()
                     logger.error(
                         "Worker for task %s exceeded layer deadline (%ds)",
-                        tid, layer_deadline,
+                        tid,
+                        layer_deadline,
                     )
                     results[tid] = WorkerResult(
-                        status="timeout", summary="",
+                        status="timeout",
+                        summary="",
                         error=f"Worker exceeded layer deadline of {layer_deadline}s",
                     )
         except KeyboardInterrupt:
@@ -539,13 +559,18 @@ def _run_worker_with_retries(
                         "task_retry",
                         agent_id=agent_spec.id,
                         task_id=task.id,
-                        data={"attempt": attempt + 1, "max_retries": max_retries,
-                              "previous_error": result.error if result else None},
+                        data={
+                            "attempt": attempt + 1,
+                            "max_retries": max_retries,
+                            "previous_error": result.error if result else None,
+                        },
                     ),
                 )
                 logger.info(
                     "Retrying task %s (attempt %d/%d)",
-                    task.id, attempt + 1, max_retries + 1,
+                    task.id,
+                    attempt + 1,
+                    max_retries + 1,
                 )
 
             result = run_worker(
@@ -564,18 +589,22 @@ def _run_worker_with_retries(
 
             if result.status != "failed":
                 # Success (or timeout/token_limit/completed) — no more retries
-                result = result.model_copy(update={
-                    "input_tokens": cumulative_input_tokens,
-                    "output_tokens": cumulative_output_tokens,
-                })
+                result = result.model_copy(
+                    update={
+                        "input_tokens": cumulative_input_tokens,
+                        "output_tokens": cumulative_output_tokens,
+                    }
+                )
                 return result
 
         # All retries exhausted, return the last failed result with cumulative tokens
         if result is not None:
-            result = result.model_copy(update={
-                "input_tokens": cumulative_input_tokens,
-                "output_tokens": cumulative_output_tokens,
-            })
+            result = result.model_copy(
+                update={
+                    "input_tokens": cumulative_input_tokens,
+                    "output_tokens": cumulative_output_tokens,
+                }
+            )
         return result  # type: ignore[return-value]
 
     def _cancel_remaining_tasks(

agent/src/swarm/serialization.py

@@ -0,0 +1,47 @@
+"""Shared serialization helpers for the swarm read boundaries.
+
+Single source of truth for projecting a :class:`SwarmTask` into the per-task
+JSON dict returned by the MCP tools (``run_swarm`` / ``get_swarm_status`` /
+``get_run_result``) and the in-process ``run_swarm`` agent tool.
+
+Before this module each boundary hand-maintained its own field allowlist and
+all three silently omitted ``SwarmTask.error``: a misconfigured provider
+produced ``status="failed"`` with no diagnosable reason anywhere the caller
+could see, even though the error was captured on disk (see P04).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from src.tools.redaction import redact_internal_paths
+
+
+def serialize_task(task: Any) -> dict:
+    """Project a SwarmTask into its public per-task dict.
+
+    ``error`` and ``iterations`` are always included so a failed or degraded
+    task is diagnosable from every read path, not only the on-disk artifacts.
+    """
+    status = task.status.value if hasattr(task.status, "value") else str(task.status)
+    return {
+        "id": task.id,
+        "agent_id": task.agent_id,
+        "status": status,
+        "summary": task.summary,
+        "iterations": getattr(task, "worker_iterations", 0),
+        "error": redact_internal_paths(getattr(task, "error", None)) or None,
+    }
+
+
+def run_level_error(run: Any) -> str | None:
+    """First failed task's error, for a top-level ``error`` field.
+
+    Returns ``None`` (an explicit null, not an absent key) when no task carries
+    an error, so a caller that only reads the top level still gets a signal.
+    """
+    for task in getattr(run, "tasks", None) or []:
+        err = getattr(task, "error", None)
+        if err:
+            return f"{task.id}/{task.agent_id}: {redact_internal_paths(err)}"
+    return None