feat(sleep): add pi-coding-agent transcript source

skillopt_sleep/__main__.py

@@ -69,13 +69,15 @@ def _report_payload(rep, outcome) -> Dict[str, Any]:
 def _add_common(p: argparse.ArgumentParser) -> None:
     p.add_argument("--project", default="")
     p.add_argument("--scope", default="", choices=["", "all", "invoked"])
-    p.add_argument("--backend", default="", choices=["", "mock", "claude", "codex", "copilot"])
+    p.add_argument("--backend", default="", choices=["", "mock", "claude", "codex", "copilot", "pi"])
     p.add_argument("--model", default="")
     p.add_argument("--codex-path", default="", help="path to the real @openai/codex binary")
+    p.add_argument("--pi-path", default="", help="path to the pi binary (default: pi on PATH)")
     p.add_argument("--claude-home", default="", help="override ~/.claude (also isolates state)")
     p.add_argument("--codex-home", default="", help="override ~/.codex for archived session harvest")
-    p.add_argument("--source", default="", choices=["", "claude", "codex", "auto"],
+    p.add_argument("--source", default="", choices=["", "claude", "codex", "pi", "auto"],
                    help="session transcript source")
+    p.add_argument("--pi-home", default="", help="override ~/.pi for pi-coding-agent session harvest")
     p.add_argument("--lookback-hours", type=int, default=None,
                    help="harvest window in hours; 0 = scan full history")
     p.add_argument("--edit-budget", type=int, default=0)
@@ -106,10 +108,14 @@ def _cfg_from_args(args, task_meta: Dict[str, Any] | None = None) -> Any:
         overrides["model"] = args.model
     if getattr(args, "codex_path", ""):
         overrides["codex_path"] = os.path.abspath(args.codex_path)
+    if getattr(args, "pi_path", ""):
+        overrides["pi_path"] = args.pi_path
     if getattr(args, "claude_home", ""):
         overrides["claude_home"] = os.path.abspath(args.claude_home)
     if getattr(args, "codex_home", ""):
         overrides["codex_home"] = os.path.abspath(args.codex_home)
+    if getattr(args, "pi_home", ""):
+        overrides["pi_home"] = os.path.abspath(args.pi_home)
     if getattr(args, "source", ""):
         overrides["transcript_source"] = args.source
     lh = getattr(args, "lookback_hours", None)

skillopt_sleep/backend.py

@@ -541,6 +541,89 @@ def tokens_used(self) -> int:
         return self._tokens
 
 
+# ── Pi CLI backend ────────────────────────────────────────────────
+
+
+class PiCliBackend(CliBackend):
+    """Drives the authenticated `pi` CLI: pi -p "<prompt>".
+
+    pi (the pi coding agent) speaks the open Agent Skills standard and supports
+    a `-p` / `--print` headless mode, so it slots in alongside the claude/codex
+    CLI backends. Using pi here means the replay model is whatever the user has
+    configured in pi (e.g. `zai/glm-5.2`), keeping source and backend on the same
+    agent — which is the design intent of `--source pi`.
+    """
+
+    name = "pi"
+
+    def __init__(self, model: str = "", pi_path: str = "pi", timeout: int = 180) -> None:
+        super().__init__(model=model or os.environ.get("SKILLOPT_SLEEP_PI_MODEL", ""),
+                         timeout=timeout)
+        self.pi_path = pi_path
+
+    _CLI_ERROR_MARKERS = (
+        "Not logged in",
+        "Authentication required",
+        "Invalid API key",
+        "Unauthorized",
+        "provider not found",
+        "no provider",
+    )
+
+    def _detect_cli_error(self, stdout: str, stderr: str) -> None:
+        import logging
+        check_stdout = stdout if len(stdout) < 300 else ""
+        combined = check_stdout + "\n" + stderr
+        for marker in self._CLI_ERROR_MARKERS:
+            if marker.lower() in combined.lower():
+                logging.getLogger("skillopt_sleep").warning(
+                    "pi CLI returned a likely auth/config error: %s",
+                    combined[:200].replace("\n", " "),
+                )
+                self.last_call_error = combined[:500]
+                return
+
+    def _call(self, prompt: str, *, max_tokens: int = 1024) -> str:
+        # Run ISOLATED so the ambient pi environment does not leak into the
+        # optimizer/target call: disable tools, skills, and context files, and
+        # run from a clean temp cwd so no project AGENTS.md is picked up.
+        #   --no-tools             no tool use during replay
+        #   --no-skills            do not load the user's installed skills
+        #   --no-context-files     do not load AGENTS.md/CLAUDE.md
+        #   --no-session           ephemeral; do not write to session history
+        #   --no-extensions        skip extension discovery
+        import shutil
+        self.last_call_error = ""
+        cmd = [self.pi_path, "-p", "--no-session"]
+        cmd += ["--no-tools", "--no-skills", "--no-context-files", "--no-extensions"]
+        if self.model:
+            cmd += ["--model", self.model]
+        cmd += [prompt]
+        clean_cwd = tempfile.mkdtemp(prefix="skillopt_sleep_pi_")
+        try:
+            try:
+                proc = subprocess.run(
+                    cmd, capture_output=True, text=True, timeout=self.timeout, cwd=clean_cwd,
+                )
+            except subprocess.TimeoutExpired:
+                self.last_call_error = f"pi CLI timed out after {self.timeout}s"
+                return ""
+            except Exception as exc:
+                self.last_call_error = f"pi CLI failed: {exc}"
+                return ""
+        finally:
+            try:
+                shutil.rmtree(clean_cwd, ignore_errors=True)
+            except Exception:
+                pass
+        out = (proc.stdout or "").strip()
+        stderr = (proc.stderr or "").strip()
+        self._detect_cli_error(out, stderr)
+        if proc.returncode != 0 and not self.last_call_error:
+            self.last_call_error = f"pi CLI exited {proc.returncode}: {stderr[:500]}"
+        return out
+
+
 # ── Claude Code CLI backend ───────────────────────────────────────────────────
 
 class ClaudeCliBackend(CliBackend):
@@ -1310,10 +1393,13 @@ def get_backend(
     model: str = "",
     claude_path: str = "claude",
     codex_path: str = "",
+    pi_path: str = "",
     azure_endpoint: str = "",
     project_dir: str = "",
 ) -> Backend:
     n = (name or "mock").strip().lower()
+    if n in {"pi", "pi_cli", "pi_coding_agent", "pi-coding-agent"}:
+        return PiCliBackend(model=model, pi_path=pi_path or "pi")
     if n in {"claude", "anthropic", "claude_cli", "claude_code"}:
         return ClaudeCliBackend(model=model, claude_path=claude_path)
     if n in {"codex", "codex_cli", "openai_codex"}:

skillopt_sleep/config.py

@@ -19,13 +19,15 @@
 HOME_STATE_DIR = os.path.expanduser("~/.skillopt-sleep")
 CLAUDE_HOME = os.path.expanduser("~/.claude")
 CODEX_HOME = os.path.expanduser("~/.codex")
+PI_HOME = os.path.expanduser("~/.pi")
 
 
 DEFAULTS: Dict[str, Any] = {
     # ── scope ──────────────────────────────────────────────────────────────
     "claude_home": CLAUDE_HOME,
     "codex_home": CODEX_HOME,
-    "transcript_source": "claude",  # "claude" | "codex" | "auto"
+    "pi_home": PI_HOME,
+    "transcript_source": "claude",  # "claude" | "codex" | "pi" | "auto"
     "projects": "invoked",        # "invoked" | "all" | [list of abs paths]
     "invoked_project": "",        # filled at runtime (cwd) when projects == "invoked"
     "lookback_hours": 72,         # harvest window when no prior sleep recorded
@@ -40,6 +42,7 @@
     "model": "",                  # backend-specific; "" => backend default
     "gate_mode": "on",            # "on" (validation-gated) | "off" (greedy, no hard filter)
     "codex_path": "",             # "" => auto-detect the real @openai/codex binary
+    "pi_path": "",                # "" => use `pi` on PATH
     "edit_budget": 4,             # textual learning rate (max edits/night)
     "gate_metric": "mixed",       # hard | soft | mixed (mixed best for tiny holdouts)
     "gate_mixed_weight": 0.5,
@@ -107,6 +110,10 @@ def transcripts_dir(self) -> str:
     def codex_archived_sessions_dir(self) -> str:
         return os.path.join(self.data["codex_home"], "archived_sessions")
 
+    @property
+    def pi_sessions_dir(self) -> str:
+        return os.path.join(self.data["pi_home"], "agent", "sessions")
+
     @property
     def history_path(self) -> str:
         return os.path.join(self.data["claude_home"], "history.jsonl")

skillopt_sleep/cycle.py

@@ -114,6 +114,7 @@ def run_sleep_cycle(
         cfg.get("backend", "mock"),
         model=cfg.get("model", ""),
         codex_path=cfg.get("codex_path", ""),
+        pi_path=cfg.get("pi_path", ""),
         project_dir=project,
     )
     _progress(cfg, f"night {night}: project={project} backend={backend.name}")