fix: address review findings — backward compat, ACP safety, result timing, security

- Fix 1: tasks_md_sync only sets MACP fields when columns exist in table headers
- Fix 2: ACP dispatch now escalates instead of falsely completing
- Fix 3: Removed premature collect_result() from dispatch_task()
- Fix 4: Yolo brief staged via temp file (0600) instead of process args
- Fix 5: cleanup_worktree validates path against configured worktree base

tools/orchestrator-matrix/controller/mosaic_orchestrator.py

@@ -225,6 +225,24 @@ def run_single_task(repo_root: pathlib.Path, orch_dir: pathlib.Path, config: dic
         rc, output, timed_out = run_shell(cmd, repo_root, log_path, timeout_sec)
 
     if rc != 0:
+        if is_macp_task(task) and str(task.get("status") or "") == "escalated":
+            task["failed_at"] = str(task.get("failed_at") or now_iso())
+            emit_event(
+                events_path,
+                "task.escalated",
+                task_id,
+                "escalated",
+                "controller",
+                str(task.get("escalation_reason") or task.get("error") or "Task requires human intervention."),
+            )
+            save_json(tasks_path, {"tasks": task_items})
+            state["running_task_id"] = None
+            state["updated_at"] = now_iso()
+            save_json(state_path, state)
+            macp_dispatcher.collect_result(task, rc, [], orch_dir)
+            if bool(config.get("macp", {}).get("cleanup_worktrees", True)):
+                macp_dispatcher.cleanup_worktree(task, config)
+            return True
         if not task.get("error"):
             task["error"] = f"Worker command timed out after {timeout_sec}s" if timed_out else f"Worker command failed with exit code {rc}"
         if attempt < max_attempts:
@@ -247,9 +265,10 @@ def run_single_task(repo_root: pathlib.Path, orch_dir: pathlib.Path, config: dic
         state["updated_at"] = now_iso()
         save_json(state_path, state)
         if is_macp_task(task):
-            macp_dispatcher.collect_result(task, rc, [], orch_dir)
-            if task["status"] == "failed" and bool(config.get("macp", {}).get("cleanup_worktrees", True)):
-                macp_dispatcher.cleanup_worktree(task)
+            if task["status"] == "failed":
+                macp_dispatcher.collect_result(task, rc, [], orch_dir)
+                if bool(config.get("macp", {}).get("cleanup_worktrees", True)):
+                    macp_dispatcher.cleanup_worktree(task, config)
         else:
             save_json(
                 results_dir / f"{task_id}.json",
@@ -269,7 +288,7 @@ def run_single_task(repo_root: pathlib.Path, orch_dir: pathlib.Path, config: dic
         save_json(state_path, state)
         macp_dispatcher.collect_result(task, rc, [], orch_dir)
         if bool(config.get("macp", {}).get("cleanup_worktrees", True)):
-            macp_dispatcher.cleanup_worktree(task)
+            macp_dispatcher.cleanup_worktree(task, config)
         return True
 
     task["status"] = "gated"
@@ -332,9 +351,10 @@ def run_single_task(repo_root: pathlib.Path, orch_dir: pathlib.Path, config: dic
     state["updated_at"] = now_iso()
     save_json(state_path, state)
     if is_macp_task(task):
-        macp_dispatcher.collect_result(task, rc, gate_results, orch_dir)
-        if task["status"] in {"completed", "escalated"} and bool(config.get("macp", {}).get("cleanup_worktrees", True)):
-            macp_dispatcher.cleanup_worktree(task)
+        if task["status"] in {"completed", "failed", "escalated"}:
+            macp_dispatcher.collect_result(task, rc, gate_results, orch_dir)
+            if bool(config.get("macp", {}).get("cleanup_worktrees", True)):
+                macp_dispatcher.cleanup_worktree(task, config)
     else:
         save_json(
             results_dir / f"{task_id}.json",

tools/orchestrator-matrix/controller/tasks_md_sync.py

@@ -35,9 +35,9 @@ def split_pipe_row(line: str) -> list[str]:
     return [c.strip() for c in row.split("|")]
 
 
-def parse_tasks_markdown(path: pathlib.Path) -> list[dict[str, str]]:
+def parse_tasks_markdown(path: pathlib.Path) -> tuple[set[str], list[dict[str, str]]]:
     if not path.exists():
-        return []
+        return set(), []
     lines = path.read_text(encoding="utf-8").splitlines()
 
     header_idx = -1
@@ -51,7 +51,7 @@ def parse_tasks_markdown(path: pathlib.Path) -> list[dict[str, str]]:
             headers = cells
             break
     if header_idx < 0:
-        return []
+        return set(), []
 
     rows: list[dict[str, str]] = []
     for line in lines[header_idx + 2 :]:
@@ -67,7 +67,7 @@ def parse_tasks_markdown(path: pathlib.Path) -> list[dict[str, str]]:
         if not task_id or task_id.lower() == "id":
             continue
         rows.append(row)
-    return rows
+    return set(headers), rows
 
 
 def map_status(raw: str) -> str:
@@ -93,6 +93,7 @@ def parse_depends(raw: str) -> list[str]:
 
 def build_task(
     row: dict[str, str],
+    headers: set[str],
     existing: dict[str, Any],
     macp_defaults: dict[str, str],
     runtime_default: str,
@@ -114,13 +115,25 @@ def build_task(
     task["description"] = description
     task["status"] = map_status(row.get("status", "pending"))
     task["depends_on"] = depends_on
-    task["type"] = task_type or str(task.get("type") or macp_defaults.get("type") or "coding")
-    task["dispatch"] = dispatch or str(task.get("dispatch") or macp_defaults.get("dispatch") or "")
-    task["runtime"] = runtime or str(task.get("runtime") or macp_defaults.get("runtime") or runtime_default or "codex")
-    task["branch"] = branch or str(task.get("branch") or macp_defaults.get("branch") or "")
     task["issue"] = issue or str(task.get("issue") or "")
     task["command"] = str(task.get("command") or "")
     task["quality_gates"] = task.get("quality_gates") or []
+    if "type" in headers:
+        task["type"] = task_type or str(task.get("type") or macp_defaults.get("type") or "coding")
+    else:
+        task.pop("type", None)
+    if "dispatch" in headers:
+        task["dispatch"] = dispatch or str(task.get("dispatch") or macp_defaults.get("dispatch") or "")
+    else:
+        task.pop("dispatch", None)
+    if "runtime" in headers:
+        task["runtime"] = runtime or str(task.get("runtime") or macp_defaults.get("runtime") or runtime_default or "codex")
+    else:
+        task.pop("runtime", None)
+    if "branch" in headers:
+        task["branch"] = branch or str(task.get("branch") or macp_defaults.get("branch") or "")
+    else:
+        task.pop("branch", None)
     metadata = dict(task.get("metadata") or {})
     metadata.update(
         {
@@ -166,7 +179,7 @@ def main() -> int:
         "branch": "",
     }
 
-    rows = parse_tasks_markdown(docs_path)
+    headers, rows = parse_tasks_markdown(docs_path)
     try:
         source_path = str(docs_path.relative_to(repo))
     except ValueError:
@@ -187,6 +200,7 @@ def main() -> int:
         out_tasks.append(
             build_task(
                 row,
+                headers,
                 existing_by_id.get(task_id, {}),
                 macp_defaults,
                 runtime_default,