diff --git a/README.md b/README.md
index 68366e8d..3b94c399 100644
--- a/README.md
+++ b/README.md
@@ -79,13 +79,77 @@
 
 ## Quick start
 
-### Step 1: Install ccb (MetaInfer currently uses the open-source Claude Code CLI; other coding agents are not yet supported — contributions welcome)
+### Step 1: Install a coding agent CLI
+
+MetaInfer launches phase-specific coding agents through a shared
+`SubAgentManager`. The backend is selected by `METAINFER_AGENT_BACKEND`.
+
+#### Claude/ccb backend
+
+Install ccb if you want to use the Claude-compatible backend:
 
-Open-source ccb repository: https://github.com/claude-code-best/claude-code
 ```
 npm i -g claude-code-best
 ```
 
+Run MetaInfer with:
+
+```bash
+METAINFER_AGENT_BACKEND=claude ./serve.py
+```
+
+#### Codex backend
+
+Install the Codex CLI and configure it before starting MetaInfer. MetaInfer
+does not store or pass API keys; it invokes `codex exec` and reuses your local
+Codex login, environment, and config such as `$CODEX_HOME/config.toml`.
+
+```bash
+codex login
+codex doctor
+```
+
+Then start MetaInfer with the Codex backend:
+
+```bash
+METAINFER_AGENT_BACKEND=codex ./serve.py
+```
+
+If the Codex binary is not named `codex` or is not on `PATH`, set:
+
+```bash
+METAINFER_AGENT_BACKEND=codex METAINFER_CODEX_BIN=/path/to/codex ./serve.py
+```
+
+#### pi backend
+
+Install the [pi](https://pi.dev) coding agent CLI and configure a provider
+before starting MetaInfer. MetaInfer does not store or pass API keys; it
+invokes `pi --mode json -p` and reuses your local `~/.pi` settings and
+environment variables (`PI_PROVIDER`, `PI_MODEL`, provider API keys, etc.).
+
+```bash
+npm i -g @earendil-works/pi-coding-agent
+pi auth status   # verify a provider is configured
+```
+
+Then start MetaInfer with the pi backend:
+
+```bash
+METAINFER_AGENT_BACKEND=pi ./serve.py
+```
+
+If the pi binary is not named `pi` or is not on `PATH`, set:
+
+```bash
+METAINFER_AGENT_BACKEND=pi METAINFER_PI_BIN=/path/to/pi ./serve.py
+```
+
+The pi backend maps the orchestrator's `effort` knob onto pi's `--thinking`
+level (`low` / `medium` / `high` / `max`). Session continuation uses
+`--session <id> --continue`; the session id is captured from pi's leading
+`session` event so later turns can resume it.
+
 ### Step 2: Install MetaInfer
 ```bash
 git clone https://github.com/MetaInfer/MetaInfer.git
diff --git a/README_CN.md b/README_CN.md
index 7654c80a..bd05ef5d 100644
--- a/README_CN.md
+++ b/README_CN.md
@@ -79,13 +79,76 @@
 
 ## 快速开始
 
-### 第一步，安装ccb。（本项目目前使开源的claude code版本，暂不支持其他coding agent，欢迎贡献代码以支持更多coding agent）
+### 第一步，安装一个 coding agent CLI
+
+MetaInfer 通过统一的 `SubAgentManager` 拉起各阶段 coding agent。
+后端通过 `METAINFER_AGENT_BACKEND` 选择。
+
+#### Claude/ccb 后端
+
+如果使用 Claude-compatible 后端，先安装 ccb：
 
-开源ccb项目地址：https://github.com/claude-code-best/claude-code
 ```
 npm i -g claude-code-best
 ```
 
+启动时指定：
+
+```bash
+METAINFER_AGENT_BACKEND=claude ./serve.py
+```
+
+#### Codex 后端
+
+如果使用 Codex 后端，需要先在本机安装并配置 Codex CLI。MetaInfer
+不会保存或传递 API key；它只调用 `codex exec`，复用你本机已有的
+Codex 登录状态、环境变量和 `$CODEX_HOME/config.toml` 等配置。
+
+```bash
+codex login
+codex doctor
+```
+
+然后用 Codex 后端启动 MetaInfer：
+
+```bash
+METAINFER_AGENT_BACKEND=codex ./serve.py
+```
+
+如果 Codex 二进制不叫 `codex`，或者不在 `PATH` 里，可以显式指定：
+
+```bash
+METAINFER_AGENT_BACKEND=codex METAINFER_CODEX_BIN=/path/to/codex ./serve.py
+```
+
+#### pi 后端
+
+如果使用 [pi](https://pi.dev) 后端，需要先在本机安装 pi 编码代理 CLI
+并配置好 provider。MetaInfer 不会保存或传递 API key；它只调用
+`pi --mode json -p`，复用你本机已有的 `~/.pi` 设置和环境变量
+（`PI_PROVIDER`、`PI_MODEL`、各 provider 的 API key 等）。
+
+```bash
+npm i -g @earendil-works/pi-coding-agent
+pi auth status   # 确认已配置某个 provider
+```
+
+然后用 pi 后端启动 MetaInfer：
+
+```bash
+METAINFER_AGENT_BACKEND=pi ./serve.py
+```
+
+如果 pi 二进制不叫 `pi`，或者不在 `PATH` 里，可以显式指定：
+
+```bash
+METAINFER_AGENT_BACKEND=pi METAINFER_PI_BIN=/path/to/pi ./serve.py
+```
+
+pi 后端会把 orchestrator 的 `effort` 旋钮映射到 pi 的 `--thinking` 等级
+（`low` / `medium` / `high` / `max`）。会话续接使用 `--session <id> --continue`；
+session id 从 pi 流首的 `session` 事件中捕获，供后续轮次续接。
+
 ### 第二步，安装MetaInfer
 ```bash
 git clone https://github.com/MetaInfer/MetaInfer.git
diff --git a/metainfer/orchestrator/_bootstrap.py b/metainfer/orchestrator/_bootstrap.py
index 22b1d68e..c8679805 100644
--- a/metainfer/orchestrator/_bootstrap.py
+++ b/metainfer/orchestrator/_bootstrap.py
@@ -174,6 +174,9 @@ def restore() -> None:
 def make_subagent_manager(
     *,
     claude_bin: str,
+    codex_bin: Optional[str] = None,
+    pi_bin: Optional[str] = None,
+    agent_backend: Optional[str] = None,
     model: Optional[str],
     permission_mode: str,
     effort: str,
@@ -181,6 +184,7 @@ def make_subagent_manager(
     snapshot_file: Path,
     max_concurrent: int = 4,
     budget: Any = None,
+    state_dir: Optional[Path] = None,
 ) -> SubAgentManager:
     """Build a SubAgentManager with the standard settings shared by every
     orchestrator. Per-orchestrator customization happens via
@@ -189,9 +193,32 @@ def make_subagent_manager(
 
     ``budget`` (optional) wires the per-task :class:`TokenBudget` so
     every agent launch is gated + every result's cost is recorded.
+
+    ``state_dir`` (optional) scopes the pi backend's session storage to a
+    per-task directory (``<state_dir>/pi-sessions``) so orchestrator
+    sessions are invisible to a user running ``pi --continue`` from their
+    shell. Other backends ignore it. See ``SubAgentManager.pi_session_dir``.
     """
+    resolved_agent_backend = (
+        agent_backend
+        or os.environ.get("METAINFER_AGENT_BACKEND")
+        or "claude"
+    )
+    resolved_codex_bin = (
+        codex_bin
+        or os.environ.get("METAINFER_CODEX_BIN")
+        or "codex"
+    )
+    resolved_pi_bin = (
+        pi_bin
+        or os.environ.get("METAINFER_PI_BIN")
+        or "pi"
+    )
     return SubAgentManager(
         claude_bin=claude_bin,
+        codex_bin=resolved_codex_bin,
+        pi_bin=resolved_pi_bin,
+        agent_backend=resolved_agent_backend,
         default_model=model,
         permission_mode=permission_mode,
         effort=effort,
@@ -199,6 +226,7 @@ def make_subagent_manager(
         snapshot_file=snapshot_file,
         max_concurrent=max_concurrent,
         budget=budget,
+        pi_session_dir=(Path(state_dir) / "pi-sessions") if state_dir else None,
     )
 
 
diff --git a/metainfer/orchestrator/subagent_manager.py b/metainfer/orchestrator/subagent_manager.py
index 3c79c722..5d8c92a1 100644
--- a/metainfer/orchestrator/subagent_manager.py
+++ b/metainfer/orchestrator/subagent_manager.py
@@ -33,6 +33,32 @@
 # Exit codes / signals that indicate infrastructure rather than logic failure.
 # 124 = timeout (coreutils convention), 137 = SIGKILL (128+9), 143 = SIGTERM (128+15).
 _INFRA_EXIT_CODES = {124, 137, 143}
+AgentBackend = Literal["claude", "codex", "pi"]
+
+
+# Claude Code "effort" levels map onto pi's ``--thinking`` levels 1:1 for
+# the overlapping set (low / medium / high / max). pi additionally offers
+# ``off`` / ``minimal`` / ``xhigh`` but those have no Claude equivalent, so
+# we pass the value through unchanged and let pi validate it.
+_PI_THINKING_ALIASES = {
+    "low": "low",
+    "medium": "medium",
+    "high": "high",
+    "max": "max",
+}
+
+
+def _normalize_agent_backend(value: str) -> AgentBackend:
+    v = (value or "claude").strip().lower()
+    if v in {"claude", "claude-code", "ccb"}:
+        return "claude"
+    if v in {"codex", "openai-codex"}:
+        return "codex"
+    if v in {"pi", "pi-coding-agent", "earendil"}:
+        return "pi"
+    raise ValueError(
+        f"invalid agent backend {value!r}; expected 'claude', 'codex', or 'pi'"
+    )
 
 
 # --------------------------------------------------------------------------- #
@@ -149,6 +175,9 @@ class SubAgentManager:
     def __init__(
         self,
         claude_bin: str = "ccb",
+        codex_bin: str = "codex",
+        pi_bin: str = "pi",
+        agent_backend: AgentBackend = "claude",
         default_model: Optional[str] = None,
         max_concurrent: int = 4,
         permission_mode: str = "bypassPermissions",
@@ -157,8 +186,22 @@ def __init__(
         snapshot_file: Optional[Path] = None,
         budget: Any = None,
         budget_source: str = "orchestrator",
+        # pi backend only: isolate orchestrator session files into a
+        # per-task directory so they are invisible to a user running
+        # ``pi --continue`` / ``pi --resume`` from their shell (which
+        # defaults to ~/.pi). Without this, pi indexes sessions by cwd,
+        # so a user cd'ing into an iteration workdir — or passing a
+        # partial session id lifted from events.jsonl — could attach to
+        # a production orchestrator session mid-flight and corrupt the
+        # context that later ``--session <id> --continue`` turns rely on.
+        # When set, ``_build_pi_command`` prepends ``--session-dir`` so
+        # both the initial launch and every resume read/write here.
+        pi_session_dir: Optional[Path] = None,
     ) -> None:
+        self.agent_backend = _normalize_agent_backend(agent_backend)
         self.claude_bin = claude_bin
+        self.codex_bin = codex_bin
+        self.pi_bin = pi_bin
         self.default_model = default_model
         self.max_concurrent = max_concurrent
         # Per-task token / cost budget (TokenBudget instance or None to
@@ -208,6 +251,9 @@ def __init__(
         self.snapshot_file: Optional[Path] = (
             Path(snapshot_file) if snapshot_file else None
         )
+        self.pi_session_dir: Optional[Path] = (
+            Path(pi_session_dir).resolve() if pi_session_dir else None
+        )
         self._handles: Dict[str, AgentHandle] = {}
         self._results: Dict[str, AgentResult] = {}
         self._ctrl_lock = threading.Lock()
@@ -616,6 +662,13 @@ def dump_snapshot(self) -> None:
     # ------------------------------------------------------------------ #
 
     def _build_command(self, spec: AgentSpec) -> List[str]:
+        if self.agent_backend == "codex":
+            return self._build_codex_command(spec)
+        if self.agent_backend == "pi":
+            return self._build_pi_command(spec)
+        return self._build_claude_command(spec)
+
+    def _build_claude_command(self, spec: AgentSpec) -> List[str]:
         cmd = [
             self.claude_bin,
             "-p",  # print (non-interactive) mode; prompt read from stdin
@@ -658,6 +711,86 @@ def _build_command(self, spec: AgentSpec) -> List[str]:
         cmd += list(spec.extra_args)
         return cmd
 
+    def _build_codex_command(self, spec: AgentSpec) -> List[str]:
+        # Codex CLI owns auth + user configuration. Do not pass API keys
+        # or --ignore-user-config; child processes inherit the current
+        # CODEX_HOME / OPENAI_* environment exactly like a normal user run.
+        model = spec.model or self.default_model
+        if spec.resume_session_id:
+            cmd = [
+                self.codex_bin,
+                "exec",
+                "resume",
+                "--json",
+                "--skip-git-repo-check",
+            ]
+            if model:
+                cmd += ["--model", model]
+            cmd += list(spec.extra_args)
+            cmd += [spec.resume_session_id, "-"]
+        else:
+            cmd = [
+                self.codex_bin,
+                "exec",
+                "--json",
+                "--color", "never",
+                "--skip-git-repo-check",
+                "--sandbox", "workspace-write",
+                "-C", str(spec.workdir),
+                "--add-dir", "/tmp",
+            ]
+            for d in self.extra_add_dirs:
+                cmd += ["--add-dir", str(d)]
+            if model:
+                cmd += ["--model", model]
+            cmd += list(spec.extra_args)
+            cmd += ["-"]
+        return cmd
+
+    def _build_pi_command(self, spec: AgentSpec) -> List[str]:
+        # pi (https://pi.dev) owns auth + user configuration via ~/.pi
+        # settings and environment variables (PI_PROVIDER, PI_MODEL,
+        # provider API keys, etc.). We invoke `pi --mode json -p` so the
+        # orchestrator gets a stream of JSONL events on stdout (captured
+        # into events.jsonl just like claude/codex), with the prompt piped
+        # via stdin — pi merges piped stdin into the initial user message.
+        #
+        # pi has no ``--add-dir`` equivalent: its built-in tools (read,
+        # bash, edit, write, grep, find, ls) can already reach any path,
+        # so ``extra_add_dirs`` is a no-op here (cwd scoping via Popen is
+        # enough). We keep the param for interface parity.
+        model = spec.model or self.default_model
+        cmd = [
+            self.pi_bin,
+            "--mode", "json",
+            "-p",  # print (non-interactive) mode; prompt read from stdin
+        ]
+        # Isolate orchestrator sessions from the user's ~/.pi index so a
+        # production task can't be hijacked by ``pi --continue`` / ``pi
+        # --resume`` / ``pi --session <partial-id>`` from a shell sharing
+        # the same cwd. See __init__ docstring on ``pi_session_dir``.
+        if self.pi_session_dir is not None:
+            self.pi_session_dir.mkdir(parents=True, exist_ok=True)
+            cmd += ["--session-dir", str(self.pi_session_dir)]
+        if model:
+            cmd += ["--model", model]
+        # Thinking level maps from the Claude "effort" knob. The overlapping
+        # set (low / medium / high / max) is identical; unknown values are
+        # passed through and pi validates them.
+        if self.effort:
+            cmd += ["--thinking", _PI_THINKING_ALIASES.get(self.effort, self.effort)]
+        # Session continuation. ``--session <id> --continue`` resumes an
+        # existing session; ``--session-id <id>`` pins the UUID for the
+        # first turn so the caller knows what to resume later. If neither
+        # is set, pi mints a fresh session and the manager captures its
+        # id from the ``session`` event at the top of the stream.
+        if spec.resume_session_id:
+            cmd += ["--session", spec.resume_session_id, "--continue"]
+        elif spec.session_id:
+            cmd += ["--session-id", spec.session_id]
+        cmd += list(spec.extra_args)
+        return cmd
+
     def _build_env(self, spec: AgentSpec) -> Dict[str, str]:
         env = dict(os.environ)
         env.update(spec.env_overrides)
@@ -673,6 +806,8 @@ def _build_env(self, spec: AgentSpec) -> Dict[str, str]:
         env["METAINFER_ROOT"] = str(_paths.root_dir())
         # Keep the agent from going interactive
         env.setdefault("DISABLE_INTERACTIVITY", "1")
+        if self.agent_backend in ("codex", "pi"):
+            return env
         # bypassPermissions under EUID=0 normally trips a hard exit
         # ("--dangerously-skip-permissions cannot be used with root/sudo
         # privileges"). ccb skips that check when IS_SANDBOX=1, which is
@@ -703,40 +838,60 @@ def _materialize_result(
                 except json.JSONDecodeError:
                     continue
         final_text = ""
-        for ev in reversed(events):
-            if ev.get("type") == "assistant" and isinstance(ev.get("message"), dict):
-                content = ev["message"].get("content")
-                if isinstance(content, list):
-                    for blk in content:
-                        if isinstance(blk, dict) and blk.get("type") == "text":
-                            final_text = blk.get("text", "")
-                            break
-                if final_text:
-                    break
-            if ev.get("type") == "result":
-                final_text = ev.get("result", "") or final_text
-                break
-        # Session id: emitted on the very first ``system`` event of the
-        # stream and again on every ``result`` event. Prefer the result's
-        # value (it's the final, post-turn session id; for ``--resume``
-        # invocations this matches the resumed-from id and confirms the
-        # continuation actually happened).
         session_id = None
-        for ev in events:
-            sid = ev.get("session_id")
-            if sid:
-                session_id = sid
-                if ev.get("type") == "result":
-                    break
-        # Pull the cost / usage block from the final ``result`` event.
-        # This is what the token budget circuit breaker keys off. None
-        # when the agent was killed / crashed before emitting result.
         usage: Optional[Dict[str, Any]] = None
-        for ev in reversed(events):
-            if ev.get("type") == "result" and isinstance(ev, dict):
-                if "usage" in ev or "total_cost_usd" in ev:
-                    usage = ev
+        if self.agent_backend == "pi":
+            final_text, session_id, usage = self._extract_pi_result(events)
+        else:
+            for ev in reversed(events):
+                if ev.get("type") == "item.completed":
+                    item = ev.get("item")
+                    if isinstance(item, dict) and item.get("type") == "agent_message":
+                        text = item.get("text")
+                        if isinstance(text, str) and text:
+                            final_text = text
+                            break
+                if ev.get("type") == "assistant" and isinstance(ev.get("message"), dict):
+                    content = ev["message"].get("content")
+                    if isinstance(content, list):
+                        for blk in content:
+                            if isinstance(blk, dict) and blk.get("type") == "text":
+                                final_text = blk.get("text", "")
+                                break
+                    if final_text:
+                        break
+                if ev.get("type") == "result":
+                    final_text = ev.get("result", "") or final_text
                     break
+            # Session id: emitted on the very first ``system`` event of the
+            # stream and again on every ``result`` event. Prefer the result's
+            # value (it's the final, post-turn session id; for ``--resume``
+            # invocations this matches the resumed-from id and confirms the
+            # continuation actually happened).
+            for ev in events:
+                sid = ev.get("session_id")
+                if not sid:
+                    sid = ev.get("thread_id")
+                if sid:
+                    session_id = sid
+                    if ev.get("type") == "result":
+                        break
+            # Pull the cost / usage block from the final ``result`` event.
+            # This is what the token budget circuit breaker keys off. None
+            # when the agent was killed / crashed before emitting result.
+            for ev in reversed(events):
+                if ev.get("type") == "result" and isinstance(ev, dict):
+                    if "usage" in ev or "total_cost_usd" in ev:
+                        usage = ev
+                        break
+                if ev.get("type") == "turn.completed" and isinstance(ev, dict):
+                    if "usage" in ev:
+                        usage = ev
+                        break
+            if usage is not None and session_id:
+                if not usage.get("session_id") and not usage.get("thread_id"):
+                    usage = dict(usage)
+                    usage["thread_id"] = session_id
         error = None
         failure_mode: Optional[Literal["infra", "logic", "budget"]] = None
         success = (rc == 0) and not handle.killed
@@ -765,6 +920,91 @@ def _materialize_result(
             usage=usage,
         )
 
+    def _extract_pi_result(
+        self, events: List[Dict[str, Any]]
+    ) -> tuple[str, Optional[str], Optional[Dict[str, Any]]]:
+        """Parse pi's JSONL event stream into (final_text, session_id, usage).
+
+        pi emits a different schema than claude/codex stream-json:
+
+        * Session id lives on the leading ``session`` event's ``id`` field
+          (not a top-level ``session_id`` / ``thread_id``).
+        * Assistant text lives on ``message_end`` / ``turn_end`` /
+          ``agent_end`` events under ``message.content`` (a list of blocks;
+          concatenate the ``text`` blocks of the last assistant message).
+        * Usage lives on the same events under ``message.usage`` with pi's
+          own field names (``input`` / ``output`` / ``cacheRead`` /
+          ``cacheWrite`` / ``cost.total``). We normalize it into the
+          canonical shape that :func:`usage_from_result_event` already
+          understands so the token-budget circuit breaker works unchanged.
+        """
+        session_id: Optional[str] = None
+        for ev in events:
+            if ev.get("type") == "session" and isinstance(ev.get("id"), str):
+                session_id = ev["id"]
+                break
+
+        final_text = ""
+        usage: Optional[Dict[str, Any]] = None
+        # Walk newest -> oldest. The last assistant message in stream order
+        # is the first one we hit in reverse, which is what we want for both
+        # final_text and the final usage tally.
+        for ev in reversed(events):
+            etype = ev.get("type")
+            if etype not in ("message_end", "turn_end", "agent_end"):
+                continue
+            # agent_end carries the full message list; pick the last assistant.
+            if etype == "agent_end":
+                msgs = ev.get("messages")
+                if not isinstance(msgs, list) or not msgs:
+                    continue
+                msg = None
+                for m in reversed(msgs):
+                    if isinstance(m, dict) and m.get("role") == "assistant":
+                        msg = m
+                        break
+                if msg is None:
+                    continue
+            else:
+                msg = ev.get("message")
+                if not isinstance(msg, dict) or msg.get("role") != "assistant":
+                    continue
+
+            # Final assistant text: concatenate text blocks.
+            if not final_text:
+                content = msg.get("content")
+                if isinstance(content, list):
+                    parts = [
+                        blk.get("text", "")
+                        for blk in content
+                        if isinstance(blk, dict) and blk.get("type") == "text"
+                    ]
+                    text = "".join(p for p in parts if isinstance(p, str))
+                    if text:
+                        final_text = text
+
+            # Usage: pi reports cumulative usage on every assistant message.
+            # The last assistant message carries the final tally.
+            if usage is None:
+                u = msg.get("usage")
+                if isinstance(u, dict):
+                    cost = u.get("cost") if isinstance(u.get("cost"), dict) else {}
+                    usage = {
+                        "type": "result",
+                        "usage": {
+                            "input_tokens": int(u.get("input", 0) or 0),
+                            "output_tokens": int(u.get("output", 0) or 0),
+                            "cache_read_input_tokens": int(u.get("cacheRead", 0) or 0),
+                            "cache_creation_input_tokens": int(u.get("cacheWrite", 0) or 0),
+                        },
+                        "total_cost_usd": float(cost.get("total", 0.0) or 0.0),
+                        "session_id": session_id,
+                    }
+
+            if final_text and usage is not None:
+                break
+        return final_text, session_id, usage
+
     def _write_status(
         self,
         spec: AgentSpec,
diff --git a/metainfer/orchestrator/tests/test_subagent_manager.py b/metainfer/orchestrator/tests/test_subagent_manager.py
new file mode 100644
index 00000000..f87ea94e
--- /dev/null
+++ b/metainfer/orchestrator/tests/test_subagent_manager.py
@@ -0,0 +1,352 @@
+"""Focused tests for SubAgentManager backend command/parse behavior."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+import time
+from pathlib import Path
+
+from metainfer.orchestrator.subagent_manager import (
+    AgentHandle,
+    AgentSpec,
+    SubAgentManager,
+)
+from metainfer.orchestrator._bootstrap import make_subagent_manager
+
+
+class _FakeProcess:
+    def __init__(self, returncode: int = 0) -> None:
+        self.returncode = returncode
+
+
+def _spec(tmp: Path, *, resume_session_id: str | None = None) -> AgentSpec:
+    workdir = tmp / "work"
+    log_dir = tmp / "logs"
+    workdir.mkdir()
+    log_dir.mkdir()
+    prompt = tmp / "prompt.txt"
+    prompt.write_text("do work", encoding="utf-8")
+    return AgentSpec(
+        name="agent1",
+        role="tester",
+        prompt_file=prompt,
+        workdir=workdir,
+        log_dir=log_dir,
+        resume_session_id=resume_session_id,
+    )
+
+
+def test_codex_command_uses_exec_json_and_current_config():
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        kb = tmp / "kb"
+        kb.mkdir()
+        mgr = SubAgentManager(
+            agent_backend="codex",
+            codex_bin="codex-dev",
+            extra_add_dirs=[kb],
+            default_model=None,
+        )
+        cmd = mgr._build_command(_spec(tmp))
+        assert cmd[:5] == ["codex-dev", "exec", "--json", "--color", "never"]
+        assert "--skip-git-repo-check" in cmd
+        assert "--sandbox" in cmd
+        assert "workspace-write" in cmd
+        assert "-C" in cmd
+        assert str(tmp / "work") in cmd
+        assert "--add-dir" in cmd
+        assert "/tmp" in cmd
+        assert str(kb.resolve()) in cmd
+        assert "--model" not in cmd
+        assert "--ignore-user-config" not in cmd
+        assert cmd[-1] == "-"
+
+
+def test_codex_resume_command_threads_session():
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        mgr = SubAgentManager(agent_backend="codex", codex_bin="codex-dev",
+                              default_model="gpt-5-codex")
+        cmd = mgr._build_command(_spec(tmp, resume_session_id="thread-123"))
+        assert cmd == [
+            "codex-dev", "exec", "resume", "--json",
+            "--skip-git-repo-check", "--model", "gpt-5-codex",
+            "thread-123", "-",
+        ]
+
+
+def test_materialize_codex_jsonl_result():
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        spec = _spec(tmp)
+        events = [
+            {"type": "thread.started", "thread_id": "thread-123"},
+            {"type": "turn.started"},
+            {
+                "type": "item.completed",
+                "item": {"type": "agent_message", "text": "done"},
+            },
+            {
+                "type": "turn.completed",
+                "usage": {
+                    "input_tokens": 10,
+                    "cached_input_tokens": 8,
+                    "output_tokens": 2,
+                    "reasoning_output_tokens": 1,
+                },
+            },
+        ]
+        ef = spec.events_file(1)
+        ef.write_text(
+            "Reading additional input from stdin...\n"
+            + "\n".join(json.dumps(e) for e in events)
+            + "\n",
+            encoding="utf-8",
+        )
+        handle = AgentHandle(
+            spec=spec,
+            attempt=1,
+            process=_FakeProcess(0),  # type: ignore[arg-type]
+            started_at=time.time() - 1,
+            last_output_at=time.time(),
+        )
+        mgr = SubAgentManager(agent_backend="codex")
+        result = mgr._materialize_result(handle, spec, 1)
+        assert result.success
+        assert result.final_text == "done"
+        assert result.session_id == "thread-123"
+        assert result.usage == {**events[-1], "thread_id": "thread-123"}
+
+
+def test_bootstrap_factory_reads_codex_env(monkeypatch):
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        monkeypatch.setenv("METAINFER_AGENT_BACKEND", "codex")
+        monkeypatch.setenv("METAINFER_CODEX_BIN", "codex-dev")
+        mgr = make_subagent_manager(
+            claude_bin="ccb",
+            model=None,
+            permission_mode="bypassPermissions",
+            effort="max",
+            extra_add_dirs=[],
+            snapshot_file=tmp / "agents.json",
+        )
+        try:
+            assert mgr.agent_backend == "codex"
+            assert mgr.codex_bin == "codex-dev"
+        finally:
+            mgr.shutdown()
+
+
+def test_pi_command_uses_json_print_mode():
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        kb = tmp / "kb"
+        kb.mkdir()
+        mgr = SubAgentManager(
+            agent_backend="pi",
+            pi_bin="pi-dev",
+            extra_add_dirs=[kb],
+            default_model="anthropic/claude-sonnet",
+            effort="high",
+        )
+        cmd = mgr._build_command(_spec(tmp))
+        assert cmd[:4] == ["pi-dev", "--mode", "json", "-p"]
+        assert "--model" in cmd
+        assert "anthropic/claude-sonnet" in cmd
+        assert "--thinking" in cmd
+        assert "high" in cmd
+        # session-id pinning only when spec.session_id is set; fresh run
+        # leaves it to pi to mint (captured from the stream later).
+        assert "--session-id" not in cmd
+        assert "--continue" not in cmd
+
+
+def test_pi_command_resume_threads_session():
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        mgr = SubAgentManager(agent_backend="pi", pi_bin="pi-dev",
+                              default_model="anthropic/claude-sonnet")
+        cmd = mgr._build_command(_spec(tmp, resume_session_id="01abc-def"))
+        assert "--session" in cmd
+        assert "01abc-def" in cmd
+        assert "--continue" in cmd
+        assert "--model" in cmd
+
+
+def test_pi_session_dir_isolates_from_user_pi_index():
+    """Orchestrator sessions must land in a per-task --session-dir so a
+    user running ``pi --continue`` from their shell (which indexes
+    ~/.pi) cannot attach to a production orchestrator session mid-flight.
+    """
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        session_dir = tmp / "pi-sessions"
+        mgr = SubAgentManager(
+            agent_backend="pi", pi_bin="pi-dev",
+            pi_session_dir=session_dir,
+        )
+        cmd = mgr._build_command(_spec(tmp))
+        assert "--session-dir" in cmd
+        idx = cmd.index("--session-dir")
+        assert cmd[idx + 1] == str(session_dir.resolve())
+        # the dir is created lazily so pi can write into it
+        assert session_dir.is_dir()
+
+
+def test_pi_session_dir_applied_on_resume_too():
+    """Resume turns must keep using the isolated session dir so the
+    continued session stays out of the user's ~/.pi index.
+    """
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        session_dir = tmp / "pi-sessions"
+        mgr = SubAgentManager(
+            agent_backend="pi", pi_bin="pi-dev",
+            pi_session_dir=session_dir,
+        )
+        cmd = mgr._build_command(_spec(tmp, resume_session_id="01abc-def"))
+        assert "--session-dir" in cmd
+        assert "--session" in cmd
+        assert "01abc-def" in cmd
+        assert "--continue" in cmd
+
+
+def test_pi_session_dir_absent_when_not_configured():
+    """Legacy callers (no state_dir) keep the old behavior: pi picks its
+    own default session dir. No regression for backends that never set
+    pi_session_dir.
+    """
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        mgr = SubAgentManager(agent_backend="pi", pi_bin="pi-dev")
+        cmd = mgr._build_command(_spec(tmp))
+        assert "--session-dir" not in cmd
+
+
+def test_bootstrap_factory_wires_pi_session_dir_from_state_dir(monkeypatch):
+    """make_subagent_manager derives pi_session_dir from state_dir so every
+    task's orchestrator sessions are isolated under <state_dir>/pi-sessions.
+    """
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        state_dir = tmp / "state"
+        state_dir.mkdir()
+        monkeypatch.setenv("METAINFER_AGENT_BACKEND", "pi")
+        mgr = make_subagent_manager(
+            claude_bin="ccb",
+            model=None,
+            permission_mode="bypassPermissions",
+            effort="max",
+            extra_add_dirs=[],
+            snapshot_file=tmp / "agents.json",
+            state_dir=state_dir,
+        )
+        try:
+            assert mgr.agent_backend == "pi"
+            assert mgr.pi_session_dir == (state_dir / "pi-sessions").resolve()
+        finally:
+            mgr.shutdown()
+
+
+def test_bootstrap_factory_no_state_dir_leaves_pi_session_dir_none(monkeypatch):
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        monkeypatch.setenv("METAINFER_AGENT_BACKEND", "pi")
+        mgr = make_subagent_manager(
+            claude_bin="ccb",
+            model=None,
+            permission_mode="bypassPermissions",
+            effort="max",
+            extra_add_dirs=[],
+            snapshot_file=tmp / "agents.json",
+        )
+        try:
+            assert mgr.pi_session_dir is None
+        finally:
+            mgr.shutdown()
+
+
+def test_pi_normalize_accepts_aliases():
+    from metainfer.orchestrator.subagent_manager import _normalize_agent_backend
+    assert _normalize_agent_backend("pi") == "pi"
+    assert _normalize_agent_backend("PI") == "pi"
+    assert _normalize_agent_backend("pi-coding-agent") == "pi"
+    assert _normalize_agent_backend("earendil") == "pi"
+
+
+def test_materialize_pi_jsonl_result():
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        spec = _spec(tmp)
+        events = [
+            {"type": "session", "version": 3, "id": "01a00ecf-aaaa-bbbb-cccc-dddddddddddd",
+             "timestamp": "2026-08-17T08:21:39.437Z", "cwd": "/tmp"},
+            {"type": "agent_start"},
+            {"type": "turn_start"},
+            {"type": "message_end", "message": {"role": "user", "content": []}},
+            {"type": "message_end", "message": {
+                "role": "assistant",
+                "content": [{"type": "text", "text": "hello world"}],
+                "usage": {"input": 581, "output": 3, "cacheRead": 0,
+                          "cacheWrite": 0, "reasoning": 0, "totalTokens": 584,
+                          "cost": {"input": 0, "output": 0, "cacheRead": 0,
+                                   "cacheWrite": 0, "total": 0.0123}},
+                "stopReason": "stop"}},
+            {"type": "turn_end", "message": {"role": "assistant",
+                "content": [{"type": "text", "text": "hello world"}],
+                "usage": {"input": 581, "output": 3, "cacheRead": 0,
+                          "cacheWrite": 0, "totalTokens": 584,
+                          "cost": {"total": 0.0123}}}, "toolResults": []},
+            {"type": "agent_end", "messages": [
+                {"role": "user", "content": []},
+                {"role": "assistant", "content": [{"type": "text", "text": "hello world"}],
+                 "usage": {"input": 581, "output": 3, "cacheRead": 0,
+                           "cacheWrite": 0, "cost": {"total": 0.0123}}}]},
+            {"type": "agent_settled"},
+        ]
+        ef = spec.events_file(1)
+        ef.write_text(
+            "\n".join(json.dumps(e) for e in events) + "\n",
+            encoding="utf-8",
+        )
+        handle = AgentHandle(
+            spec=spec,
+            attempt=1,
+            process=_FakeProcess(0),  # type: ignore[arg-type]
+            started_at=time.time() - 1,
+            last_output_at=time.time(),
+        )
+        mgr = SubAgentManager(agent_backend="pi")
+        result = mgr._materialize_result(handle, spec, 1)
+        assert result.success
+        assert result.final_text == "hello world"
+        assert result.session_id == "01a00ecf-aaaa-bbbb-cccc-dddddddddddd"
+        assert result.usage is not None
+        assert result.usage["usage"]["input_tokens"] == 581
+        assert result.usage["usage"]["output_tokens"] == 3
+        assert result.usage["usage"]["cache_read_input_tokens"] == 0
+        assert result.usage["usage"]["cache_creation_input_tokens"] == 0
+        assert result.usage["total_cost_usd"] == 0.0123
+        assert result.usage["session_id"] == "01a00ecf-aaaa-bbbb-cccc-dddddddddddd"
+
+
+def test_bootstrap_factory_reads_pi_env(monkeypatch):
+    with tempfile.TemporaryDirectory() as td:
+        tmp = Path(td)
+        monkeypatch.setenv("METAINFER_AGENT_BACKEND", "pi")
+        monkeypatch.setenv("METAINFER_PI_BIN", "/usr/local/bin/pi")
+        mgr = make_subagent_manager(
+            claude_bin="ccb",
+            model=None,
+            permission_mode="bypassPermissions",
+            effort="max",
+            extra_add_dirs=[],
+            snapshot_file=tmp / "agents.json",
+        )
+        try:
+            assert mgr.agent_backend == "pi"
+            assert mgr.pi_bin == "/usr/local/bin/pi"
+        finally:
+            mgr.shutdown()
diff --git a/metainfer/orchestrator/tests/test_token_budget.py b/metainfer/orchestrator/tests/test_token_budget.py
index bf5d66b8..6af06b93 100644
--- a/metainfer/orchestrator/tests/test_token_budget.py
+++ b/metainfer/orchestrator/tests/test_token_budget.py
@@ -176,6 +176,24 @@ def test_usage_from_result_event():
     assert rec2.total_cost_usd == 0.0
 
 
+def test_usage_from_codex_turn_completed_event():
+    ev = {
+        "type": "turn.completed",
+        "thread_id": "thread-123",
+        "usage": {
+            "input_tokens": 100,
+            "cached_input_tokens": 80,
+            "output_tokens": 20,
+        },
+    }
+    rec = usage_from_result_event(ev, agent="codex-a", source="orchestrator")
+    assert rec.input_tokens == 100
+    assert rec.output_tokens == 20
+    assert rec.cache_read_input_tokens == 80
+    assert rec.total_cost_usd == 0.0
+    assert rec.session_id == "thread-123"
+
+
 def test_reset_clears_everything():
     with tempfile.TemporaryDirectory() as td:
         b = TokenBudget(td, max_cost_usd=1.0)
diff --git a/metainfer/orchestrator/token_budget.py b/metainfer/orchestrator/token_budget.py
index 7b0c8f86..f4c24ca3 100644
--- a/metainfer/orchestrator/token_budget.py
+++ b/metainfer/orchestrator/token_budget.py
@@ -567,17 +567,32 @@ def usage_from_result_event(
     usage = event.get("usage") if isinstance(event, dict) else None
     if not isinstance(usage, dict):
         usage = {}
+    cache_read = usage.get("cache_read_input_tokens")
+    if cache_read is None:
+        cache_read = usage.get("cached_input_tokens", 0)
+    # pi reports usage under its own camelCase names (input / output /
+    # cacheRead / cacheWrite). Tolerate them so a raw pi event also works.
+    input_tokens = usage.get("input_tokens", usage.get("input", 0))
+    output_tokens = usage.get("output_tokens", usage.get("output", 0))
+    cache_creation = usage.get(
+        "cache_creation_input_tokens", usage.get("cacheWrite", 0)
+    )
+    total_cost = event.get("total_cost_usd")
+    if total_cost is None:
+        cost = usage.get("cost")
+        if isinstance(cost, dict):
+            total_cost = cost.get("total", 0.0)
     return UsageRecord(
         agent=str(agent),
         source=str(source),
         phase=phase,
         ended_at=time.time(),
-        input_tokens=int(usage.get("input_tokens", 0) or 0),
-        output_tokens=int(usage.get("output_tokens", 0) or 0),
-        cache_read_input_tokens=int(usage.get("cache_read_input_tokens", 0) or 0),
-        cache_creation_input_tokens=int(usage.get("cache_creation_input_tokens", 0) or 0),
-        total_cost_usd=float(event.get("total_cost_usd", 0.0) or 0.0),
-        session_id=event.get("session_id"),
+        input_tokens=int(input_tokens or 0),
+        output_tokens=int(output_tokens or 0),
+        cache_read_input_tokens=int(cache_read or 0),
+        cache_creation_input_tokens=int(cache_creation or 0),
+        total_cost_usd=float(total_cost or 0.0),
+        session_id=event.get("session_id") or event.get("thread_id"),
     )
 
 
diff --git a/metainfer/tasks/calc_value/orchestrator/orchestrator.py b/metainfer/tasks/calc_value/orchestrator/orchestrator.py
index d50bd96b..518e2fbe 100644
--- a/metainfer/tasks/calc_value/orchestrator/orchestrator.py
+++ b/metainfer/tasks/calc_value/orchestrator/orchestrator.py
@@ -251,6 +251,7 @@ def run_with_requirements(
         snapshot_file=paths["agents_file"],
         max_concurrent=5,
         budget=budget,
+        state_dir=state_dir,
     )
     # Wire the hard-exhausted callback NOW that the manager exists.
     # When the hard threshold is crossed, every in-flight agent gets
diff --git a/metainfer/tasks/dcu_kernel_auto_opt/orchestrator/orchestrator.py b/metainfer/tasks/dcu_kernel_auto_opt/orchestrator/orchestrator.py
index 8d74875c..ccf9056d 100644
--- a/metainfer/tasks/dcu_kernel_auto_opt/orchestrator/orchestrator.py
+++ b/metainfer/tasks/dcu_kernel_auto_opt/orchestrator/orchestrator.py
@@ -77,6 +77,7 @@ def run_with_requirements(
         extra_add_dirs=[workspace_dir],
         snapshot_file=state_dir / "agents.json",
         max_concurrent=4,
+        state_dir=state_dir,
     )
     restore = install_subagent_shutdown_handlers(manager, pid_file=pid_file)
     try:
diff --git a/metainfer/tasks/evolve_kernel/orchestrator/orchestrator.py b/metainfer/tasks/evolve_kernel/orchestrator/orchestrator.py
index e2f0bcbb..ab8bf92f 100644
--- a/metainfer/tasks/evolve_kernel/orchestrator/orchestrator.py
+++ b/metainfer/tasks/evolve_kernel/orchestrator/orchestrator.py
@@ -109,6 +109,7 @@ def run_with_requirements(
         effort=effort,
         extra_add_dirs=[repo_root, logs_root],
         snapshot_file=paths["agents_file"],
+        state_dir=state_dir,
     )
     orch = Orchestrator(req=req, store=store, cfg=cfg, manager=manager)
 
diff --git a/metainfer/tasks/find_low_hanging_kernel/orchestrator/orchestrator.py b/metainfer/tasks/find_low_hanging_kernel/orchestrator/orchestrator.py
index 8350f887..1735ac58 100644
--- a/metainfer/tasks/find_low_hanging_kernel/orchestrator/orchestrator.py
+++ b/metainfer/tasks/find_low_hanging_kernel/orchestrator/orchestrator.py
@@ -131,6 +131,7 @@ def run_with_requirements(
         effort=effort,
         extra_add_dirs=[repo_root, logs_root, workspace_dir, *user_paths],
         snapshot_file=paths["agents_file"],
+        state_dir=state_dir,
     )
     pipeline = Pipeline(req=req, store=store, cfg=cfg, manager=manager)
 
diff --git a/metainfer/tasks/gen_cpp_infer_framework/orchestrator/orchestrator.py b/metainfer/tasks/gen_cpp_infer_framework/orchestrator/orchestrator.py
index 1cbf1570..ce9c3bf4 100644
--- a/metainfer/tasks/gen_cpp_infer_framework/orchestrator/orchestrator.py
+++ b/metainfer/tasks/gen_cpp_infer_framework/orchestrator/orchestrator.py
@@ -225,6 +225,7 @@ def run_with_requirements(
         extra_add_dirs=[notebooks_dir, repo_root, workspace_dir, logs_root, state_dir],
         snapshot_file=paths["agents_file"],
         budget=budget,
+        state_dir=state_dir,
     )
     # Wire the hard-exhausted callback NOW that the manager exists.
     # When the hard threshold is crossed, every in-flight agent gets
diff --git a/metainfer/tasks/gen_infer_framework/orchestrator/orchestrator.py b/metainfer/tasks/gen_infer_framework/orchestrator/orchestrator.py
index 4ee576aa..9b070f30 100644
--- a/metainfer/tasks/gen_infer_framework/orchestrator/orchestrator.py
+++ b/metainfer/tasks/gen_infer_framework/orchestrator/orchestrator.py
@@ -213,6 +213,7 @@ def run_with_requirements(
         extra_add_dirs=[notebooks_dir, repo_root, workspace_dir, logs_root],
         snapshot_file=paths["agents_file"],
         budget=budget,
+        state_dir=state_dir,
     )
     # Wire the hard-exhausted callback NOW that the manager exists.
     # When the hard threshold is crossed, every in-flight agent gets
diff --git a/metainfer/tasks/gen_infer_framework_cpp/orchestrator/orchestrator.py b/metainfer/tasks/gen_infer_framework_cpp/orchestrator/orchestrator.py
index da8c8f63..739dae69 100644
--- a/metainfer/tasks/gen_infer_framework_cpp/orchestrator/orchestrator.py
+++ b/metainfer/tasks/gen_infer_framework_cpp/orchestrator/orchestrator.py
@@ -262,6 +262,7 @@ def run_with_requirements(
         ],
         snapshot_file=paths["agents_file"],
         budget=budget,
+        state_dir=state_dir,
     )
     # Wire the hard-exhausted callback NOW that the manager exists.
     # When the hard threshold is crossed, every in-flight agent gets
diff --git a/metainfer/tasks/opt_kernel/orchestrator/orchestrator.py b/metainfer/tasks/opt_kernel/orchestrator/orchestrator.py
index 9307aeed..4e111b61 100644
--- a/metainfer/tasks/opt_kernel/orchestrator/orchestrator.py
+++ b/metainfer/tasks/opt_kernel/orchestrator/orchestrator.py
@@ -103,6 +103,7 @@ def run_with_requirements(
         effort=effort,
         extra_add_dirs=[notebooks_dir, repo_root, logs_root],
         snapshot_file=paths["agents_file"],
+        state_dir=state_dir,
     )
     orch = Orchestrator(req=req, store=store, cfg=cfg, manager=manager)