github/microsoft-SkillOpt
1
0
Fork 0
mirror of https://github.com/microsoft/SkillOpt.git synced 2026-08-03 07:02:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(sleep): harden Pi integration for merge

Browse Source
This commit is contained in:
Yif-Yang
2026-08-02 19:18:50 +00:00
parent 6ca6808c82
commit 5b63e1c2be
13 changed files with 1030 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
docs/guide/installation.md
View File

@@ -27,9 +27,9 @@ checkout for those files.
These docs track the latest `main`. The current PyPI release is `0.2.0`.
The generic research `openai_compatible` backend, SkillOpt-Sleep handoff,
Sleep support for non-Azure OpenAI-compatible endpoints, the Sleep
`--preferences` flag, and Cursor source/backend/plugin support landed after
that release and require a source install from `main` until the next
release.
`--preferences` flag, Cursor source/backend/plugin support, and Pi
source/backend support landed after that release and require a source
install from `main` until the next release.
### Source checkout
@@ -63,6 +63,19 @@ Install extras for specific benchmarks or backends:
Claude Code CLI separately. The SDK extra is only needed when selecting an
SDK-backed Claude Code exec path.
=== "Pi coding-agent CLI (optional)"
```bash
npm install -g --ignore-scripts @earendil-works/pi-coding-agent
```
Install and authenticate the [Pi coding-agent CLI](https://github.com/earendil-works/pi)
only when using SkillOpt-Sleep with `--backend pi`. Harvesting local Pi
transcripts with `--source pi` does not require the CLI or provider
authentication. By default, the source reads below
`~/.pi/agent/sessions`; `--pi-home` selects the parent directory that
contains `agent/sessions`.
=== "Qwen (Local)"
```bash

54
docs/reference/cli.md
View File

@@ -4,8 +4,9 @@
> include the generic research `openai_compatible` backend, Sleep handoff,
> Sleep support for non-Azure OpenAI-compatible endpoints, the Sleep
> `--preferences` flag, the research `cursor_exec` target harness, or Cursor
> source/backend/plugin support or VS Code Copilot transcript harvesting; use
> a source install from `main` for those features until the next release.
> source/backend/plugin support, Pi source/backend support, or VS Code Copilot
> transcript harvesting; use a source install from `main` for those features
> until the next release.
## Training
@@ -128,12 +129,14 @@ Actions are `run`, `dry-run`, `status`, `adopt`, `harvest`, `schedule`, and
|---|---|
| `--project PATH` | Project used for transcript scope, targets, state, and staging (default: current directory) |
| `--scope invoked\|all` | Harvest this project or all projects |
| `--source claude\|codex\|copilot\|cursor\|auto` | Transcript source; `auto` keeps Codex-then-Claude precedence and does not select Copilot or Cursor |
| `--backend mock\|claude\|codex\|copilot\|cursor\|handoff\|azure_openai` | Replay/optimizer backend |
| `--source claude\|codex\|copilot\|cursor\|pi\|auto` | Transcript source; `auto` keeps Codex-then-Claude precedence and does not select Copilot, Cursor, or Pi |
| `--backend mock\|claude\|codex\|copilot\|cursor\|pi\|handoff\|azure_openai` | Replay/optimizer backend |
| `--model NAME` | Backend-specific model override |
| `--cursor-home PATH` | Override `~/.cursor` for Cursor transcript harvesting |
| `--pi-home PATH` | Parent directory containing Pi's `agent/sessions` tree (default: `~/.pi`) |
| `--vscode-workspace-storage PATH` | Override VS Code's `User/workspaceStorage` root for Copilot transcript harvesting |
| `--cursor-path PATH` | Path to the installed Cursor Agent CLI |
| `--pi-path PATH` | Path to the installed Pi coding-agent CLI |
| `--preferences TEXT` | House rules supplied to reflection |
| `--lookback-hours N` | Initial transcript lookback; `0` scans all history |
| `--max-sessions N` / `--max-tasks N` | Bound the harvested workload |
@@ -143,6 +146,11 @@ Actions are `run`, `dry-run`, `status`, `adopt`, `harvest`, `schedule`, and
| `--progress` / `--json` | Progress or machine-readable output |
| `--auto-adopt` | Apply an accepted staged proposal automatically |
The `mock` and `handoff` backends make no network calls. A real backend sends
mining, replay, judging, and reflection prompts derived from harvested
transcripts and tasks to its selected provider. Review that provider's
data-retention and privacy policy before processing sensitive sessions.
### VS Code GitHub Copilot Chat source
`--source copilot` reads local VS Code GitHub Copilot Chat session logs from
@@ -167,6 +175,44 @@ The managed `schedule` command does not persist `--source` or
`"vscode_workspace_storage": "/absolute/path/to/workspaceStorage"` in
`~/.skillopt-sleep/config.json`.
### Pi source and backend
`--source pi` reads local session JSONL files below
`~/.pi/agent/sessions`; use `--pi-home PATH` to select the parent directory that
contains `agent/sessions`. This local source does not require the Pi CLI or
provider authentication. It retains user/assistant text, tool names, and lexical
feedback found in user text, while excluding thinking, tool arguments, tool
outputs, images, and unrelated metadata. The absolute project `cwd` from the
session header is retained for scope filtering and may appear in miner prompts
sent to a real backend and its provider. Known secret-shaped strings in retained
message text are redacted as defense in depth, not as a guarantee. Pi is an explicit source: `--source auto`
retains Codex-then-Claude precedence and does not select it.
Transcript source and model backend are independent. `--backend pi` launches a
locally installed, authenticated Pi CLI and makes real provider calls for
mining, replay, judging, and reflection. Use `--pi-path PATH` to select its
executable and `--model NAME` to override its configured model:
```bash
skillopt-sleep run --project "$(pwd)" \
--source pi --backend pi --pi-path /absolute/path/to/pi \
--model provider/model --max-sessions 5 --max-tasks 3 --progress
```
For these calls, SkillOpt disables Pi tools, skills, context files, extensions,
prompt templates, themes, and session writes. Pi authentication and model
configuration remain available. It also enables Pi's offline startup mode, so
configured npm/git packages are not installed or updated and model catalogs are
not refreshed; the selected model provider is still contacted for generation.
These controls should not be treated as permanent or complete isolation. The
provider selected in Pi receives the transcript-derived prompts.
The managed `schedule` command preserves the backend but not `--source`,
`--pi-home`, `--pi-path`, or `--model`. Before scheduling Pi, put
`transcript_source`, `pi_home`, `pi_path`, and `model` in
`~/.skillopt-sleep/config.json`; use an absolute `pi_path` and verify
authentication for the scheduled account.
### Cursor source and backend
`--source cursor` reads local Cursor JSONL transcripts from

59
docs/sleep/README.md
View File

@@ -17,7 +17,7 @@ normal agent requests.
One "night":
```
harvest Claude Code / Codex / VS Code Copilot / Cursor transcripts → mine recurring tasks → replay via the configured backend (isolation varies by backend; mock/handoff make no network calls)
harvest Claude Code / Codex / VS Code Copilot / Cursor / Pi transcripts → mine recurring tasks → replay via the configured backend (isolation varies by backend; mock/handoff make no network calls)
→ consolidate (reflect → bounded edit → GATE on real held-out tasks)
→ stage proposal → (you) adopt
```
@@ -49,6 +49,22 @@ experience → long-term competence).
> context, and account/model metadata. Known secret-shaped strings are
> redacted, but this remains defense in depth rather than a guarantee.
>
> The Pi source reads local sessions below `~/.pi/agent/sessions`, retaining
> user/assistant text, tool names, and lexical feedback found in user text. It
> excludes thinking, tool arguments, tool outputs, images, and unrelated
> metadata. The absolute project `cwd` from the session header is retained for
> scope filtering and may appear in miner prompts sent to a real backend and its
> provider. Known secret-shaped strings in retained message text are redacted
> only as defense in depth.
> The Pi backend uses the installed, authenticated Pi CLI to contact the user's
> selected model provider. Calls disable tools, skills, context files, extensions, prompt
> templates, themes, and session writes, while retaining Pi authentication and
> model configuration. Pi's offline startup mode also prevents configured
> npm/git package installation, package updates, and model-catalog refresh; it
> does not prevent the selected provider call. This is not a guarantee of
> permanent or complete isolation. Review the provider's retention and privacy
> policy before sending transcript-derived prompts from sensitive sessions.
>
> By default, each stateful night also writes a local `evidence.jsonl` under
> the project staging tree (beside the report when one is staged); dry-runs
> write evidence under the configured Sleep state directory. The log contains
@@ -73,9 +89,9 @@ skillopt-sleep schedule # install a nightly cron entry for this project
> **Version note.** This page tracks `main`. PyPI 0.2.0 provides the base
> commands above. Cursor source/backend/plugin support, VS Code Copilot
> transcript harvesting, Sleep handoff, non-Azure OpenAI-compatible endpoints,
> and `--preferences` landed later and require a source install from `main`
> until the next release.
> transcript harvesting, Pi source/backend support, Sleep handoff, non-Azure
> OpenAI-compatible endpoints, and `--preferences` landed later and require a
> source install from `main` until the next release.
The per-agent integrations below still come from the repo; the CLI above is the
standalone, pip-only way to run a cycle. Claude Code, Codex, Cursor, Copilot, and
@@ -119,6 +135,41 @@ The managed scheduler does not preserve `--source` or
`"vscode_workspace_storage": "/absolute/path/to/workspaceStorage"` in
`~/.skillopt-sleep/config.json`.
### Pi
Pi transcript harvesting and model execution are independent. Use `--source pi`
to read local session JSONL files below `~/.pi/agent/sessions`, or set
`--pi-home` to the parent directory that contains `agent/sessions` (the default
is `~/.pi`). The source alone does not require Pi CLI installation or provider
authentication. Pi is never selected implicitly:
`--source auto` retains Codex-then-Claude precedence.
`--backend pi` uses a locally installed and authenticated Pi CLI for real
model-provider calls during mining, replay, judging, and reflection. Select its
executable with `--pi-path` and override its configured model with `--model`:
```bash
skillopt-sleep run --project "$(pwd)" \
--source pi --backend pi --pi-path /absolute/path/to/pi \
--model provider/model --max-sessions 5 --max-tasks 3 --progress
```
These calls disable tools, skills, context files, extensions, prompt templates,
themes, and session writes, but still use the user's Pi authentication and model
configuration. They also enable Pi's offline startup mode to prevent configured
npm/git package installation, package updates, and model-catalog refresh; the
selected model provider is still contacted. Treat those controls as bounded
invocation setup, not permanent or complete isolation. A real Pi backend sends
transcript-derived prompts to the provider configured in Pi; inspect that
provider's retention and privacy policy first. The `mock` and `handoff` backends
make no network calls.
The managed scheduler records the backend but does not preserve `--source`,
`--pi-home`, `--pi-path`, or `--model`. Before scheduling Pi, set
`transcript_source`, `pi_home`, `pi_path`, and `model` in
`~/.skillopt-sleep/config.json`. Use an absolute `pi_path` and verify the
scheduled account's Pi authentication.
### Cursor
Cursor transcript harvesting and model execution are independent. Use

73
plugins/README.md
View File

@@ -47,8 +47,9 @@ an importable `skillopt_sleep` module. Install with `uv tool install skillopt` o
> **Version note.** This integration reference tracks `main`. PyPI 0.2.0
> supports the base Sleep CLI, while Cursor source/backend/plugin support,
> handoff, Sleep support for non-Azure OpenAI-compatible endpoints, and
> `--preferences` require a source checkout from `main` until the next release.
> Pi source/backend support, handoff, Sleep support for non-Azure
> OpenAI-compatible endpoints, and `--preferences` require a source checkout
> from `main` until the next release.
## One sleep cycle
@@ -64,10 +65,10 @@ optimization.
## Data boundary
- Harvesting is local and read-only. The `mock` backend has no model-provider
data path and no API spend.
- A real backend sends truncated transcript excerpts and derived task content to
the provider selected for mining, replay, judging, and reflection.
- Harvesting is local and read-only. The `mock` and `handoff` backends make no
network calls; handoff writes prompts for separate, user-controlled completion.
- A real backend sends mining, replay, judging, and reflection prompts derived
from truncated transcript excerpts and tasks to the selected provider.
- The Cursor source reads local user/assistant message text, explicit turn
errors, and tool names from `~/.cursor/projects/*/agent-transcripts`; it does
not retain tool arguments, tool outputs, or other record types. Known
@@ -79,6 +80,19 @@ optimization.
`tool_called` validation fail before Agent mode starts; use another backend for
those tasks. Cursor and the model provider selected by Cursor can receive the
resulting prompt content.
- The Pi backend sends prompts through the installed, authenticated Pi CLI to
the provider configured by the user. It disables tools, skills, context files,
extensions, prompt templates, themes, and session writes for these calls, but
retains the user's Pi authentication and model configuration. Pi's offline
startup mode prevents configured npm/git package installation, package
updates, and model-catalog refresh; it does not prevent the selected provider
call. These controls are not a guarantee of permanent or complete isolation.
- The Pi source retains user/assistant text, tool names, and lexical feedback
found in user text. It excludes thinking, tool arguments, tool outputs, images,
and unrelated metadata. The absolute project `cwd` from the session header is
retained for scope filtering and may appear in miner prompts sent to a real
backend and its provider. Known secret-shaped strings in retained message text
are redacted only as defense in depth.
- Outbound prompts are not currently guaranteed to be free of secrets. Do not
use a third-party provider on sensitive transcripts without reviewing the data
source and the provider's retention policy.
@@ -114,11 +128,13 @@ Common implemented flags include:
| Flag | Default | Purpose |
|---|---|---|
| `--backend mock\|claude\|codex\|cursor\|copilot\|handoff\|azure_openai` | `mock` | select who performs model calls |
| `--backend mock\|claude\|codex\|cursor\|copilot\|pi\|handoff\|azure_openai` | `mock` | select who performs model calls |
| `--model NAME` | backend default | select a backend-specific model |
| `--source claude\|codex\|cursor\|auto` | `claude` | select the transcript source; `auto` retains Codex-then-Claude precedence and does not select Cursor |
| `--source claude\|codex\|copilot\|cursor\|pi\|auto` | `claude` | select the transcript source; `auto` retains Codex-then-Claude precedence and does not select Copilot, Cursor, or Pi |
| `--cursor-home PATH` | `~/.cursor` | override the Cursor transcript home |
| `--cursor-path PATH` | auto-detect `cursor-agent` | select the Cursor Agent CLI executable |
| `--pi-home PATH` | `~/.pi` | select the parent directory containing `agent/sessions` |
| `--pi-path PATH` | auto-detect `pi` | select the Pi coding-agent CLI executable |
| `--project PATH` | current directory | select the project and invoked harvest scope |
| `--scope invoked\|all` | `invoked` | limit transcript harvesting |
| `--target-skill-path PATH` | managed skill | select a specific `SKILL.md` to stage/adopt |
@@ -153,6 +169,45 @@ python -m skillopt_sleep run --backend codex --project "$(pwd)" \
Preferences guide reflection but remain subject to the validation gate.
### Pi source and backend
Pi transcript harvesting is explicit: `--source pi` reads session JSONL files
below `~/.pi/agent/sessions`; use `--pi-home` to select the parent directory
that contains `agent/sessions`. This source does not require the Pi CLI or
provider authentication. It retains user/assistant text, tool names, and lexical
feedback found in user text, while excluding thinking, tool arguments, tool
outputs, images, and unrelated metadata. The absolute project `cwd` from the
session header is retained for scope filtering and may appear in miner prompts
sent to a real backend and its provider. Known secret-shaped strings in retained
message text are redacted as defense in depth, not as a guarantee. `--source auto` keeps Codex-then-Claude
precedence and does not select Pi.
The source and backend are independent. `--backend pi` uses a locally installed,
authenticated Pi CLI to make real model-provider calls for mining, replay,
judging, and reflection. Select another executable with `--pi-path` and a model
with `--model`:
```bash
python -m skillopt_sleep run --project "$(pwd)" \
--source pi --backend pi --pi-path /absolute/path/to/pi \
--model provider/model --max-sessions 5 --max-tasks 3 --progress
```
Pi calls disable tools, skills, context files, extensions, prompt templates,
themes, and session writes. They still use the user's Pi authentication and
model configuration. Pi's offline startup mode also prevents configured npm/git
package installation, package updates, and model-catalog refresh; it does not
prevent the selected provider call. This is bounded invocation setup rather
than permanent or complete isolation. Transcript-derived prompts reach the
provider configured in Pi; review that provider's data-retention and privacy
policy before using sensitive sessions.
The managed scheduler stores the selected backend but does not persist
`--source`, `--pi-home`, `--pi-path`, or `--model`. Before scheduling Pi, set
`transcript_source`, `pi_home`, `pi_path`, and `model` in
`~/.skillopt-sleep/config.json`; prefer an absolute `pi_path` and verify that the
scheduled account is authenticated.
### Cursor source and backend
Cursor transcript harvesting is explicit: use `--source cursor` rather than
@@ -221,7 +276,7 @@ the shipping CLI defaults.
## Safety summary
- Session harvesting is read-only.
- `mock` replay makes no provider calls.
- `mock` and `handoff` make no network calls.
- `run` stages proposals; `adopt` is the normal live-change boundary.
- Adoption backs up existing target files.
- `--max-sessions` and `--max-tasks` bound work, but the main CLI does not yet

15
plugins/openclaw/run_sleep.py
View File

@@ -35,14 +35,27 @@ _b._BACKENDS["openclaw-deepseek"] = OpenClawDeepSeekBackend
# Patch get_backend to know about our backend
_orig_get_backend = _b.get_backend
def get_backend(name, model="", codex_path="", cursor_path="", project_dir=""):
def get_backend(
name,
*,
model="",
claude_path="claude",
codex_path="",
pi_path="",
cursor_path="",
azure_endpoint="",
project_dir="",
):
if name == "openclaw-deepseek":
return OpenClawDeepSeekBackend(model=model or "deepseek-v4-pro")
return _orig_get_backend(
name,
model=model,
claude_path=claude_path,
codex_path=codex_path,
pi_path=pi_path,
cursor_path=cursor_path,
azure_endpoint=azure_endpoint,
project_dir=project_dir,
)

7
plugins/openclaw/skillopt_sleep_openclaw.py
View File

@@ -204,7 +204,12 @@ Return ONLY a single float 0.0-1.0 on one line. No explanation. No markdown.
rubric_text = ""
if failures:
rubric_text = f"\n\n## REFERENCE ANSWERS\n{chr(10).join(f'Q: {t.intent[:120]}\\nA: {t.reference}' for t, _ in failures[:3] if t.reference)}"
reference_answers = "\n".join(
f"Q: {t.intent[:120]}\nA: {t.reference}"
for t, _ in failures[:3]
if t.reference
)
rubric_text = f"\n\n## REFERENCE ANSWERS\n{reference_answers}"
sys = (
"You are SkillOpt-Sleep's bounded-edit optimizer. Your job is to propose 1-4 MINIMAL text edits to a skill or memory document "

5
skillopt_sleep/backend.py
View File

@@ -599,6 +599,10 @@ class PiCliBackend(CliBackend):
def _cached_call(self, key: str, prompt: str, *, max_tokens: int = 1024) -> str:
"""Do not make a transient Pi failure sticky in the response cache."""
if key in self._cache:
# A cached success must not expose an unrelated previous failure
# through diagnostics/evidence attached to this call.
self.last_call_error = ""
out = super()._cached_call(key, prompt, max_tokens=max_tokens)
if not out:
self._cache.pop(key, None)
@@ -640,6 +644,7 @@ class PiCliBackend(CliBackend):
env = os.environ.copy()
# A replay should contact the selected model provider, not Pi's update
# or install-telemetry endpoints during startup.
env["PI_OFFLINE"] = "1"
env["PI_SKIP_VERSION_CHECK"] = "1"
env["PI_TELEMETRY"] = "0"
try:

182
skillopt_sleep/harvest_pi.py
View File

@@ -5,16 +5,18 @@ Reads pi session transcript JSONL files (one per session, stored under
into :class:`SessionDigest` records without copying tool arguments, private
reasoning blocks (``thinking``), or raw tool outputs.
pi schema (verified against real transcripts):
pi schema (verified against the upstream session format):
* A session file is a JSONL stream of entries with a ``type`` discriminator.
* ``type == "session"`` — exactly one per file; carries ``cwd`` + ``timestamp``.
* Version 1 sessions are linear. Versions 2 and 3 form an append-only tree
using entry ``id`` / ``parentId`` fields; only the branch ending at the last
entry is the active conversation and is harvested.
* ``type == "message"`` — a conversational turn. ``message.role`` ∈
{user, assistant, toolResult}; ``message.content`` is either a string or a
list of content blocks. Block types include ``text`` (kept), ``thinking``
(private reasoning, skipped), and ``toolCall`` (carries ``name``).
* toolResult messages carry ``isError`` (bool) and ``toolName`` — a rare
per-call success/failure signal, surfaced here as a feedback signal so the
miner/gate can exploit checkable outcomes.
* toolResult messages can carry ``isError`` and ``toolName``. Tool names are
retained, while transient per-call errors are not treated as task feedback.
* Other types (``model_change``, ``thinking_level_change``, ``custom``, ...) are
metadata / tool-result payloads and are skipped for digestion.
@@ -22,6 +24,7 @@ This module performs NO writes and NO network calls.
"""
from __future__ import annotations
import json
import os
import re
from typing import Any, Iterable, List, Optional
@@ -30,41 +33,45 @@ from skillopt_sleep.harvest import (
_detect_feedback,
_is_headless_replay,
_is_meta_prompt,
_iter_jsonl,
_project_matches,
_text_from_content,
)
from skillopt_sleep.staging import redact_secrets
from skillopt_sleep.types import SessionDigest
# Mirror of skillopt_sleep.harvest_codex._SECRET_PATTERNS. Kept duplicated (not
# imported) so each harvester stays self-contained; if a third source appears,
# consider promoting these into a shared ``redact`` module.
_SECRET_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
(re.compile(r"sk-[A-Za-z0-9_-]{10,}"), "[REDACTED_OPENAI_KEY]"),
(re.compile(r"(?i)(Authorization:\s*Bearer\s+)[^\s\"']+"), r"\1[REDACTED]"),
(re.compile(r"(?i)(Authorization:\s*Basic\s+)[^\s\"']+"), r"\1[REDACTED]"),
(
re.compile(r"(?i)\b(api[_-]?key|token|password|secret)\b(\s*[:=]\s*)[^\s\"']+"),
r"\1\2[REDACTED]",
),
(
re.compile(r"(?i)\b(api[_-]?key|token|password|secret)\b(\s+)[^\s\"']+"),
r"\1\2[REDACTED]",
),
(
re.compile(
r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----",
re.DOTALL,
),
"[REDACTED_PRIVATE_KEY]",
),
)
def _redact_secrets(text: str) -> str:
for pattern, replacement in _SECRET_PATTERNS:
text = pattern.sub(replacement, text)
return text
"""Backward-compatible string wrapper around shared secret redaction."""
return str(redact_secrets(text))
def _sanitize_tool_name(name: str) -> str:
return re.sub(r"[^A-Za-z0-9_.:-]+", "_", str(name))[:80]
def _read_pi_jsonl(path: str) -> Optional[List[dict[str, Any]]]:
"""Read one Pi transcript, rejecting the whole session on corruption.
For tree-shaped v2/v3 sessions, silently skipping a malformed record could
turn an older entry into the apparent active leaf. Pi transcripts therefore
use stricter parsing than the legacy shared harvester: blank lines are fine,
but every non-blank line must be a JSON object and the complete file must be
readable.
"""
records: List[dict[str, Any]] = []
try:
with open(path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if not line:
continue
record = json.loads(line)
if not isinstance(record, dict):
return None
records.append(record)
except (OSError, UnicodeError, ValueError):
return None
return records
def _pi_tool_names_from_content(content: Any) -> List[str]:
@@ -76,12 +83,81 @@ def _pi_tool_names_from_content(content: Any) -> List[str]:
if isinstance(content, list):
for b in content:
if isinstance(b, dict) and b.get("type") == "toolCall" and b.get("name"):
names.append(str(b["name"]))
names.append(_sanitize_tool_name(str(b["name"])))
return names
def _sanitize_tool_name(name: str) -> str:
return re.sub(r"[^A-Za-z0-9_.:-]+", "_", str(name))[:80]
def _active_branch_entries(records: List[dict[str, Any]]) -> List[dict[str, Any]]:
"""Return the linear history for v1 or the active leaf branch for v2/v3.
Tree sessions are treated as an integrity boundary: a malformed graph is
skipped in full instead of exposing an arbitrary reachable suffix to the
optimizer. Multiple roots remain valid because navigating back to the
beginning of a Pi session can legitimately create one.
"""
if (
not records
or any(not isinstance(rec, dict) for rec in records)
or records[0].get("type") != "session"
):
return []
if sum(rec.get("type") == "session" for rec in records) != 1:
return []
header = records[0]
entries = records[1:]
header_id = header.get("id")
if not isinstance(header_id, str) or not header_id:
return []
# Legacy v1 headers predate the version field; Pi itself interprets an
# absent field as v1 and migrates those sessions on load.
if "version" not in header:
return entries
version = header["version"]
# ``bool`` is an ``int`` subclass in Python but is not a schema version.
if type(version) is not int or version not in (1, 2, 3):
return []
if version == 1:
return entries
if not entries:
return []
by_id: dict[str, dict[str, Any]] = {}
for rec in entries:
entry_id = rec.get("id")
if not isinstance(entry_id, str) or not entry_id or entry_id in by_id:
return []
if "parentId" not in rec:
return []
parent_id = rec["parentId"]
if parent_id is not None:
if not isinstance(parent_id, str) or not parent_id:
return []
# Pi's writer is append-only: every child is appended after its
# parent. Reject forward references rather than accepting a graph
# that the official writer cannot produce.
if parent_id not in by_id:
return []
by_id[entry_id] = rec
branch: List[dict[str, Any]] = []
seen: set[str] = set()
current: Optional[dict[str, Any]] = entries[-1]
while current is not None:
entry_id = current["id"]
if entry_id in seen:
return []
seen.add(entry_id)
branch.append(current)
parent_id = current["parentId"]
if parent_id is None:
break
current = by_id[parent_id]
branch.reverse()
return branch
def _dedup(xs: Iterable[str]) -> List[str]:
@@ -96,6 +172,10 @@ def _dedup(xs: Iterable[str]) -> List[str]:
def digest_pi_session(path: str, project: str = "") -> Optional[SessionDigest]:
"""Build a :class:`SessionDigest` from one pi session transcript."""
records = _read_pi_jsonl(path)
if not records:
return None
active_entries = _active_branch_entries(records)
session_id = os.path.splitext(os.path.basename(path))[0]
started = ""
ended = ""
@@ -107,19 +187,23 @@ def digest_pi_session(path: str, project: str = "") -> Optional[SessionDigest]:
n_user = 0
n_asst = 0
for rec in _iter_jsonl(path):
header = records[0] if records[0].get("type") == "session" else None
if isinstance(header, dict):
ts = header.get("timestamp")
if isinstance(ts, str) and ts:
started = ts
ended = ts
cwd = header.get("cwd")
if isinstance(cwd, str) and cwd:
session_project = cwd
for rec in active_entries:
rtype = rec.get("type")
ts = rec.get("timestamp")
if isinstance(ts, str) and ts:
if not started:
started = ts
ended = ts
# cwd lives on the `session` entry, not on individual messages.
if rtype == "session":
cwd = rec.get("cwd")
if isinstance(cwd, str) and cwd and not session_project:
session_project = cwd
continue
if rtype != "message":
continue
@@ -199,15 +283,21 @@ def harvest_pi(
if not os.path.isdir(sessions_dir):
return digests
paths: List[str] = []
paths: List[tuple[float, str]] = []
for root, _dirs, files in os.walk(sessions_dir):
for fn in files:
if fn.endswith(".jsonl"):
paths.append(os.path.join(root, fn))
paths.sort(key=lambda p: os.path.getmtime(p), reverse=True)
path = os.path.join(root, fn)
try:
paths.append((os.path.getmtime(path), path))
except OSError:
# A session can disappear while Pi rotates or cleans its
# store; skip that file without aborting the whole harvest.
continue
paths.sort(key=lambda item: item[0], reverse=True)
project_hint = invoked_project if scope == "invoked" else ""
for path in paths:
for _mtime, path in paths:
digest = digest_pi_session(path, project=project_hint)
if digest is None:
continue

261
tests/test_backend_pi.py
View File

@@ -1,9 +1,18 @@
"""Tests for the pi CLI backend (`--backend pi`)."""
from __future__ import annotations
import os
import subprocess
from unittest import mock
from skillopt_sleep.backend import PiCliBackend, get_backend
from skillopt_sleep.backend import (
_NO_WINDOW,
DualBackend,
PiCliBackend,
build_backend,
get_backend,
)
from skillopt_sleep.types import TaskRecord
class _FakeProc:
@@ -33,7 +42,7 @@ def test_call_builds_isolated_command_and_returns_stdout():
def fake_run(cmd, **kwargs):
captured["cmd"] = cmd
captured["cwd"] = kwargs.get("cwd")
captured.update(kwargs)
return _FakeProc("answer text")
with mock.patch("skillopt_sleep.backend.subprocess.run", side_effect=fake_run):
@@ -42,42 +51,274 @@ def test_call_builds_isolated_command_and_returns_stdout():
assert out == "answer text"
cmd = captured["cmd"]
assert cmd[0:2] == ["/usr/local/bin/pi", "-p"]
# isolation flags must be present (no ambient skills/context/tools)
# Prompts go over stdin rather than argv (important for long/Windows calls).
assert captured["input"] == "do the thing"
assert "do the thing" not in cmd
# Isolation flags must be present (no ambient skills/context/tools).
assert "--no-tools" in cmd
assert "--no-skills" in cmd
assert "--no-context-files" in cmd
assert "--no-extensions" in cmd
assert "--no-prompt-templates" in cmd
assert "--no-themes" in cmd
assert "--no-session" in cmd
assert cmd[cmd.index("--system-prompt") + 1] == ""
assert cmd[cmd.index("--append-system-prompt") + 1] == ""
assert "--model" in cmd and "zai/glm-5.2" in cmd
assert cmd[-1] == "do the thing"
# ran from a clean temp cwd, not inherited
assert captured["cwd"] is not None and captured["cwd"] != ""
assert captured["creationflags"] == _NO_WINDOW
assert captured["env"]["PI_OFFLINE"] == "1"
assert captured["env"]["PI_SKIP_VERSION_CHECK"] == "1"
assert captured["env"]["PI_TELEMETRY"] == "0"
def test_call_detects_auth_error_and_logs():
def test_path_expands_and_resolves_windows_shim(monkeypatch):
monkeypatch.setattr(os.path, "expanduser", lambda value: "/home/u/bin/pi" if value == "~/bin/pi" else value)
with mock.patch("shutil.which", return_value="C:\\npm\\pi.CMD") as which:
be = PiCliBackend(pi_path="~/bin/pi")
which.assert_called_once_with("/home/u/bin/pi")
assert be.pi_path == "C:\\npm\\pi.CMD"
def test_call_records_auth_error_from_nonzero_exit():
be = PiCliBackend()
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
return_value=_FakeProc("", stderr="Authentication required: not logged in"),
return_value=_FakeProc(
"", stderr="No API key found for openai", returncode=1
),
):
out = be._call("hi")
assert out == "" # empty stdout
assert "Authentication required" in be.last_call_error
assert "No API key found for openai" in be.last_call_error
def test_call_records_nonzero_exit():
def test_successful_short_answers_that_mention_cli_errors_are_preserved():
answers = (
"Not logged in means the session needs authentication.",
"Authentication required is an error message.",
"Invalid API key should be reported to the user.",
"Unauthorized requests receive HTTP 401.",
"The provider not found error comes from configuration.",
"No provider is required for this local operation.",
)
be = PiCliBackend()
proc = _FakeProc("", stderr="pi: unknown option", returncode=2)
for answer in answers:
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
return_value=_FakeProc(answer),
):
assert be._call("explain the error") == answer
assert be.last_call_error == ""
def test_call_records_nonzero_exit_even_with_stdout():
be = PiCliBackend()
proc = _FakeProc("misleading answer", stderr="pi: unknown option", returncode=2)
with mock.patch("skillopt_sleep.backend.subprocess.run", return_value=proc):
assert be._call("hi") == ""
assert "exited 2" in be.last_call_error
assert "misleading answer" not in be.last_call_error
def test_call_records_empty_success_response():
be = PiCliBackend()
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
return_value=_FakeProc(" \n"),
):
assert be._call("hi") == ""
assert be.last_call_error == "Pi CLI returned an empty response"
def test_call_redacts_stderr_from_empty_success(caplog):
be = PiCliBackend()
secret = "sk-1234567890abcdefghij"
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
return_value=_FakeProc("", stderr=f"warning: key={secret}"),
):
assert be._call("hi") == ""
assert "warning" in be.last_call_error
assert secret not in be.last_call_error
assert secret not in caplog.text
def test_call_preserves_nonempty_success_with_stderr_warning():
be = PiCliBackend()
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
return_value=_FakeProc("answer", stderr="provider warning"),
):
assert be._call("hi") == "answer"
assert be.last_call_error == ""
def test_call_redacts_secrets_in_error(caplog):
be = PiCliBackend()
secret = "sk-1234567890abcdefghij"
proc = _FakeProc("", stderr=f"Invalid API key: {secret}", returncode=1)
with mock.patch("skillopt_sleep.backend.subprocess.run", return_value=proc):
assert be._call("hi") == ""
assert secret not in be.last_call_error
assert secret not in caplog.text
def test_call_records_timeout():
be = PiCliBackend(timeout=1)
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
side_effect=__import__("subprocess").TimeoutExpired("pi", 1),
side_effect=subprocess.TimeoutExpired("pi", 1),
):
assert be._call("hi") == ""
assert "timed out" in be.last_call_error
def test_call_normalizes_and_redacts_timeout_bytes(caplog):
be = PiCliBackend(timeout=1)
secret = "sk-1234567890abcdefghij"
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
side_effect=subprocess.TimeoutExpired(
"pi", 1, stderr=f"API key: {secret}".encode()
),
):
assert be._call("hi") == ""
assert "timed out" in be.last_call_error
assert "b'" not in be.last_call_error
assert secret not in be.last_call_error
assert secret not in caplog.text
def test_call_records_and_redacts_spawn_failure(caplog):
be = PiCliBackend()
secret = "sk-1234567890abcdefghij"
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
side_effect=OSError(f"cannot launch with token={secret}"),
):
assert be._call("hi") == ""
assert secret not in be.last_call_error
assert secret not in caplog.text
def test_failed_empty_response_is_not_cached():
be = PiCliBackend()
with mock.patch.object(be, "_call", side_effect=["", "recovered"]) as call:
assert be._cached_call("attempt:key", "prompt") == ""
assert be._cached_call("attempt:key", "prompt") == "recovered"
assert call.call_count == 2
def test_success_clears_previous_call_error():
be = PiCliBackend()
be.last_call_error = "an older failure"
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
return_value=_FakeProc("recovered"),
):
assert be._call("hi") == "recovered"
assert be.last_call_error == ""
def test_cache_and_retry_token_accounting():
prompt = "p" * 20
response = "r" * 12
be = PiCliBackend()
with mock.patch.object(be, "_call", return_value=response) as call:
assert be._cached_call("attempt:success", prompt) == response
spent = be.tokens_used()
assert spent == len(prompt) // 4 + len(response) // 4
assert be._cached_call("attempt:success", prompt) == response
assert call.call_count == 1
assert be.tokens_used() == spent
retrying = PiCliBackend()
with mock.patch.object(retrying, "_call", side_effect=["", response]) as call:
assert retrying._cached_call("attempt:retry", prompt) == ""
after_failure = retrying.tokens_used()
assert after_failure == len(prompt) // 4
assert retrying._cached_call("attempt:retry", prompt) == response
after_recovery = retrying.tokens_used()
assert retrying._cached_call("attempt:retry", prompt) == response
assert call.call_count == 2
assert after_recovery == after_failure + len(prompt) // 4 + len(response) // 4
assert retrying.tokens_used() == after_recovery
def test_cached_success_clears_stale_call_error_without_spending_tokens():
be = PiCliBackend()
prompt = "cached prompt"
with mock.patch.object(be, "_call", return_value="cached answer") as call:
assert be._cached_call("attempt:cached", prompt) == "cached answer"
spent = be.tokens_used()
be.last_call_error = "unrelated later failure"
assert be._cached_call("attempt:cached", prompt) == "cached answer"
assert call.call_count == 1
assert be.last_call_error == ""
assert be.tokens_used() == spent
def test_dual_backend_tokens_are_summed_once():
target = PiCliBackend()
optimizer = PiCliBackend()
dual = DualBackend(target=target, optimizer=optimizer)
with (
mock.patch.object(target, "_call", return_value="t" * 8),
mock.patch.object(optimizer, "_call", return_value="o" * 12),
):
target._cached_call("attempt:target", "a" * 16)
optimizer._cached_call("judge:optimizer", "b" * 20)
expected = target.tokens_used() + optimizer.tokens_used()
assert expected > 0
assert dual.tokens_used() == expected
def test_dual_attempt_failure_retries_then_caches_without_double_counting():
target = PiCliBackend()
optimizer = PiCliBackend()
dual = DualBackend(target=target, optimizer=optimizer)
task = TaskRecord(id="pi-retry", project="/repo", intent="fix the test")
failed = _FakeProc("", stderr="Authentication required", returncode=1)
recovered = _FakeProc("fixed response")
with mock.patch(
"skillopt_sleep.backend.subprocess.run",
side_effect=[failed, recovered],
) as run:
assert dual.attempt(task, "", "") == ""
assert target.last_call_error
after_failure = dual.tokens_used()
assert dual.attempt(task, "", "") == "fixed response"
assert target.last_call_error == ""
after_recovery = dual.tokens_used()
assert dual.attempt(task, "", "") == "fixed response"
assert run.call_count == 2
assert after_recovery > after_failure
assert dual.tokens_used() == after_recovery
assert optimizer.tokens_used() == 0
def test_pi_path_reaches_single_and_dual_backends():
single = build_backend(backend="pi", pi_path="/opt/pi")
assert isinstance(single, PiCliBackend)
assert single.pi_path == "/opt/pi"
dual = build_backend(
backend="mock",
optimizer_backend="pi",
target_backend="pi",
pi_path="/opt/pi",
)
assert isinstance(dual.optimizer, PiCliBackend)
assert isinstance(dual.target, PiCliBackend)
assert dual.optimizer.pi_path == dual.target.pi_path == "/opt/pi"

308
tests/test_harvest_pi.py
View File

@@ -2,13 +2,14 @@
from __future__ import annotations
import json
import os
from skillopt_sleep.harvest_pi import _redact_secrets, digest_pi_session, harvest_pi
def _write_session(tmp_path, slug, name, entries):
d = tmp_path / slug
d.mkdir(parents=True)
d.mkdir(parents=True, exist_ok=True)
p = d / f"{name}.jsonl"
with open(p, "w") as f:
for rec in entries:
@@ -17,9 +18,9 @@ def _write_session(tmp_path, slug, name, entries):
PI_SESSION = [
{"type": "session", "version": 1, "id": "s1", "timestamp": "2026-06-23T11:52:04.333Z", "cwd": "/home/u/proj"},
{"type": "model_change", "id": "m", "timestamp": "2026-06-23T11:52:05.000Z", "modelId": "gpt-x"},
{"type": "message", "id": "a1", "parentId": "s1", "timestamp": "2026-06-23T11:52:06.000Z",
{"type": "session", "version": 3, "id": "s1", "timestamp": "2026-06-23T11:52:04.333Z", "cwd": "/home/u/proj"},
{"type": "model_change", "id": "m", "parentId": None, "timestamp": "2026-06-23T11:52:05.000Z", "modelId": "gpt-x"},
{"type": "message", "id": "a1", "parentId": "m", "timestamp": "2026-06-23T11:52:06.000Z",
"message": {"role": "user", "content": [{"type": "text", "text": "fix the failing tests"}]}},
{"type": "message", "id": "a2", "parentId": "a1", "timestamp": "2026-06-23T11:52:07.000Z",
"message": {"role": "assistant", "content": [
@@ -74,7 +75,304 @@ def test_harvest_scope_filter(tmp_path):
assert invoked[0].project == "/home/u/proj"
def test_harvest_skips_session_removed_during_discovery(tmp_path, monkeypatch):
p = _write_session(tmp_path, "vanishing-project", "vanishing", PI_SESSION)
real_getmtime = os.path.getmtime
def remove_then_stat(path):
if os.fspath(path) == os.fspath(p):
os.unlink(path)
return real_getmtime(path)
monkeypatch.setattr(os.path, "getmtime", remove_then_stat)
assert harvest_pi(str(tmp_path), scope="all") == []
def test_secret_redaction():
out = _redact_secrets("Authorization: Bearer sk-1234567890abcdefghij")
assert "sk-1234567890abcdefghij" not in out
assert "[REDACTED]" in out
assert "[REDACTED_OPENAI_KEY]" in out
def test_shared_secret_redaction_covers_github_tokens():
token = "ghp_1234567890abcdefghijklmnop"
out = _redact_secrets(f"token leaked in output: {token}")
assert token not in out
assert "[REDACTED_GITHUB_TOKEN]" in out
def test_v1_session_remains_linear(tmp_path):
entries = [
{"type": "session", "version": 1, "id": "v1", "timestamp": "2026-06-23T12:00:00Z", "cwd": "/v1/project"},
{"type": "message", "timestamp": "2026-06-23T12:00:01Z", "message": {"role": "user", "content": "first request"}},
{"type": "message", "timestamp": "2026-06-23T12:00:02Z", "message": {"role": "assistant", "content": "first answer"}},
{"type": "message", "timestamp": "2026-06-23T12:00:03Z", "message": {"role": "user", "content": "second request"}},
{"type": "message", "timestamp": "2026-06-23T12:00:04Z", "message": {"role": "assistant", "content": "second answer"}},
]
p = _write_session(tmp_path, "v1-project", "linear", entries)
d = digest_pi_session(p)
assert d is not None
assert d.user_prompts == ["first request", "second request"]
assert d.assistant_finals == ["first answer", "second answer"]
def test_unknown_session_version_fails_closed(tmp_path):
entries = [
{"type": "session", "version": 4, "id": "future", "timestamp": "2026-06-23T12:00:00Z", "cwd": "/future/project"},
{"type": "message", "id": "u", "parentId": None, "timestamp": "2026-06-23T12:00:01Z", "message": {"role": "user", "content": "future request"}},
{"type": "message", "id": "a", "parentId": "u", "timestamp": "2026-06-23T12:00:02Z", "message": {"role": "assistant", "content": "future answer"}},
]
p = _write_session(tmp_path, "future-project", "unknown", entries)
assert digest_pi_session(p) is None
def test_v3_harvests_only_the_active_branch(tmp_path):
entries = [
{"type": "session", "version": 3, "id": "session", "timestamp": "2026-06-23T12:00:00Z", "cwd": "/branch/project"},
{"type": "message", "id": "root", "parentId": None, "timestamp": "2026-06-23T12:00:01Z", "message": {"role": "user", "content": "shared request"}},
{"type": "message", "id": "old-user", "parentId": "root", "timestamp": "2026-06-23T12:00:02Z", "message": {"role": "user", "content": "abandoned prompt, this is wrong"}},
{"type": "message", "id": "old-answer", "parentId": "old-user", "timestamp": "2026-06-23T12:00:03Z", "message": {"role": "assistant", "content": [{"type": "text", "text": "abandoned final"}, {"type": "toolCall", "name": "abandoned/tool"}]}},
{"type": "message", "id": "old-result", "parentId": "old-answer", "timestamp": "2026-06-23T12:00:04Z", "message": {"role": "toolResult", "toolName": "abandoned result", "content": "old output"}},
{"type": "message", "id": "new-user", "parentId": "root", "timestamp": "2026-06-23T12:00:05Z", "message": {"role": "user", "content": "active prompt"}},
{"type": "message", "id": "new-answer", "parentId": "new-user", "timestamp": "2026-06-23T12:00:06Z", "message": {"role": "assistant", "content": [{"type": "text", "text": "active final"}, {"type": "toolCall", "name": "active-tool"}]}},
]
p = _write_session(tmp_path, "branch-project", "branching", entries)
d = digest_pi_session(p)
assert d is not None
assert d.user_prompts == ["shared request", "active prompt"]
assert d.assistant_finals == ["active final"]
assert d.tools_used == ["active-tool"]
assert not d.feedback_signals
assert d.started_at == "2026-06-23T12:00:00Z"
assert d.ended_at == "2026-06-23T12:00:06Z"
def test_v3_cycle_fails_closed(tmp_path):
entries = [
{"type": "session", "version": 3, "id": "cycle", "timestamp": "2026-06-23T12:00:00Z", "cwd": "/cycle/project"},
{"type": "message", "id": "a", "parentId": "b", "timestamp": "2026-06-23T12:00:01Z", "message": {"role": "user", "content": "cycle request"}},
{"type": "message", "id": "b", "parentId": "a", "timestamp": "2026-06-23T12:00:10Z", "message": {"role": "assistant", "content": "cycle answer"}},
]
p = _write_session(tmp_path, "cycle-project", "cycle", entries)
assert digest_pi_session(p) is None
def test_v3_missing_parent_fails_closed(tmp_path):
entries = [
{"type": "session", "version": 3, "id": "missing", "timestamp": "2026-06-23T12:00:00Z", "cwd": "/missing/project"},
{"type": "message", "id": "unrelated", "parentId": None, "timestamp": "2026-06-23T12:00:01Z", "message": {"role": "user", "content": "abandoned prompt"}},
{"type": "message", "id": "reachable-user", "parentId": "missing", "timestamp": "2026-06-23T12:00:05Z", "message": {"role": "user", "content": "reachable request"}},
{"type": "message", "id": "leaf", "parentId": "reachable-user", "timestamp": "2026-06-23T12:00:10Z", "message": {"role": "assistant", "content": "reachable answer"}},
]
p = _write_session(tmp_path, "missing-project", "missing", entries)
assert digest_pi_session(p) is None
def test_v2_harvests_only_the_active_branch(tmp_path):
entries = [
{"type": "session", "version": 2, "id": "v2", "timestamp": "2026-06-23T12:00:00Z", "cwd": "/v2/project"},
{"type": "message", "id": "root", "parentId": None, "timestamp": "2026-06-23T12:00:01Z", "message": {"role": "user", "content": "shared"}},
{"type": "message", "id": "old", "parentId": "root", "timestamp": "2026-06-23T12:00:02Z", "message": {"role": "assistant", "content": "abandoned"}},
{"type": "message", "id": "active", "parentId": "root", "timestamp": "2026-06-23T12:00:05Z", "message": {"role": "assistant", "content": "active"}},
]
p = _write_session(tmp_path, "v2-project", "branching", entries)
d = digest_pi_session(p)
assert d is not None
assert d.user_prompts == ["shared"]
assert d.assistant_finals == ["active"]
def test_tree_session_requires_first_unique_header(tmp_path):
misplaced = [
{"type": "message", "id": "u", "parentId": None, "message": {"role": "user", "content": "request"}},
{"type": "session", "version": 3, "id": "misplaced", "cwd": "/bad/project"},
]
duplicate = [PI_SESSION[0], dict(PI_SESSION[0]), *PI_SESSION[1:]]
p1 = _write_session(tmp_path, "bad-project", "misplaced", misplaced)
p2 = _write_session(tmp_path, "bad-project", "duplicate", duplicate)
assert digest_pi_session(p1) is None
assert digest_pi_session(p2) is None
def test_tree_session_rejects_invalid_ids_and_parents(tmp_path):
base_header = {"type": "session", "version": 3, "id": "bad", "cwd": "/bad/project"}
cases = {
"missing-id": [{"type": "message", "parentId": None, "message": {"role": "user", "content": "request"}}],
"duplicate-id": [
{"type": "message", "id": "same", "parentId": None, "message": {"role": "user", "content": "request"}},
{"type": "message", "id": "same", "parentId": None, "message": {"role": "assistant", "content": "answer"}},
],
"missing-parent-field": [{"type": "message", "id": "u", "message": {"role": "user", "content": "request"}}],
"empty-parent": [{"type": "message", "id": "u", "parentId": "", "message": {"role": "user", "content": "request"}}],
"non-string-parent": [{"type": "message", "id": "u", "parentId": 7, "message": {"role": "user", "content": "request"}}],
}
for name, body in cases.items():
p = _write_session(tmp_path, f"bad-{name}", name, [base_header, *body])
assert digest_pi_session(p) is None, name
def test_tree_session_rejects_complete_forward_reference(tmp_path):
entries = [
{"type": "session", "version": 3, "id": "forward", "cwd": "/forward/project"},
{"type": "message", "id": "old-child", "parentId": "old-root", "message": {"role": "assistant", "content": "abandoned answer"}},
{"type": "message", "id": "old-root", "parentId": None, "message": {"role": "user", "content": "abandoned request"}},
{"type": "message", "id": "active-root", "parentId": None, "message": {"role": "user", "content": "active request"}},
{"type": "message", "id": "active-leaf", "parentId": "active-root", "message": {"role": "assistant", "content": "active answer"}},
]
p = _write_session(tmp_path, "forward-project", "forward", entries)
assert digest_pi_session(p) is None
def test_corrupt_abandoned_branch_fails_closed(tmp_path):
header = {"type": "session", "version": 3, "id": "corrupt", "cwd": "/corrupt/project"}
active = [
{"type": "message", "id": "active-root", "parentId": None, "message": {"role": "user", "content": "active request"}},
{"type": "message", "id": "active-leaf", "parentId": "active-root", "message": {"role": "assistant", "content": "active answer"}},
]
corrupt_branches = {
"cycle": [
{"type": "message", "id": "bad-a", "parentId": "bad-b", "message": {"role": "user", "content": "abandoned"}},
{"type": "message", "id": "bad-b", "parentId": "bad-a", "message": {"role": "assistant", "content": "abandoned"}},
],
"missing-parent": [
{"type": "message", "id": "bad-child", "parentId": "absent", "message": {"role": "assistant", "content": "abandoned"}},
],
}
for name, branch in corrupt_branches.items():
p = _write_session(
tmp_path,
f"corrupt-{name}",
name,
[header, *branch, *active],
)
assert digest_pi_session(p) is None, name
def test_legacy_session_without_version_remains_linear(tmp_path):
entries = [
{"type": "session", "id": "legacy", "cwd": "/legacy/project"},
{"type": "message", "message": {"role": "user", "content": "request"}},
{"type": "message", "message": {"role": "assistant", "content": "answer"}},
]
p = _write_session(tmp_path, "legacy-project", "legacy", entries)
d = digest_pi_session(p)
assert d is not None
assert d.user_prompts == ["request"]
assert d.assistant_finals == ["answer"]
def test_explicit_session_version_requires_supported_integer(tmp_path):
body = [
{"type": "message", "id": "u", "parentId": None, "message": {"role": "user", "content": "request"}},
{"type": "message", "id": "a", "parentId": "u", "message": {"role": "assistant", "content": "answer"}},
]
for name, version in (("null", None), ("bool", True), ("float", 3.0), ("string", "3")):
header = {"type": "session", "version": version, "id": name, "cwd": "/bad/project"}
p = _write_session(tmp_path, f"bad-version-{name}", name, [header, *body])
assert digest_pi_session(p) is None, name
def test_malformed_final_jsonl_line_fails_closed(tmp_path):
p = tmp_path / "truncated.jsonl"
with open(p, "w", encoding="utf-8") as f:
for record in PI_SESSION:
f.write(json.dumps(record) + "\n")
f.write('{"type":"message","id":"new-active-leaf"')
assert digest_pi_session(str(p)) is None
def test_malformed_middle_jsonl_line_fails_closed(tmp_path):
p = tmp_path / "corrupt-middle.jsonl"
with open(p, "w", encoding="utf-8") as f:
f.write(json.dumps(PI_SESSION[0]) + "\n")
f.write("not-json\n")
for record in PI_SESSION[1:]:
f.write(json.dumps(record) + "\n")
assert digest_pi_session(str(p)) is None
def test_non_object_jsonl_record_fails_closed(tmp_path):
p = tmp_path / "non-object.jsonl"
with open(p, "w", encoding="utf-8") as f:
f.write(json.dumps(PI_SESSION[0]) + "\n")
f.write("[]\n")
for record in PI_SESSION[1:]:
f.write(json.dumps(record) + "\n")
assert digest_pi_session(str(p)) is None
def test_blank_jsonl_lines_remain_valid(tmp_path):
p = tmp_path / "blank-lines.jsonl"
with open(p, "w", encoding="utf-8") as f:
f.write("\n")
for record in PI_SESSION:
f.write(json.dumps(record) + "\n\n")
digest = digest_pi_session(str(p))
assert digest is not None
assert digest.user_prompts == ["fix the failing tests", "thanks, that works now"]
def test_tree_session_allows_multiple_roots(tmp_path):
entries = [
{"type": "session", "version": 3, "id": "roots", "cwd": "/roots/project"},
{"type": "message", "id": "old-root", "parentId": None, "message": {"role": "user", "content": "abandoned"}},
{"type": "message", "id": "new-root", "parentId": None, "message": {"role": "user", "content": "active request"}},
{"type": "message", "id": "leaf", "parentId": "new-root", "message": {"role": "assistant", "content": "active answer"}},
]
p = _write_session(tmp_path, "roots-project", "roots", entries)
d = digest_pi_session(p)
assert d is not None
assert d.user_prompts == ["active request"]
assert d.assistant_finals == ["active answer"]
def test_tool_call_name_is_sanitized(tmp_path):
entries = [
{"type": "session", "version": 1, "id": "tool-call", "timestamp": "2026-06-23T12:00:00Z", "cwd": "/tool/project"},
{"type": "message", "timestamp": "2026-06-23T12:00:01Z", "message": {"role": "user", "content": "run a tool"}},
{"type": "message", "timestamp": "2026-06-23T12:00:10Z", "message": {"role": "assistant", "content": [{"type": "toolCall", "name": "bad tool/<arg>"}]}},
]
p = _write_session(tmp_path, "tool-project", "tool-call", entries)
d = digest_pi_session(p)
assert d is not None
assert d.tools_used == ["bad_tool_arg_"]
def test_tool_result_name_is_sanitized(tmp_path):
entries = [
{"type": "session", "version": 1, "id": "tool-result", "timestamp": "2026-06-23T12:00:00Z", "cwd": "/tool/project"},
{"type": "message", "timestamp": "2026-06-23T12:00:01Z", "message": {"role": "user", "content": "run a tool"}},
{"type": "message", "timestamp": "2026-06-23T12:00:10Z", "message": {"role": "toolResult", "toolName": "bad result/<arg>", "content": "done"}},
]
p = _write_session(tmp_path, "tool-project", "tool-result", entries)
d = digest_pi_session(p)
assert d is not None
assert d.tools_used == ["bad_result_arg_"]

17
tests/test_model_change_warning.py
View File

@@ -102,6 +102,23 @@ def test_model_key_resolution_failure_uses_safe_config_fallback(monkeypatch) ->
)
def test_model_key_forwards_pi_executable_path(monkeypatch) -> None:
cycle_module = importlib.import_module("skillopt_sleep.cycle")
captured = {}
real_build = cycle_module.build_backend
def capture_without_recursion(**kwargs):
captured.update(kwargs)
return real_build(backend="mock")
monkeypatch.setattr(cycle_module, "build_backend", capture_without_recursion)
cfg = load_config(backend="pi", pi_path="/opt/custom/pi")
assert cycle_module._make_model_key(cfg) == "mock::"
assert captured["pi_path"] == "/opt/custom/pi"
def test_legacy_unresolved_model_key_migrates_without_false_warning(
tmp_path, capsys
) -> None:

83
tests/test_pi_integration.py Normal file
View File

@@ -0,0 +1,83 @@
"""Configuration and source-routing coverage for Pi integration."""
from __future__ import annotations
import argparse
import os
from unittest import mock
from skillopt_sleep.__main__ import _add_common, _cfg_from_args
from skillopt_sleep.config import load_config
from skillopt_sleep.harvest_sources import harvest_for_config
from skillopt_sleep.types import SessionDigest
def test_cli_maps_pi_backend_source_and_paths(monkeypatch):
parser = argparse.ArgumentParser()
_add_common(parser)
args = parser.parse_args(
[
"--backend",
"pi",
"--source",
"pi",
"--pi-home",
"~/.pi-test",
"--pi-path",
"~/bin/pi",
]
)
monkeypatch.setattr("skillopt_sleep.config._user_config_path", lambda: None)
cfg = _cfg_from_args(args)
assert cfg.get("backend") == "pi"
assert cfg.get("transcript_source") == "pi"
assert cfg.get("pi_home") == os.path.abspath(os.path.expanduser("~/.pi-test"))
assert cfg.get("pi_path") == os.path.abspath(os.path.expanduser("~/bin/pi"))
assert cfg.pi_sessions_dir == os.path.join(
os.path.abspath(os.path.expanduser("~/.pi-test")), "agent", "sessions"
)
def test_explicit_pi_source_routes_only_to_pi_harvester():
cfg = load_config(
transcript_source="pi",
projects="invoked",
invoked_project="/repo/project",
pi_home="/tmp/pi-home",
)
expected = [SessionDigest(session_id="pi-session", project="/repo/project")]
with (
mock.patch("skillopt_sleep.harvest_sources.harvest_pi", return_value=expected) as pi,
mock.patch("skillopt_sleep.harvest_sources.harvest_codex") as codex,
mock.patch("skillopt_sleep.harvest_sources.harvest") as claude,
):
actual = harvest_for_config(cfg, since_iso="2026-01-01T00:00:00Z", limit=3)
assert actual == expected
pi.assert_called_once_with(
"/tmp/pi-home/agent/sessions",
scope="invoked",
invoked_project="/repo/project",
since_iso="2026-01-01T00:00:00Z",
limit=3,
)
codex.assert_not_called()
claude.assert_not_called()
def test_auto_source_does_not_silently_add_pi_precedence():
cfg = load_config(
transcript_source="auto",
projects="invoked",
invoked_project="/repo/project",
)
expected = [SessionDigest(session_id="claude-session", project="/repo/project")]
with (
mock.patch("skillopt_sleep.harvest_sources.harvest_codex", return_value=[]),
mock.patch("skillopt_sleep.harvest_sources.harvest", return_value=expected),
mock.patch("skillopt_sleep.harvest_sources.harvest_pi") as pi,
):
assert harvest_for_config(cfg) == expected
pi.assert_not_called()

30
tests/test_plugin_sync.py
View File

@@ -4,6 +4,8 @@ Run: python3 -m pytest tests/test_plugin_sync.py -v
"""
import json
import os
import subprocess
import sys
import unittest
REPO = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
@@ -107,12 +109,40 @@ class TestPluginParity(unittest.TestCase):
def test_openclaw_wrapper_matches_shared_backend_signature(self):
text = _read(OPENCLAW_RUNNER)
self.assertIn('claude_path="claude"', text)
self.assertIn('pi_path=""', text)
self.assertIn('cursor_path=""', text)
self.assertIn('azure_endpoint=""', text)
self.assertIn('project_dir=""', text)
self.assertIn("claude_path=claude_path", text)
self.assertIn("pi_path=pi_path", text)
self.assertIn("cursor_path=cursor_path", text)
self.assertIn("azure_endpoint=azure_endpoint", text)
self.assertIn("project_dir=project_dir", text)
self.assertNotIn("**kwargs", text)
script = f"""
import runpy
import sys
sys.path.insert(0, {os.path.dirname(OPENCLAW_RUNNER)!r})
runpy.run_path({OPENCLAW_RUNNER!r}, run_name="openclaw_runner_test")
from skillopt_sleep.backend import build_backend
backend = build_backend(
backend="mock",
pi_path="unused-pi",
azure_endpoint="https://unused.invalid",
)
assert backend.name == "mock", backend.name
"""
result = subprocess.run(
[sys.executable, "-c", script],
cwd=REPO,
capture_output=True,
text=True,
check=False,
)
self.assertEqual(result.returncode, 0, result.stderr)
def test_all_skill_mds_mention_all_backends(self):
for name, path in PLUGIN_SKILL_MDS.items():
text = _read(path)