diff --git a/tests/tools/test_delegate.py b/tests/tools/test_delegate.py
index cc89785eef7b..699d4d817a03 100644
--- a/tests/tools/test_delegate.py
+++ b/tests/tools/test_delegate.py
@@ -10,6 +10,7 @@
 """
 
 import json
+import io
 import os
 import threading
 import time
@@ -28,6 +29,7 @@
     _build_child_agent,
     _build_child_progress_callback,
     _build_child_system_prompt,
+    _run_single_child,
     _extract_output_tail,
     _strip_blocked_tools,
     _resolve_child_credential_pool,
@@ -2827,5 +2829,423 @@ def test_child_gets_no_fallback_when_parent_chain_empty(self):
         self.assertIsNone(kwargs["fallback_model"])
 
 
+# =========================================================================
+# Codex exec delegation tests
+# =========================================================================
+
+class TestCodexExecDelegation(unittest.TestCase):
+    """Tests for the codex exec --json delegation path in _run_single_child."""
+
+    def _make_codex_child(self, cwd="/tmp/test"):
+        """Build a mock child that looks like a codex ACP child."""
+        child = MagicMock()
+        child.acp_command = "codex"
+        child.acp_args = ["--acp", "--stdio"]
+        child._subagent_id = "sa-0-test1234"
+        child._delegate_depth = 1
+        child._parent_subagent_id = None
+        child._delegate_role = "leaf"
+        child.session_prompt_tokens = 0
+        child.session_completion_tokens = 0
+        child.model = "codex"
+        # Provide a cwd for the subprocess
+        child._acp_cwd = cwd
+        return child
+
+    def _make_codex_jsonl(self, *events):
+        """Build a JSONL string from event dicts."""
+        return "\n".join(json.dumps(e) for e in events) + "\n"
+
+    # ------------------------------------------------------------------
+    # Test 1: command-start event arrives before process completion
+    # ------------------------------------------------------------------
+    @patch("subprocess.Popen")
+    def test_command_start_relays_progress(self, mock_popen):
+        """item.started command_execution → child_progress_cb receives message."""
+        child = self._make_codex_child()
+        progress_calls = []
+        child.tool_progress_callback = lambda et, tn=None, pv=None, *a, **kw: progress_calls.append((et, pv, kw))
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+
+        jsonl = self._make_codex_jsonl(
+            {"type": "thread.started", "thread_id": "t1"},
+            {"type": "turn.started"},
+            {"type": "item.started", "item": {
+                "id": "item_1", "type": "command_execution",
+                "command": "pytest -q", "status": "in_progress",
+            }},
+            {"type": "item.completed", "item": {
+                "id": "item_1", "type": "command_execution",
+                "command": "pytest -q", "exit_code": 0, "status": "completed",
+            }},
+            {"type": "item.completed", "item": {
+                "id": "item_2", "type": "agent_message",
+                "text": "All tests passed.",
+            }},
+            {"type": "turn.completed"},
+        )
+        mock_proc = MagicMock()
+        mock_proc.stdout = io.StringIO(jsonl)
+        mock_proc.stderr = io.StringIO("")
+        mock_proc.poll.return_value = None
+        mock_proc.returncode = 0
+        mock_popen.return_value = mock_proc
+
+        from tools.delegate_tool import _run_codex_exec
+        _run_codex_exec("test goal", child.tool_progress_callback, child, 0, time.monotonic())
+
+        # Check command-start was relayed
+        command_starts = [kw.get("preview") for et, pv, kw in progress_calls if et == "subagent.text" and "Codex is running" in (kw.get("preview") or "")]
+        self.assertTrue(len(command_starts) > 0, "command_start should relay progress")
+        self.assertIn("pytest", command_starts[0])
+
+    # ------------------------------------------------------------------
+    # Test 2: command completion reports success/failure
+    # ------------------------------------------------------------------
+    @patch("subprocess.Popen")
+    def test_command_completed_success(self, mock_popen):
+        """item.completed command_execution status=completed → success message."""
+        child = self._make_codex_child()
+        progress_calls = []
+        child.tool_progress_callback = lambda et, tn=None, pv=None, *a, **kw: progress_calls.append((et, pv, kw))
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+
+        jsonl = self._make_codex_jsonl(
+            {"type": "thread.started", "thread_id": "t1"},
+            {"type": "turn.started"},
+            {"type": "item.completed", "item": {
+                "id": "item_1", "type": "command_execution",
+                "command": "ls", "exit_code": 0, "status": "completed",
+            }},
+            {"type": "item.completed", "item": {
+                "id": "item_2", "type": "agent_message", "text": "ok",
+            }},
+            {"type": "turn.completed"},
+        )
+        mock_proc = MagicMock()
+        mock_proc.stdout = io.StringIO(jsonl)
+        mock_proc.stderr = io.StringIO("")
+        mock_proc.poll.return_value = None
+        mock_proc.returncode = 0
+        mock_popen.return_value = mock_proc
+
+        from tools.delegate_tool import _run_codex_exec
+        _run_codex_exec("test", child.tool_progress_callback, child, 0, time.monotonic())
+
+        completes = [kw.get("preview") for et, pv, kw in progress_calls if et == "subagent.text" and "completed" in (kw.get("preview") or "").lower()]
+        self.assertTrue(len(completes) > 0, "command success should relay completed message")
+
+    @patch("subprocess.Popen")
+    def test_command_completed_failure(self, mock_popen):
+        """item.completed command_execution status=failed → failure message with exit code."""
+        child = self._make_codex_child()
+        progress_calls = []
+        child.tool_progress_callback = lambda et, tn=None, pv=None, *a, **kw: progress_calls.append((et, pv, kw))
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+
+        jsonl = self._make_codex_jsonl(
+            {"type": "thread.started", "thread_id": "t1"},
+            {"type": "turn.started"},
+            {"type": "item.completed", "item": {
+                "id": "item_1", "type": "command_execution",
+                "command": "false", "exit_code": 1, "status": "failed",
+            }},
+            {"type": "item.completed", "item": {
+                "id": "item_2", "type": "agent_message", "text": "command failed",
+            }},
+            {"type": "turn.completed"},
+        )
+        mock_proc = MagicMock()
+        mock_proc.stdout = io.StringIO(jsonl)
+        mock_proc.stderr = io.StringIO("")
+        mock_proc.poll.return_value = None
+        mock_proc.returncode = 0
+        mock_popen.return_value = mock_proc
+
+        from tools.delegate_tool import _run_codex_exec
+        result = _run_codex_exec("test", child.tool_progress_callback, child, 0, time.monotonic())
+
+        failures = [kw.get("preview") for et, pv, kw in progress_calls if et == "subagent.text" and "failed" in (kw.get("preview") or "").lower()]
+        self.assertTrue(len(failures) > 0, "command failure should relay failed message")
+        self.assertIn("exit 1", failures[0].lower())
+
+    # ------------------------------------------------------------------
+    # Test 3: file_change produces concise interim message
+    # ------------------------------------------------------------------
+    @patch("subprocess.Popen")
+    def test_file_change_relays_message(self, mock_popen):
+        """item.completed file_change → concise list of changed files."""
+        child = self._make_codex_child()
+        progress_calls = []
+        child.tool_progress_callback = lambda et, tn=None, pv=None, *a, **kw: progress_calls.append((et, pv, kw))
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+
+        jsonl = self._make_codex_jsonl(
+            {"type": "thread.started", "thread_id": "t1"},
+            {"type": "turn.started"},
+            {"type": "item.completed", "item": {
+                "id": "item_1", "type": "file_change",
+                "changes": [
+                    {"path": "/abs/path/src/auth.py", "kind": "update"},
+                    {"path": "/abs/path/tests/test_auth.py", "kind": "update"},
+                ],
+                "status": "completed",
+            }},
+            {"type": "item.completed", "item": {
+                "id": "item_2", "type": "agent_message", "text": "done",
+            }},
+            {"type": "turn.completed"},
+        )
+        mock_proc = MagicMock()
+        mock_proc.stdout = io.StringIO(jsonl)
+        mock_proc.stderr = io.StringIO("")
+        mock_proc.poll.return_value = None
+        mock_proc.returncode = 0
+        mock_popen.return_value = mock_proc
+
+        from tools.delegate_tool import _run_codex_exec
+        _run_codex_exec("test", child.tool_progress_callback, child, 0, time.monotonic())
+
+        file_msgs = [kw.get("preview") for et, pv, kw in progress_calls if et == "subagent.text" and "Codex changed" in (kw.get("preview") or "")]
+        self.assertTrue(len(file_msgs) > 0, "file_change should relay progress")
+        self.assertIn("auth.py", file_msgs[0])
+        self.assertIn("test_auth.py", file_msgs[0])
+
+    # ------------------------------------------------------------------
+    # Test 4: agent_message relayed and returned as final summary
+    # ------------------------------------------------------------------
+    @patch("subprocess.Popen")
+    def test_agent_message_is_summary(self, mock_popen):
+        """agent_message → relayed + returned as final summary, not duplicated."""
+        child = self._make_codex_child()
+        progress_calls = []
+        child.tool_progress_callback = lambda et, tn=None, pv=None, *a, **kw: progress_calls.append((et, pv, kw))
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+
+        jsonl = self._make_codex_jsonl(
+            {"type": "thread.started", "thread_id": "t1"},
+            {"type": "turn.started"},
+            {"type": "item.completed", "item": {
+                "id": "item_1", "type": "agent_message", "text": "First message",
+            }},
+            {"type": "item.completed", "item": {
+                "id": "item_2", "type": "agent_message",
+                "text": "Final answer: task completed.",
+            }},
+            {"type": "turn.completed"},
+        )
+        mock_proc = MagicMock()
+        mock_proc.stdout = io.StringIO(jsonl)
+        mock_proc.stderr = io.StringIO("")
+        mock_proc.poll.return_value = None
+        mock_proc.returncode = 0
+        mock_popen.return_value = mock_proc
+
+        from tools.delegate_tool import _run_codex_exec
+        result = _run_codex_exec("test", child.tool_progress_callback, child, 0, time.monotonic())
+
+        # Both messages should be relayed
+        text_msgs = [kw.get("preview") for et, pv, kw in progress_calls if et == "subagent.text" and kw.get("preview")]
+        self.assertGreaterEqual(len(text_msgs), 2)
+
+        # Final summary should be the LAST agent_message
+        self.assertEqual(result["summary"], "Final answer: task completed.")
+        self.assertEqual(result["status"], "completed")
+
+    # ------------------------------------------------------------------
+    # Test 5a: turn.failed → failed result
+    # ------------------------------------------------------------------
+    @patch("subprocess.Popen")
+    def test_turn_failed_produces_failed_result(self, mock_popen):
+        """turn.failed event → failed delegation result with error message."""
+        child = self._make_codex_child()
+        progress_calls = []
+        child.tool_progress_callback = lambda et, tn=None, pv=None, *a, **kw: progress_calls.append((et, pv, kw))
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+
+        jsonl = self._make_codex_jsonl(
+            {"type": "thread.started", "thread_id": "t1"},
+            {"type": "turn.started"},
+            {"type": "turn.failed", "error": {"message": "Auth token expired"}},
+        )
+        mock_proc = MagicMock()
+        mock_proc.stdout = io.StringIO(jsonl)
+        mock_proc.stderr = io.StringIO("")
+        mock_proc.poll.return_value = None
+        mock_proc.returncode = 1
+        mock_popen.return_value = mock_proc
+
+        from tools.delegate_tool import _run_codex_exec
+        result = _run_codex_exec("test", child.tool_progress_callback, child, 0, time.monotonic())
+
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("Auth token expired", result["error"])
+
+    # ------------------------------------------------------------------
+    # Test 5b: top-level error → failed result
+    # ------------------------------------------------------------------
+    @patch("subprocess.Popen")
+    def test_top_level_error_produces_failed_result(self, mock_popen):
+        """Top-level error event → failed delegation result."""
+        child = self._make_codex_child()
+        progress_calls = []
+        child.tool_progress_callback = lambda et, tn=None, pv=None, *a, **kw: progress_calls.append((et, pv, kw))
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+
+        jsonl = self._make_codex_jsonl(
+            {"type": "error", "message": "codex not found"},
+        )
+        mock_proc = MagicMock()
+        mock_proc.stdout = io.StringIO(jsonl)
+        mock_proc.stderr = io.StringIO("")
+        mock_proc.poll.return_value = None
+        mock_proc.returncode = 1
+        mock_popen.return_value = mock_proc
+
+        from tools.delegate_tool import _run_codex_exec
+        result = _run_codex_exec("test", child.tool_progress_callback, child, 0, time.monotonic())
+
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("codex not found", result["error"])
+
+    # ------------------------------------------------------------------
+    # Test 6: malformed/unknown JSONL does not crash
+    # ------------------------------------------------------------------
+    @patch("subprocess.Popen")
+    def test_malformed_jsonl_does_not_crash(self, mock_popen):
+        """Malformed lines and unknown events are skipped without crashing."""
+        child = self._make_codex_child()
+        progress_calls = []
+        child.tool_progress_callback = lambda et, tn=None, pv=None, *a, **kw: progress_calls.append((et, pv, kw))
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+
+        # Mix of malformed, unknown, and valid events
+        lines = [
+            "not json at all",
+            "",
+            json.dumps({"type": "unknown_event", "data": "ignored"}),
+            json.dumps({"type": "thread.started", "thread_id": "t1"}),
+            json.dumps({"type": "turn.started"}),
+            json.dumps({"type": "item.completed", "item": {
+                "id": "i1", "type": "agent_message", "text": "still works",
+            }}),
+            json.dumps({"type": "turn.completed"}),
+            "{broken json",
+            "",
+        ]
+        jsonl = "\n".join(lines) + "\n"
+
+        mock_proc = MagicMock()
+        mock_proc.stdout = io.StringIO(jsonl)
+        mock_proc.stderr = io.StringIO("")
+        mock_proc.poll.return_value = None
+        mock_proc.returncode = 0
+        mock_popen.return_value = mock_proc
+
+        from tools.delegate_tool import _run_codex_exec
+        result = _run_codex_exec("test", child.tool_progress_callback, child, 0, time.monotonic())
+
+        # Should complete without error
+        self.assertEqual(result["status"], "completed")
+        self.assertEqual(result["summary"], "still works")
+
+    # ------------------------------------------------------------------
+    # Test 7: non-codex delegation unchanged
+    # ------------------------------------------------------------------
+    def test_non_codex_child_uses_normal_path(self):
+        """Child without codex acp_command does NOT trigger codex exec path."""
+        child = self._make_codex_child()
+        child.acp_command = "claude"  # not codex
+        child.run_conversation = MagicMock(return_value={
+            "final_response": "done", "completed": True,
+            "api_calls": 1, "messages": [],
+        })
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+        child.get_activity_summary = MagicMock(return_value={
+            "current_tool": None, "api_call_count": 0,
+            "max_iterations": 50, "last_activity_desc": "",
+        })
+
+        from tools.delegate_tool import _run_single_child
+        result = _run_single_child(0, "test", child=child, parent_agent=None)
+
+        # Should have called run_conversation, not codex exec
+        child.run_conversation.assert_called_once()
+        self.assertEqual(result["summary"], "done")
+
+    # ------------------------------------------------------------------
+    # Test 8: subprocess args use argv list, correct cwd, --json, --sandbox
+    # ------------------------------------------------------------------
+    @patch("subprocess.Popen")
+    def test_codex_exec_uses_correct_subprocess_args(self, mock_popen):
+        """Subprocess is spawned with argv list, --json, --sandbox workspace-write."""
+        child = self._make_codex_child(cwd="/my/project")
+        child.tool_progress_callback = None
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+
+        jsonl = self._make_codex_jsonl(
+            {"type": "thread.started", "thread_id": "t1"},
+            {"type": "turn.started"},
+            {"type": "item.completed", "item": {
+                "id": "i1", "type": "agent_message", "text": "done",
+            }},
+            {"type": "turn.completed"},
+        )
+        mock_proc = MagicMock()
+        mock_proc.stdout = io.StringIO(jsonl)
+        mock_proc.stderr = io.StringIO("")
+        mock_proc.poll.return_value = None
+        mock_proc.returncode = 0
+        mock_popen.return_value = mock_proc
+
+        from tools.delegate_tool import _run_codex_exec
+        _run_codex_exec("add a function", child.tool_progress_callback, child, 0, time.monotonic())
+
+        # Verify Popen was called with argv list (not shell=True)
+        call_args, call_kwargs = mock_popen.call_args
+        cmd = call_args[0]
+        self.assertIsInstance(cmd, list, "command must be list (not shell string)")
+        self.assertIn("--json", cmd)
+        self.assertIn("--sandbox", cmd)
+        self.assertIn("workspace-write", cmd)
+        self.assertIn("add a function", cmd)
+        self.assertEqual(call_kwargs.get("cwd"), "/my/project")
+        self.assertNotEqual(call_kwargs.get("shell"), True)
+
+    # ------------------------------------------------------------------
+    # Test 9: rollback — reverting the codex branch restores normal delegation
+    # ------------------------------------------------------------------
+    def test_no_codex_child_goes_normal_path(self):
+        """Child without acp_command set uses normal delegation (not codex exec)."""
+        child = self._make_codex_child()
+        child.acp_command = None  # no ACP command at all
+        child.run_conversation = MagicMock(return_value={
+            "final_response": "normal", "completed": True,
+            "api_calls": 1, "messages": [],
+        })
+        child._delegate_saved_tool_names = []
+        child._credential_pool = None
+        child.get_activity_summary = MagicMock(return_value={
+            "current_tool": None, "api_call_count": 0,
+            "max_iterations": 50, "last_activity_desc": "",
+        })
+
+        from tools.delegate_tool import _run_single_child
+        result = _run_single_child(0, "test", child=child, parent_agent=None)
+
+        child.run_conversation.assert_called_once()
+        self.assertEqual(result["summary"], "normal")
+
+
 if __name__ == "__main__":
     unittest.main()
diff --git a/tests/tools/test_skill_manager_tool.py b/tests/tools/test_skill_manager_tool.py
index 9b4e83bbfd0f..eb1ce2da3315 100644
--- a/tests/tools/test_skill_manager_tool.py
+++ b/tests/tools/test_skill_manager_tool.py
@@ -1091,3 +1091,28 @@ def test_out_of_tree_path_refused(self, tmp_path):
         assert result["success"] is False
         assert "skills root" in result["error"].lower()
         assert outside.exists()
+
+
+class TestFindSkillFollowsSymlinks:
+    """_find_skill must follow symlinks, matching skill_view's behavior."""
+
+    def test_finds_skill_via_symlinked_directory(self, tmp_path):
+        """A skill reached through a symlink should be found, just as skill_view finds it."""
+        from tools.skill_manager_tool import _find_skill
+
+        skills = tmp_path / "local-skills"
+        skills.mkdir()
+        real_dir = tmp_path / "real-skill"
+        real_dir.mkdir()
+        (real_dir / "SKILL.md").write_text("---\nname: test-symlink\ndescription: test\n---\n# Test\n")
+        symlink = skills / "test-symlink"
+        symlink.symlink_to(real_dir, target_is_directory=True)
+
+        with patch("agent.skill_utils.get_all_skills_dirs", return_value=[skills]):
+            found = _find_skill("test-symlink")
+
+        assert found is not None, (
+            f"_find_skill should discover skill via symlink; got None. "
+            f"Symlink: {symlink} -> {real_dir}"
+        )
+        assert found["path"] == symlink
diff --git a/tools/delegate_tool.py b/tools/delegate_tool.py
index bbe79bc8512a..79b3bb1e6234 100644
--- a/tools/delegate_tool.py
+++ b/tools/delegate_tool.py
@@ -22,6 +22,7 @@
 
 logger = logging.getLogger(__name__)
 import os
+import subprocess
 import threading
 import time
 from concurrent.futures import (
@@ -1469,6 +1470,227 @@ def _w(line: str = "") -> None:
         return None
 
 
+def _run_codex_exec(
+    goal: str,
+    child_progress_cb,
+    child,
+    task_index: int,
+    child_start: float,
+) -> Dict[str, Any]:
+    """Run ``codex exec --json`` directly and relay progress events
+    through the child's existing progress callback.
+
+    This is a narrow Codex-specific execution path that bypasses
+    ``CopilotACPClient`` (which requires ``--acp --stdio``, unsupported
+    by the current ``@openai/codex`` CLI).  All non-Codex delegation
+    continues through the normal subagent path.
+    """
+    # Determine workspace directory from the child's ACP cwd, falling
+    # back to the session working directory.
+    cwd = getattr(child, "_acp_cwd", None) or os.getcwd()
+
+    cmd = [
+        "codex", "exec", "--json",
+        "--sandbox", "workspace-write",
+        goal,
+    ]
+
+    proc = subprocess.Popen(
+        cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        cwd=cwd,
+    )
+
+    final_message: Optional[str] = None
+    turn_failed = False
+    error_message: Optional[str] = None
+
+    try:
+        for line in proc.stdout:
+            line = line.strip()
+            if not line:
+                continue
+            try:
+                event = json.loads(line)
+            except json.JSONDecodeError:
+                logger.debug("Codex exec: malformed JSONL line: %s", line[:200])
+                continue
+
+            event_type = event.get("type", "")
+
+            if event_type == "item.started":
+                item = event.get("item", {})
+                if isinstance(item, dict) and item.get("type") == "command_execution":
+                    cmd_text = str(item.get("command", ""))
+                    if len(cmd_text) > 150:
+                        cmd_text = cmd_text[:147] + "..."
+                    if child_progress_cb:
+                        try:
+                            child_progress_cb(
+                                "subagent.text",
+                                preview=f"\U0001f527 Codex is running: {cmd_text}",
+                            )
+                        except Exception:
+                            pass
+
+            elif event_type == "item.completed":
+                item = event.get("item", {})
+                if not isinstance(item, dict):
+                    continue
+                item_type = item.get("type", "")
+
+                if item_type == "command_execution":
+                    status = str(item.get("status") or "")
+                    exit_code = item.get("exit_code")
+                    if status == "completed":
+                        msg = "\u2705 Codex command completed"
+                        if exit_code is not None:
+                            msg += f" (exit {exit_code})"
+                    elif status == "failed":
+                        msg = f"\u274c Codex command failed (exit {exit_code})"
+                    else:
+                        msg = f"Codex command: {status}"
+                    if child_progress_cb:
+                        try:
+                            child_progress_cb("subagent.text", preview=msg)
+                        except Exception:
+                            pass
+
+                elif item_type == "file_change":
+                    changes = item.get("changes") or []
+                    if changes:
+                        paths: list[str] = []
+                        for c in changes[:3]:
+                            p = str(c.get("path", ""))
+                            paths.append(os.path.basename(p) if p else "?")
+                        suffix = (
+                            f" +{len(changes) - 3} more"
+                            if len(changes) > 3
+                            else ""
+                        )
+                        if child_progress_cb:
+                            try:
+                                child_progress_cb(
+                                    "subagent.text",
+                                    preview=(
+                                        "\u270f\ufe0f Codex changed: "
+                                        + ", ".join(paths)
+                                        + suffix
+                                    ),
+                                )
+                            except Exception:
+                                pass
+
+                elif item_type == "agent_message":
+                    text = str(item.get("text") or "")
+                    if text:
+                        final_message = text
+                        if child_progress_cb:
+                            try:
+                                child_progress_cb("subagent.text", preview=text)
+                            except Exception:
+                                pass
+
+            elif event_type == "turn.failed":
+                turn_failed = True
+                err_info = event.get("error", {})
+                if isinstance(err_info, dict):
+                    error_message = str(err_info.get("message") or "Codex turn failed")
+                else:
+                    error_message = str(err_info or "Codex turn failed")
+                if child_progress_cb:
+                    try:
+                        child_progress_cb(
+                            "subagent.text",
+                            preview=f"\u274c Codex turn failed: {error_message}",
+                        )
+                    except Exception:
+                        pass
+
+            elif event_type == "error":
+                error_message = str(event.get("message") or "Unknown Codex error")
+                if child_progress_cb:
+                    try:
+                        child_progress_cb(
+                            "subagent.text",
+                            preview=f"\u274c Codex error: {error_message}",
+                        )
+                    except Exception:
+                        pass
+
+            # thread.started, turn.started, turn.completed, and unknown
+            # event types are intentionally ignored.
+    finally:
+        # Ensure the subprocess is cleaned up even on interrupt.
+        _cleanup_codex_proc(proc)
+
+    duration = round(time.monotonic() - child_start, 2)
+
+    if turn_failed or error_message:
+        return {
+            "task_index": task_index,
+            "status": "failed",
+            "summary": error_message or "Codex task failed",
+            "error": error_message or "Codex task failed",
+            "exit_reason": "error",
+            "api_calls": 0,
+            "duration_seconds": duration,
+            "model": "codex",
+            "tokens": {"input": 0, "output": 0},
+            "tool_trace": [],
+            "messages": [],
+        }
+
+    if final_message:
+        return {
+            "task_index": task_index,
+            "status": "completed",
+            "summary": final_message,
+            "api_calls": 0,
+            "duration_seconds": duration,
+            "model": "codex",
+            "exit_reason": "completed",
+            "tokens": {"input": 0, "output": 0},
+            "tool_trace": [],
+            "messages": [],
+        }
+
+    return {
+        "task_index": task_index,
+        "status": "completed",
+        "summary": "Codex completed the task.",
+        "api_calls": 0,
+        "duration_seconds": duration,
+        "model": "codex",
+        "exit_reason": "completed",
+        "tokens": {"input": 0, "output": 0},
+        "tool_trace": [],
+        "messages": [],
+    }
+
+
+def _cleanup_codex_proc(proc: subprocess.Popen) -> None:
+    """Terminate a codex subprocess, waiting briefly for graceful exit."""
+    if proc.poll() is not None:
+        return  # Already exited
+    try:
+        proc.terminate()
+        proc.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        try:
+            proc.kill()
+            proc.wait(timeout=2)
+        except Exception:
+            pass
+    except Exception:
+        try:
+            proc.kill()
+        except Exception:
+            pass
+
+
 def _run_single_child(
     task_index: int,
     goal: str,
@@ -1624,10 +1846,23 @@ def _heartbeat_loop():
             except Exception as e:
                 logger.debug("Progress callback start failed: %s", e)
 
+        # ── Codex exec fast path ───────────────────────────────────────
+        # When the delegated command is exactly "codex", bypass the normal
+        # subagent / CopilotACPClient path and run codex exec --json
+        # directly in-process.  This gives us structured JSONL events that
+        # we relay as interim messages through the existing progress
+        # callback, without building a new transport abstraction.
+        if getattr(child, "acp_command", None) == "codex":
+            try:
+                return _run_codex_exec(
+                    goal, child_progress_cb, child, task_index, child_start
+                )
+            finally:
+                _heartbeat_stop.set()
+                if _subagent_id:
+                    _unregister_subagent(_subagent_id)
+
         # File-state coordination: reuse the stable subagent_id as the child's
-        # task_id so file_state writes, active-subagents registry, and TUI
-        # events all share one key.  Falls back to a fresh uuid only if the
-        # pre-built id is somehow missing.
         import uuid as _uuid
 
         child_task_id = _subagent_id or f"subagent-{task_index}-{_uuid.uuid4().hex[:8]}"
diff --git a/tools/skill_manager_tool.py b/tools/skill_manager_tool.py
index 3a6f315b2242..92cf4a6b3785 100644
--- a/tools/skill_manager_tool.py
+++ b/tools/skill_manager_tool.py
@@ -430,11 +430,11 @@ def _find_skill(name: str) -> Optional[Dict[str, Any]]:
     external dirs configured via skills.external_dirs.  Returns
     {"path": Path} or None.
     """
-    from agent.skill_utils import get_all_skills_dirs, is_excluded_skill_path
+    from agent.skill_utils import get_all_skills_dirs, is_excluded_skill_path, iter_skill_index_files
     for skills_dir in get_all_skills_dirs():
         if not skills_dir.exists():
             continue
-        for skill_md in skills_dir.rglob("SKILL.md"):
+        for skill_md in iter_skill_index_files(skills_dir, "SKILL.md"):
             if is_excluded_skill_path(skill_md):
                 continue
             if skill_md.parent.name == name: