From 9baa1e409619446823350bd325fa787128631a99 Mon Sep 17 00:00:00 2001 From: TaoRunguo Date: Thu, 3 Sep 2026 16:08:11 +0800 Subject: [PATCH 1/2] feat(sleep): add DeepSeek Harness transcript source --- docs/reference/cli.md | 3 +- docs/sleep/README.md | 42 +++ pyproject.toml | 2 + skillopt_sleep/__main__.py | 9 +- skillopt_sleep/config.py | 15 +- skillopt_sleep/harvest_dsh.py | 499 ++++++++++++++++++++++++++++++ skillopt_sleep/harvest_sources.py | 9 + tests/test_harvest_dsh.py | 371 ++++++++++++++++++++++ 8 files changed, 946 insertions(+), 4 deletions(-) create mode 100644 skillopt_sleep/harvest_dsh.py create mode 100644 tests/test_harvest_dsh.py diff --git a/docs/reference/cli.md b/docs/reference/cli.md index ff972443..aa3dc4d6 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -158,7 +158,7 @@ Common options for the nightly actions include: |---|---| | `--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\|pi\|opencode\|auto` | Transcript source; `auto` keeps Codex-then-Claude precedence and does not select Copilot, Cursor, Pi, or OpenCode | +| `--source claude\|codex\|copilot\|cursor\|pi\|opencode\|dsh\|auto` | Transcript source; `auto` keeps Codex-then-Claude precedence and does not select Copilot, Cursor, Pi, OpenCode, or DSH | | `--backend mock\|claude\|codex\|copilot\|cursor\|pi\|opencode\|handoff\|azure_openai` | Replay/optimizer backend | | `--model NAME` | Backend-specific model override | | `--cursor-home PATH` | Override `~/.cursor` for Cursor transcript harvesting | @@ -168,6 +168,7 @@ Common options for the nightly actions include: | `--pi-path PATH` | Path to the installed Pi coding-agent CLI | | `--opencode-path PATH` | Path to the installed OpenCode CLI | | `--opencode-db PATH` | Path to the OpenCode SQLite history database | +| `--dsh-session-root PATH` | Override the DSH JSONL session root for `--source dsh` (default: `$DSH_HOME/sessions`, or `~/.dsh/sessions`) | | `--opencode-tool-replay` | Enable OpenCode tool-aware replay for `tool_called` checks in rule judges | | `--preferences TEXT` | House rules supplied to reflection | | `--lookback-hours N` | Initial transcript lookback; `0` scans all history | diff --git a/docs/sleep/README.md b/docs/sleep/README.md index f47556ee..35942554 100644 --- a/docs/sleep/README.md +++ b/docs/sleep/README.md @@ -190,6 +190,48 @@ The managed scheduler records the backend but does not preserve `--source`, `~/.skillopt-sleep/config.json`. Use an absolute `pi_path` and verify the scheduled account's Pi authentication. +### DeepSeek Harness (DSH) + +Use `--source dsh` to read local DSH JSONL sessions. By default, SkillOpt uses +DSH's standard `$DSH_HOME/sessions` directory, or `~/.dsh/sessions` when +`DSH_HOME` is unset. Install the optional Zstandard reader first: + +```bash +python -m pip install -e ".[dsh]" +skillopt-sleep harvest --project "$(pwd)" --source dsh --progress +``` + +`--dsh-session-root PATH` remains available only to override that default. + +The source reads `session.jsonl` and the default compressed +`session.jsonl.zstd` files below DSH's project/session directory layout. It +does not start DSH, require DSH login, connect to a model provider, or modify +the stored logs. `--source auto` retains Codex-then-Claude precedence and does +not select DSH. + +DSH harvesting keeps human user text, visible assistant text, short tool names, +timestamps, and positive/negative feedback signals derived from the immutable +`feedback/record` event. It excludes reasoning, tool arguments and results, +request/provider metadata, attachments, feedback remarks themselves, and +injected user-role context. Malformed sessions are silently skipped as a whole; +other sessions continue. Ordinary fork sessions are retained, while sessions +explicitly marked as subagents and SkillOpt replay sessions are excluded. + +The source follows the current observed DSH session format rather than a fixed +application-version matrix. A session whose format or event structure cannot be +safely understood is skipped. Plugin integration and DSH execution are outside +this source's scope. + +The managed scheduler does not preserve `--source` or `--dsh-session-root`. +Before scheduling, set the source in `~/.skillopt-sleep/config.json`; add +`dsh_session_root` only when overriding DSH's normal session directory: + +```json +{ + "transcript_source": "dsh" +} +``` + ### OpenCode Use `--source opencode` to read local OpenCode SQLite history without launching diff --git a/pyproject.toml b/pyproject.toml index 5d50b8fe..a93a515c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -55,6 +55,8 @@ docs = ["mkdocs-material>=9.5.0", "mkdocstrings[python]>=0.24.0"] webui = ["gradio>=5.50.0,<7"] # Development tools dev = ["ruff>=0.4.0", "pytest>=8.0.0"] +# DeepSeek Harness JSONL transcript source +dsh = ["zstandard>=0.22.0"] # All optional dependencies (except docs/dev/webui) all = [ "alfworld>=0.4.0", diff --git a/skillopt_sleep/__main__.py b/skillopt_sleep/__main__.py index e3b7794c..49b2ed36 100644 --- a/skillopt_sleep/__main__.py +++ b/skillopt_sleep/__main__.py @@ -15,10 +15,11 @@ --target-skill-path PATH explicit live SKILL.md to stage/adopt --tasks-file PATH reviewed TaskRecord JSON file to replay instead of harvesting --backend mock|claude|codex|copilot|cursor|pi|opencode|handoff|azure_openai - --source claude|codex|copilot|copilot_cli|cursor|pi|opencode|auto + --source claude|codex|copilot|copilot_cli|cursor|pi|opencode|dsh|auto --vscode-workspace-storage PATH --copilot-cli-session-store PATH --opencode-db PATH + --dsh-session-root PATH --model NAME --lookback-hours N --auto-adopt @@ -115,7 +116,7 @@ def _add_common(p: argparse.ArgumentParser) -> None: p.add_argument("--cursor-home", default="", help="override ~/.cursor for Cursor session harvest") p.add_argument("--pi-home", default="", help="override ~/.pi for Pi session harvest") p.add_argument("--source", default="", - choices=["", "claude", "codex", "copilot", "copilot_cli", "cursor", "pi", "opencode", "auto"], + choices=["", "claude", "codex", "copilot", "copilot_cli", "cursor", "pi", "opencode", "dsh", "auto"], help="session transcript source") p.add_argument("--vscode-workspace-storage", default="", help="override VS Code User/workspaceStorage root for copilot source") @@ -123,6 +124,8 @@ def _add_common(p: argparse.ArgumentParser) -> None: help="override ~/.copilot/session-store.db for copilot_cli source") p.add_argument("--opencode-db", default="", help="override the local OpenCode transcript database") + p.add_argument("--dsh-session-root", default="", + help="override DSH session root (default: $DSH_HOME/sessions or ~/.dsh/sessions)") 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) @@ -194,6 +197,8 @@ def _cfg_from_args(args, task_meta: Dict[str, Any] | None = None) -> Any: if args.opencode_db == ":memory:" else os.path.abspath(os.path.expanduser(args.opencode_db)) ) + if getattr(args, "dsh_session_root", ""): + overrides["dsh_session_root"] = os.path.abspath(os.path.expanduser(args.dsh_session_root)) lh = getattr(args, "lookback_hours", None) if lh is not None: # --lookback-hours was explicitly passed (0 = full history) overrides["lookback_hours"] = lh diff --git a/skillopt_sleep/config.py b/skillopt_sleep/config.py index d4a008d1..2a6361fb 100644 --- a/skillopt_sleep/config.py +++ b/skillopt_sleep/config.py @@ -32,7 +32,8 @@ "vscode_workspace_storage": "", # "" => auto-detect platform defaults "copilot_cli_session_store": "", # "" => ~/.copilot/session-store.db "opencode_db": "", # "" => OPENCODE_DB or the OpenCode XDG data path - # Explicit sources also include copilot, copilot_cli, cursor, pi, and opencode. + "dsh_session_root": "", # "" => $DSH_HOME/sessions, or ~/.dsh/sessions + # Explicit sources also include copilot, copilot_cli, cursor, pi, opencode, and dsh. # ``auto`` keeps the established Codex-then-Claude precedence. "transcript_source": "claude", "projects": "invoked", # "invoked" | "all" | [list of abs paths] @@ -164,6 +165,18 @@ def opencode_db_path(self) -> str: return ":memory:" return os.path.abspath(os.path.expanduser(str(value))) + @property + def dsh_session_root(self) -> str: + value = self.data.get("dsh_session_root", "") or "" + if value: + return os.path.abspath(os.path.expanduser(str(value))) + # Match the DSH base bundle: its JSONL persistence root is + # dshHomePath("sessions"), where DSH_HOME defaults to ~/.dsh. + dsh_home = str(os.environ.get("DSH_HOME", "")) + if not dsh_home.strip(): + dsh_home = os.path.join(os.path.expanduser("~"), ".dsh") + return os.path.abspath(os.path.expanduser(os.path.join(dsh_home, "sessions"))) + @property def vscode_workspace_storage(self) -> str: value = self.data.get("vscode_workspace_storage", "") or "" diff --git a/skillopt_sleep/harvest_dsh.py b/skillopt_sleep/harvest_dsh.py new file mode 100644 index 00000000..50e9584c --- /dev/null +++ b/skillopt_sleep/harvest_dsh.py @@ -0,0 +1,499 @@ +"""Read DeepSeek Harness JSONL session logs into ``SessionDigest`` records. + +The DSH JSONL persistence backend stores one append-only event log per session. +This reader is deliberately read-only and privacy-bounded: it keeps human user +text, visible assistant text, short tool names, timestamps, and derived +positive/negative feedback signals. It never persists reasoning, tool +arguments/results, request metadata, or feedback remarks themselves. + +Malformed sessions are silently discarded as a whole. DSH event sequences are +integrity-sensitive, so salvaging a suffix after a bad record could produce a +misleading conversation. A bad file must not prevent other sessions from +being harvested. +""" +from __future__ import annotations + +import io +import json +import os +import re +from datetime import datetime, timezone +from typing import Any, Iterable, Iterator, Optional + +from skillopt_sleep.harvest import _detect_feedback, _is_meta_prompt, _project_matches +from skillopt_sleep.staging import redact_secrets +from skillopt_sleep.types import SessionDigest + +_LOG_NAMES = {"session.jsonl", "session.jsonl.zstd"} +_PACKED_TYPES = {"text-chunks", "reasoning-chunks", "tool-call-chunks"} +_KNOWN_EVENT_TYPES = { + # Current DSH lifecycle and presentation metadata. These records advance + # the durable sequence but never contribute transcript content. + "permission/preset", + "sandbox/mode", + "approval/policy", + "approval/asked", + "approval/decided", + "agent/inbox/spliced", + "turn/start", + "turn/end", + "step/start", + "step/end", + "session/title", + "session/title-llm-request", + "user/message", + "assistant/chunk", + "assistant/message", + "tool/call", + "tool/result", + "request/header", + "request/context", + "session/end-seed", + "feedback/record", + "web/deepseek-search-llm-request", +} +_TOOL_NAME_RE = re.compile(r"[^A-Za-z0-9_.:-]+") + +# There is no DSH replay producer in this change. This stable, namespaced +# marker is reserved for a future producer so its sessions never feed the next +# harvest cycle. Do not infer replay from ordinary natural-language prompts. +DSH_REPLAY_SENTINEL = "" + + +class _DshFormatError(ValueError): + """Internal sentinel used to discard one invalid session silently.""" + + +def _is_safe_int(value: Any) -> bool: + return type(value) is int and 0 <= value <= 9_007_199_254_740_991 + + +def _sanitize_text(value: Any) -> str: + if not isinstance(value, str): + return "" + try: + text = str(redact_secrets(value)).replace("\x00", "").strip() + except Exception: + return "" + return "" if not text else text + + +def _sanitize_tool_name(value: Any) -> str: + if not isinstance(value, str) or not value: + return "" + return _TOOL_NAME_RE.sub("_", value)[:80] + + +def _dedup(values: Iterable[str]) -> list[str]: + return list(dict.fromkeys(value for value in values if value)) + + +def _iso_timestamp(value: Any) -> str: + """Turn a DSH epoch-millisecond timestamp into a stable ISO string.""" + if not _is_safe_int(value): + return "" + try: + return ( + datetime.fromtimestamp(value / 1000.0, tz=timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + except (OverflowError, OSError, ValueError): + return "" + + +def _utf16_units(value: str) -> Iterator[int]: + raw = value.encode("utf-16-le", "surrogatepass") + for offset in range(0, len(raw), 2): + yield int.from_bytes(raw[offset : offset + 2], "little") + + +def _encode_segment(value: str) -> str: + """Mirror DSH's injective safe-path encoding for ordinary Python strings.""" + if not value: + raise _DshFormatError("empty segment") + if value == ".": + return "~002E" + if value == "..": + return "~002E~002E" + pieces: list[str] = [] + for code in _utf16_units(value): + char = chr(code) + if char != "~" and (char.isascii() and (char.isalnum() or char in "._-")): + pieces.append(char) + else: + pieces.append(f"~{code:04X}") + return "".join(pieces) + + +def _project_key(cwd: str) -> str: + """Mirror DSH's readable, intentionally lossy project directory key.""" + if not cwd: + raise _DshFormatError("empty cwd") + pieces: list[str] = [] + separator_run = False + for code in _utf16_units(cwd): + char = chr(code) + if char in "/\\:": + if not separator_run: + pieces.append("-") + separator_run = True + elif char != "~" and (char.isascii() and (char.isalnum() or char in "._-")): + pieces.append(char) + separator_run = False + else: + pieces.append(f"~{code:04X}") + separator_run = False + slug = "".join(pieces).lstrip("-") or "root" + return f"--{slug[:251]}--" + + +def _is_within(root: str, candidate: str) -> bool: + try: + return os.path.commonpath([root, candidate]) == root + except ValueError: + return False + + +def _is_candidate_path(root: str, path: str) -> bool: + """Accept only DSH's fixed root/project/session/log layout.""" + if os.path.basename(path) not in _LOG_NAMES: + return False + real_path = os.path.realpath(path) + if not _is_within(root, real_path): + return False + try: + parts = os.path.relpath(real_path, root).split(os.sep) + except ValueError: + return False + return len(parts) == 3 and parts[-1] in _LOG_NAMES + + +def _iter_plain_lines(path: str) -> Iterator[str]: + try: + with open(path, "r", encoding="utf-8", newline="") as handle: + yield from handle + except (OSError, UnicodeError) as exc: + raise _DshFormatError("unreadable raw log") from exc + + +def _iter_zstd_lines(path: str) -> Iterator[str]: + try: + import zstandard as zstd + except ImportError as exc: + raise _DshFormatError("zstandard unavailable") from exc + try: + with open(path, "rb") as source: + decoder = zstd.ZstdDecompressor() + with decoder.stream_reader(source, read_across_frames=True) as reader: + with io.TextIOWrapper(reader, encoding="utf-8", newline="") as text: + yield from text + except (OSError, UnicodeError, zstd.ZstdError, ValueError) as exc: + raise _DshFormatError("unreadable zstd log") from exc + + +def _iter_records(path: str) -> Iterator[dict[str, Any]]: + lines = _iter_zstd_lines(path) if path.endswith(".zstd") else _iter_plain_lines(path) + saw_record = False + for line in lines: + if not line.strip(): + continue + # A DSH writer terminates every committed JSONL record. Do not use a + # possibly torn final line as a session event. + if not line.endswith(("\n", "\r")): + raise _DshFormatError("unterminated record") + try: + record = json.loads(line) + except (TypeError, ValueError) as exc: + raise _DshFormatError("invalid JSON record") from exc + if not isinstance(record, dict): + raise _DshFormatError("non-object record") + saw_record = True + yield record + if not saw_record: + raise _DshFormatError("empty session") + + +def _header_from_record(record: dict[str, Any], path: str, root: str) -> dict[str, Any]: + if record.get("type") != "session": + raise _DshFormatError("missing header") + version = record.get("version") + if version != 0: + raise _DshFormatError("unsupported format") + session_id = record.get("id") + created = record.get("createdAt") + depth = record.get("delegationDepth") + if not isinstance(session_id, str) or not session_id or not _is_safe_int(created) or not _is_safe_int(depth): + raise _DshFormatError("invalid header") + cwd = record.get("cwd") + if cwd is not None and (not isinstance(cwd, str) or not cwd): + raise _DshFormatError("invalid cwd") + parent = record.get("parentSession") + if parent is not None and (not isinstance(parent, str) or not parent): + raise _DshFormatError("invalid parent session") + if record.get("origin") not in {None, "subagent"}: + raise _DshFormatError("invalid origin") + if record.get("agentPreset") is not None and not isinstance(record.get("agentPreset"), str): + raise _DshFormatError("invalid agent preset") + seed_length = record.get("seedLength") + if seed_length is not None and not _is_safe_int(seed_length): + raise _DshFormatError("invalid seed length") + if "sandboxMode" in record or "approvalPolicy" in record: + raise _DshFormatError("retired header field") + + session_dir = os.path.dirname(path) + project_dir = os.path.dirname(session_dir) + expected_project = "_no-cwd" if cwd is None else _project_key(cwd) + if os.path.normcase(os.path.basename(session_dir)) != os.path.normcase(_encode_segment(session_id)): + raise _DshFormatError("session path mismatch") + if os.path.normcase(os.path.basename(project_dir)) != os.path.normcase(expected_project): + raise _DshFormatError("project path mismatch") + if not _is_within(root, os.path.realpath(path)): + raise _DshFormatError("path outside root") + return record + + +def _packed_count(record: dict[str, Any]) -> int: + row_type = record.get("type") + if row_type not in _PACKED_TYPES or not _is_safe_int(record.get("seq0")) or not _is_safe_int(record.get("time0")): + raise _DshFormatError("invalid packed row") + data = record.get("data") + if not isinstance(data, dict): + raise _DshFormatError("invalid packed data") + for key in ("turn", "step", "index"): + if not _is_safe_int(data.get(key)): + raise _DshFormatError("invalid packed position") + values = data.get("texts") if row_type in {"text-chunks", "reasoning-chunks"} else data.get("args") + if not isinstance(values, list) or len(values) < 3 or any(not isinstance(value, str) for value in values): + raise _DshFormatError("invalid packed members") + if row_type == "tool-call-chunks": + if not isinstance(data.get("callId"), str) or not data.get("callId"): + raise _DshFormatError("invalid packed tool call") + if data.get("name") is not None and not isinstance(data.get("name"), str): + raise _DshFormatError("invalid packed tool name") + deltas = data.get("dt") + if not isinstance(deltas, list) or any(not _is_safe_int(delta) for delta in deltas): + raise _DshFormatError("invalid packed timing") + # Current DSH writes a leading zero delta for the first member. Accept the + # equivalent n-1 representation too, because both reconstruct the same + # event stream and older logs may omit that redundant first zero. + if len(deltas) == len(values): + if deltas[0] != 0: + raise _DshFormatError("invalid packed first delta") + elif len(deltas) != len(values) - 1: + raise _DshFormatError("invalid packed delta count") + return len(values) + + +def _validate_event(record: dict[str, Any], expected_seq: int) -> None: + event_type = record.get("type") + if not isinstance(event_type, str) or event_type not in _KNOWN_EVENT_TYPES: + if record.get("ignorable") is True: + return + raise _DshFormatError("unknown event") + if record.get("seq") != expected_seq or not _is_safe_int(record.get("seq")): + raise _DshFormatError("non-contiguous sequence") + if not _is_safe_int(record.get("time")) or not isinstance(record.get("data"), dict): + raise _DshFormatError("invalid event envelope") + + +def _is_append_surface(record: dict[str, Any]) -> bool: + operation = record.get("surfaceOp") + return operation is None or operation == "append" + + +def _text_blocks(content: Any) -> list[str]: + if isinstance(content, str): + return [_sanitize_text(content)] + if not isinstance(content, list): + return [] + return [ + _sanitize_text(block.get("text")) + for block in content + if isinstance(block, dict) and block.get("type") == "text" + ] + + +def _tool_names_from_content(content: Any) -> list[str]: + if not isinstance(content, list): + return [] + return [ + _sanitize_tool_name(block.get("name")) + for block in content + if isinstance(block, dict) and block.get("type") == "tool-call" + ] + + +def _human_user_text(record: dict[str, Any]) -> str: + if not _is_append_surface(record): + return "" + data = record["data"] + if data.get("role") != "user": + raise _DshFormatError("invalid user message") + source = data.get("source") + if not isinstance(source, dict) or source.get("kind") != "user": + return "" + text = "\n".join(part for part in _text_blocks(data.get("content")) if part).strip() + return "" if _is_meta_prompt(text) else text + + +def _assistant_message(record: dict[str, Any]) -> tuple[str, list[str]]: + if not _is_append_surface(record): + return "", [] + data = record["data"] + message = data.get("message") + if not isinstance(message, dict) or message.get("role") != "assistant": + raise _DshFormatError("invalid assistant message") + text = "\n".join(part for part in _text_blocks(message.get("content")) if part).strip() + return text, _tool_names_from_content(message.get("content")) + + +def _tool_call_name(record: dict[str, Any]) -> str: + data = record["data"] + name = data.get("name") + if not isinstance(name, str) or not name: + raise _DshFormatError("invalid tool call") + return _sanitize_tool_name(name) + + +def _feedback_signals(record: dict[str, Any]) -> list[str]: + text = _sanitize_text(record["data"].get("text")) + return _detect_feedback(text) if text else [] + + +def _is_dsh_replay(digest: SessionDigest) -> bool: + return bool(digest.user_prompts) and digest.user_prompts[0].lstrip().startswith(DSH_REPLAY_SENTINEL) + + +def digest_dsh_session(path: str, *, root: str) -> Optional[SessionDigest]: + """Parse one complete DSH session file, returning ``None`` on any failure.""" + try: + records = _iter_records(path) + header = _header_from_record(next(records), path, root) + if header.get("origin") == "subagent": + return None + + session_id = str(header["id"]) + project = str(header.get("cwd") or "") + started_at = _iso_timestamp(header["createdAt"]) + ended_at = started_at + user_prompts: list[str] = [] + assistant_finals: list[str] = [] + tools: list[str] = [] + feedback: list[str] = [] + expected_seq = 0 + n_user = 0 + n_assistant = 0 + + for record in records: + row_type = record.get("type") + if row_type in _PACKED_TYPES: + count = _packed_count(record) + if record["seq0"] != expected_seq: + raise _DshFormatError("packed sequence gap") + expected_seq += count + # Packed chunks are intentionally not retained, but their final + # timestamp is still the best session-end timestamp. + deltas = record["data"]["dt"] + ended_at = _iso_timestamp(record["time0"] + sum(deltas)) + continue + + _validate_event(record, expected_seq) + if record.get("type") in _KNOWN_EVENT_TYPES: + expected_seq += 1 + ended_at = _iso_timestamp(record["time"]) + # An ignorable extension has a normal event envelope and therefore + # still occupies one sequence number. + elif record.get("ignorable") is True: + if record.get("seq") != expected_seq or not _is_safe_int(record.get("time")): + raise _DshFormatError("invalid ignorable event") + expected_seq += 1 + ended_at = _iso_timestamp(record["time"]) + + event_type = record.get("type") + if event_type == "user/message": + text = _human_user_text(record) + if text: + user_prompts.append(text) + feedback.extend(_detect_feedback(text)) + n_user += 1 + elif event_type == "assistant/message": + text, names = _assistant_message(record) + tools.extend(names) + n_assistant += 1 + if text: + assistant_finals.append(text) + elif event_type == "tool/call": + tools.append(_tool_call_name(record)) + elif event_type == "feedback/record": + feedback.extend(_feedback_signals(record)) + + if not user_prompts and not assistant_finals: + return None + + digest = SessionDigest( + session_id=session_id, + project=project, + started_at=started_at, + ended_at=ended_at, + user_prompts=user_prompts, + assistant_finals=assistant_finals[-5:], + tools_used=_dedup(tools), + files_touched=[], + feedback_signals=_dedup(feedback), + n_user_turns=n_user, + n_assistant_turns=n_assistant, + raw_path=path, + ) + return None if _is_dsh_replay(digest) else digest + except (OSError, StopIteration, _DshFormatError, ValueError, TypeError, json.JSONDecodeError): + return None + + +def harvest_dsh( + session_root: str, + *, + scope: Any = "all", + invoked_project: str = "", + since_iso: Optional[str] = None, + limit: int = 0, +) -> list[SessionDigest]: + """Discover valid DSH session logs below one explicitly supplied root.""" + if not session_root: + return [] + root = os.path.realpath(os.path.abspath(os.path.expanduser(session_root))) + if not os.path.isdir(root): + return [] + + candidates: list[tuple[float, str]] = [] + for directory, _dirs, files in os.walk(root, followlinks=False): + for filename in files: + if filename not in _LOG_NAMES: + continue + path = os.path.join(directory, filename) + if not _is_candidate_path(root, path): + continue + try: + candidates.append((os.path.getmtime(path), path)) + except OSError: + continue + candidates.sort(key=lambda item: (-item[0], item[1])) + + digests: list[SessionDigest] = [] + seen_ids: set[str] = set() + for _mtime, path in candidates: + digest = digest_dsh_session(path, root=root) + if digest is None or digest.session_id in seen_ids: + continue + seen_ids.add(digest.session_id) + if not digest.project and scope != "all": + continue + if not _project_matches(digest.project, scope, invoked_project): + continue + if since_iso and digest.ended_at and digest.ended_at < since_iso: + continue + digests.append(digest) + if limit and len(digests) >= limit: + break + return digests diff --git a/skillopt_sleep/harvest_sources.py b/skillopt_sleep/harvest_sources.py index 12506e7b..9bded4cd 100644 --- a/skillopt_sleep/harvest_sources.py +++ b/skillopt_sleep/harvest_sources.py @@ -8,6 +8,7 @@ from skillopt_sleep.harvest_copilot import harvest_copilot from skillopt_sleep.harvest_copilot_cli import harvest_copilot_cli from skillopt_sleep.harvest_cursor import harvest_cursor +from skillopt_sleep.harvest_dsh import harvest_dsh from skillopt_sleep.harvest_opencode import harvest_opencode from skillopt_sleep.harvest_pi import harvest_pi from skillopt_sleep.types import SessionDigest @@ -66,6 +67,14 @@ def harvest_for_config(cfg, *, since_iso: Optional[str] = None, limit: int = 0) since_iso=since_iso, limit=limit, ) + if source == "dsh": + return harvest_dsh( + cfg.dsh_session_root, + scope=scope, + invoked_project=invoked_project, + since_iso=since_iso, + limit=limit, + ) if source == "auto": codex_digests = harvest_codex( cfg.codex_archived_sessions_dir, diff --git a/tests/test_harvest_dsh.py b/tests/test_harvest_dsh.py new file mode 100644 index 00000000..96391574 --- /dev/null +++ b/tests/test_harvest_dsh.py @@ -0,0 +1,371 @@ +"""Coverage for the read-only DeepSeek Harness transcript harvester.""" +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from unittest import mock + +import pytest + +from skillopt_sleep.__main__ import _add_common, _cfg_from_args +from skillopt_sleep.config import load_config +from skillopt_sleep.harvest_dsh import ( + DSH_REPLAY_SENTINEL, + _encode_segment, + _project_key, + digest_dsh_session, + harvest_dsh, +) +from skillopt_sleep.harvest_sources import harvest_for_config +from skillopt_sleep.types import SessionDigest + +_BASE_TIME = 1_800_000_000_000 + + +def _header(session_id: str, cwd: str | None, **extra): + value = { + "type": "session", + "version": 0, + "id": session_id, + "createdAt": _BASE_TIME, + "delegationDepth": 0, + } + if cwd is not None: + value["cwd"] = cwd + value.update(extra) + return value + + +def _user(seq: int, text: str, *, source="user", append=True): + event = { + "type": "user/message", + "seq": seq, + "time": _BASE_TIME + 1000 * (seq + 1), + "data": { + "id": f"user-{seq}", + "role": "user", + "content": [{"type": "text", "text": text}], + "source": {"kind": source}, + }, + } + if append: + event["surfaceOp"] = "append" + return event + + +def _assistant(seq: int, text: str, *, tool_name="", replace=False): + content = [ + {"type": "reasoning", "text": "private chain of thought"}, + {"type": "text", "text": text}, + ] + if tool_name: + content.append({"type": "tool-call", "id": f"call-{seq}", "name": tool_name, "arguments": '{"secret":true}'}) + event = { + "type": "assistant/message", + "seq": seq, + "time": _BASE_TIME + 1000 * (seq + 1), + "data": { + "turn": 1, + "step": 1, + "message": { + "id": f"assistant-{seq}", + "role": "assistant", + "content": content, + "source": {"kind": "model", "provider": "test", "model": "test-model"}, + }, + }, + "surfaceOp": {"op": "replace", "start": 0, "end": 0} if replace else "append", + } + return event + + +def _tool_call(seq: int, name: str): + return { + "type": "tool/call", + "seq": seq, + "time": _BASE_TIME + 1000 * (seq + 1), + "data": {"turn": 1, "step": 1, "callId": f"call-{seq}", "name": name, "arguments": '{"api_key":"secret"}'}, + } + + +def _feedback(seq: int, text: str): + return { + "type": "feedback/record", + "seq": seq, + "time": _BASE_TIME + 1000 * (seq + 1), + "data": {"text": text}, + } + + +def _metadata(seq: int, event_type: str): + return { + "type": event_type, + "seq": seq, + "time": _BASE_TIME + 1000 * (seq + 1), + "data": {}, + } + + +def _write_raw(root: Path, session_id: str, cwd: str | None, records: list[dict], **header_extra) -> Path: + project_dir = "_no-cwd" if cwd is None else _project_key(cwd) + path = root / project_dir / _encode_segment(session_id) / "session.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + rows = [_header(session_id, cwd, **header_extra), *records] + path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") + return path + + +def test_digest_extracts_only_safe_dsh_fields(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + path = _write_raw( + tmp_path, + "session-1", + project, + [ + _user(0, "Fix the tests. Authorization: Bearer sk-1234567890abcdefghij"), + _assistant(1, "I fixed it.", tool_name="shell/run "), + _tool_call(2, "shell/run "), + _feedback(3, "Perfect, that works now. token=super-secret"), + ], + ) + + digest = digest_dsh_session(str(path), root=str(tmp_path)) + + assert digest is not None + assert digest.project == project + assert digest.user_prompts and "sk-1234567890abcdefghij" not in digest.user_prompts[0] + assert digest.assistant_finals == ["I fixed it."] + assert digest.tools_used == ["shell_run_unsafe_"] + assert digest.n_user_turns == 1 + assert digest.n_assistant_turns == 1 + assert any(signal.startswith("pos:") for signal in digest.feedback_signals) + persisted = json.dumps(digest.to_dict()) + assert "private chain of thought" not in persisted + assert '"api_key"' not in persisted + assert "super-secret" not in persisted + + +def test_packed_rows_are_validated_and_do_not_leak_chunks(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + packed = { + "type": "text-chunks", + "seq0": 1, + "time0": _BASE_TIME + 2000, + "data": { + "turn": 1, + "step": 1, + "index": 0, + "dt": [0, 7, 9], + "texts": ["private", " streamed", " output"], + }, + } + path = _write_raw(tmp_path, "packed", project, [_user(0, "request"), packed, _assistant(4, "final")]) + + digest = digest_dsh_session(str(path), root=str(tmp_path)) + + assert digest is not None + assert digest.user_prompts == ["request"] + assert digest.assistant_finals == ["final"] + assert "private streamed output" not in json.dumps(digest.to_dict()) + + +def test_malformed_packed_row_rejects_the_whole_session(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + path = _write_raw( + tmp_path, + "bad-packed", + project, + [ + _user(0, "request"), + { + "type": "text-chunks", + "seq0": 1, + "time0": _BASE_TIME + 2000, + "data": {"turn": 1, "step": 1, "index": 0, "dt": [0], "texts": ["a", "b", "c"]}, + }, + ], + ) + + assert digest_dsh_session(str(path), root=str(tmp_path)) is None + + +def test_fork_is_retained_but_subagent_and_replay_are_excluded(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + _write_raw( + tmp_path, + "fork", + project, + [_user(0, "inherited request"), _assistant(1, "superseded", replace=True), _assistant(2, "active final")], + parentSession="parent", + seedLength=1, + ) + _write_raw( + tmp_path, + "subagent", + project, + [_user(0, "machine task"), _assistant(1, "machine final")], + origin="subagent", + delegationDepth=1, + ) + _write_raw( + tmp_path, + "replay", + project, + [_user(0, DSH_REPLAY_SENTINEL + "\nrun internal task"), _assistant(1, "internal")], + ) + + digests = harvest_dsh(str(tmp_path), scope="all") + + assert [digest.session_id for digest in digests] == ["fork"] + assert digests[0].assistant_finals == ["active final"] + + +def test_bad_session_is_silent_and_does_not_block_other_sessions(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + _write_raw(tmp_path, "good", project, [_user(0, "good request"), _assistant(1, "good final")]) + bad = tmp_path / _project_key(project) / _encode_segment("bad") / "session.jsonl" + bad.parent.mkdir(parents=True) + bad.write_text('{"type":"session"}\nnot-json\n', encoding="utf-8") + + digests = harvest_dsh(str(tmp_path), scope="all") + + assert [digest.session_id for digest in digests] == ["good"] + + +def test_unknown_required_event_rejects_but_ignorable_event_is_skipped(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + _write_raw( + tmp_path, + "ignorable", + project, + [ + _user(0, "request"), + {"type": "plugin/info", "seq": 1, "time": _BASE_TIME + 2000, "data": {}, "ignorable": True}, + _assistant(2, "final"), + ], + ) + _write_raw( + tmp_path, + "required", + project, + [ + _user(0, "request"), + {"type": "plugin/required", "seq": 1, "time": _BASE_TIME + 2000, "data": {}}, + _assistant(2, "final"), + ], + ) + + assert [digest.session_id for digest in harvest_dsh(str(tmp_path), scope="all")] == ["ignorable"] + + +def test_current_dsh_lifecycle_metadata_is_accepted_without_retention(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + metadata_types = [ + "permission/preset", + "sandbox/mode", + "approval/policy", + "agent/inbox/spliced", + "session/title", + "session/title-llm-request", + "web/deepseek-search-llm-request", + "approval/asked", + "approval/decided", + ] + records = [_metadata(index, event_type) for index, event_type in enumerate(metadata_types)] + records.extend([_user(len(records), "actual user request"), _assistant(len(records) + 1, "actual final")]) + + path = _write_raw(tmp_path, "metadata", project, records) + digest = digest_dsh_session(str(path), root=str(tmp_path)) + + assert digest is not None + assert digest.user_prompts == ["actual user request"] + assert digest.assistant_finals == ["actual final"] + + +def test_scope_since_limit_and_identity_checks(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + other = str((tmp_path / "other").resolve()) + _write_raw(tmp_path, "one", project, [_user(0, "one"), _assistant(1, "one")]) + _write_raw(tmp_path, "two", other, [_user(0, "two"), _assistant(1, "two")]) + wrong = tmp_path / _project_key(project) / "not-the-id" / "session.jsonl" + wrong.parent.mkdir(parents=True) + wrong.write_text(json.dumps(_header("wrong", project)) + "\n", encoding="utf-8") + + invoked = harvest_dsh(str(tmp_path), scope="invoked", invoked_project=project, limit=1) + assert [digest.session_id for digest in invoked] == ["one"] + assert harvest_dsh(str(tmp_path), scope="all", since_iso="2030-01-01T00:00:00Z") == [] + + +def test_zstd_concatenated_frames_are_read(tmp_path: Path): + zstd = pytest.importorskip("zstandard") + project = str((tmp_path / "repo").resolve()) + path = tmp_path / _project_key(project) / _encode_segment("compressed") / "session.jsonl.zstd" + path.parent.mkdir(parents=True) + compressor = zstd.ZstdCompressor(write_checksum=True) + header = json.dumps(_header("compressed", project)).encode() + b"\n" + events = b"".join( + json.dumps(row).encode() + b"\n" + for row in [_user(0, "compressed request"), _assistant(1, "compressed final")] + ) + path.write_bytes(compressor.compress(header) + compressor.compress(events)) + + digest = digest_dsh_session(str(path), root=str(tmp_path)) + + assert digest is not None + assert digest.user_prompts == ["compressed request"] + assert digest.assistant_finals == ["compressed final"] + + +def test_cli_config_and_source_dispatch_for_dsh(monkeypatch, tmp_path: Path): + parser = argparse.ArgumentParser() + _add_common(parser) + args = parser.parse_args(["--source", "dsh", "--dsh-session-root", "~/dsh-sessions"]) + monkeypatch.setattr("skillopt_sleep.config._user_config_path", lambda: None) + + cfg = _cfg_from_args(args) + expected_root = str(Path("~/dsh-sessions").expanduser().resolve()) + assert cfg.get("transcript_source") == "dsh" + assert cfg.dsh_session_root == expected_root + + project = str((tmp_path / "repo").resolve()) + configured = load_config(transcript_source="dsh", dsh_session_root=str(tmp_path), invoked_project=project) + expected = [SessionDigest(session_id="dsh", project=project)] + with ( + mock.patch("skillopt_sleep.harvest_sources.harvest_dsh", return_value=expected) as dsh, + mock.patch("skillopt_sleep.harvest_sources.harvest") as claude, + mock.patch("skillopt_sleep.harvest_sources.harvest_codex") as codex, + ): + assert harvest_for_config(configured, since_iso="2026-01-01T00:00:00Z", limit=2) == expected + dsh.assert_called_once_with( + configured.dsh_session_root, + scope="invoked", + invoked_project=project, + since_iso="2026-01-01T00:00:00Z", + limit=2, + ) + claude.assert_not_called() + codex.assert_not_called() + + +def test_dsh_uses_the_standard_home_sessions_directory(monkeypatch, tmp_path: Path): + monkeypatch.setattr("skillopt_sleep.config._user_config_path", lambda: None) + dsh_home = tmp_path / "dsh-home" + monkeypatch.setenv("DSH_HOME", str(dsh_home)) + + cfg = load_config(transcript_source="dsh") + + assert cfg.dsh_session_root == str((dsh_home / "sessions").resolve()) + + +def test_auto_source_does_not_add_dsh_precedence(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + cfg = load_config(transcript_source="auto", invoked_project=project, dsh_session_root=str(tmp_path)) + expected = [SessionDigest(session_id="claude", project=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_dsh") as dsh, + ): + assert harvest_for_config(cfg) == expected + dsh.assert_not_called() From 22615819b33ad2e73441b728913107715df94d28 Mon Sep 17 00:00:00 2001 From: TaoRunguo Date: Tue, 8 Sep 2026 14:28:21 +0800 Subject: [PATCH 2/2] fix(sleep): support current DSH session format --- docs/sleep/README.md | 17 +- skillopt_sleep/harvest_dsh.py | 464 ++++++++++++++---- skillopt_sleep/harvest_sources.py | 1 + .../master-v2-fixture/session.v2.jsonl.zstd | Bin 0 -> 2032 bytes tests/fixtures/dsh/README.md | 5 + tests/test_harvest_dsh.py | 442 ++++++++++++++++- 6 files changed, 816 insertions(+), 113 deletions(-) create mode 100644 tests/fixtures/dsh/--fixture-project--/master-v2-fixture/session.v2.jsonl.zstd create mode 100644 tests/fixtures/dsh/README.md diff --git a/docs/sleep/README.md b/docs/sleep/README.md index 35942554..adea837d 100644 --- a/docs/sleep/README.md +++ b/docs/sleep/README.md @@ -203,11 +203,18 @@ skillopt-sleep harvest --project "$(pwd)" --source dsh --progress `--dsh-session-root PATH` remains available only to override that default. -The source reads `session.jsonl` and the default compressed -`session.jsonl.zstd` files below DSH's project/session directory layout. It -does not start DSH, require DSH login, connect to a model provider, or modify -the stored logs. `--source auto` retains Codex-then-Claude precedence and does -not select DSH. +The source discovers canonical DSH generations below the project/session +directory layout: `session.jsonl` for v0 and `session.vN.jsonl` for numbered +generations, with the corresponding `.zstd` forms. It selects the numerically +highest generation for each session. The current v2 generation and retained +historical v0/v1 generations are supported; an unsupported highest generation +is skipped instead of falling back to an older file. It does not start DSH, +require DSH login, connect to a model provider, or modify the stored logs. +Normal harvesting remains silent for skipped sessions; use `--progress` or +enable debug logging to see the selected file, highest generation, and skip +reason. V2 headers, inherited-session markers, and replacement provenance are +validated before visible text is exported. +`--source auto` retains Codex-then-Claude precedence and does not select DSH. DSH harvesting keeps human user text, visible assistant text, short tool names, timestamps, and positive/negative feedback signals derived from the immutable diff --git a/skillopt_sleep/harvest_dsh.py b/skillopt_sleep/harvest_dsh.py index 50e9584c..3cd41e23 100644 --- a/skillopt_sleep/harvest_dsh.py +++ b/skillopt_sleep/harvest_dsh.py @@ -9,14 +9,23 @@ Malformed sessions are silently discarded as a whole. DSH event sequences are integrity-sensitive, so salvaging a suffix after a bad record could produce a misleading conversation. A bad file must not prevent other sessions from -being harvested. +being harvested. When DSH retains multiple immutable format generations in +one session directory, only the numerically highest generation is considered; +an unsupported highest generation is never replaced by an older predecessor. """ from __future__ import annotations import io import json +import logging +import ntpath import os +import posixpath import re +import sys +from collections import deque +from contextlib import closing +from dataclasses import dataclass from datetime import datetime, timezone from typing import Any, Iterable, Iterator, Optional @@ -24,8 +33,20 @@ from skillopt_sleep.staging import redact_secrets from skillopt_sleep.types import SessionDigest -_LOG_NAMES = {"session.jsonl", "session.jsonl.zstd"} +_LOGGER = logging.getLogger("skillopt_sleep.harvest_dsh") +_LOG_NAME_RE = re.compile(r"^session(?:\.v([1-9][0-9]*))?\.jsonl(?:\.zstd)?$") +# DSH's current writer is v2. v0/v1 are retained historical generations and +# use the same validated logical event boundary here. A future generation is +# selected first and then discarded if it is outside this set. +_SUPPORTED_FORMAT_VERSIONS = frozenset({0, 1, 2}) +_V2_HEADER_REQUIRED_KEYS = frozenset( + {"type", "version", "id", "createdAt", "isSeeded", "delegationDepth"} +) +_V2_HEADER_ALLOWED_KEYS = _V2_HEADER_REQUIRED_KEYS | frozenset( + {"cwd", "parentSession", "origin", "agentPreset"} +) _PACKED_TYPES = {"text-chunks", "reasoning-chunks", "tool-call-chunks"} +_SURFACE_EVENT_TYPES = {"user/message", "assistant/message", "tool/result"} _KNOWN_EVENT_TYPES = { # Current DSH lifecycle and presentation metadata. These records advance # the durable sequence but never contribute transcript content. @@ -44,6 +65,7 @@ "user/message", "assistant/chunk", "assistant/message", + "assistant/attempt", "tool/call", "tool/result", "request/header", @@ -64,6 +86,57 @@ class _DshFormatError(ValueError): """Internal sentinel used to discard one invalid session silently.""" +def _is_absolute_dsh_path(value: str) -> bool: + """Recognize the POSIX and Windows absolute paths DSH can persist.""" + return posixpath.isabs(value) or ntpath.isabs(value) + + +@dataclass(frozen=True, slots=True) +class _SurfaceEntry: + """Only digest content and the identity needed by later replacements.""" + + seq: int + kind: str + text: str = "" + tools: tuple[str, ...] = () + + +@dataclass(frozen=True, slots=True) +class _Replacement: + start: int + end: int + # Legacy replacements do not require provenance coverage. + sources: Optional[frozenset[int]] = None + + +@dataclass(frozen=True, slots=True) +class _DshEvent: + """Validated event projection; never holds a raw message or tool payload.""" + + next_seq: int + time: int + surface: Optional[_SurfaceEntry] = None + replacement: Optional[_Replacement] = None + tool: str = "" + feedback: tuple[str, ...] = () + inherited_seed_marker: bool = False + + +def _diagnostic(message: str, *args: object, progress: bool = False) -> None: + rendered = message % args if args else message + _LOGGER.debug("%s", rendered) + if progress: + print(f"[sleep] dsh: {rendered}", file=sys.stderr, flush=True) + + +def _parse_log_filename(filename: str) -> Optional[tuple[int, str]]: + """Return ``(generation, encoding)`` for a canonical DSH log filename.""" + match = _LOG_NAME_RE.fullmatch(filename) + if match is None: + return None + return (int(match.group(1) or 0), "zstd" if filename.endswith(".zstd") else "raw") + + def _is_safe_int(value: Any) -> bool: return type(value) is int and 0 <= value <= 9_007_199_254_740_991 @@ -95,7 +168,6 @@ def _iso_timestamp(value: Any) -> str: try: return ( datetime.fromtimestamp(value / 1000.0, tz=timezone.utc) - .replace(microsecond=0) .isoformat() .replace("+00:00", "Z") ) @@ -103,6 +175,17 @@ def _iso_timestamp(value: Any) -> str: return "" +def _iso_epoch(value: Optional[str]) -> Optional[float]: + """Compare instants, interpreting offset-free Sleep checkpoints locally.""" + if not value: + return None + try: + normalized = value[:-1] + "+00:00" if value[-1:] in {"Z", "z"} else value + return datetime.fromisoformat(normalized).timestamp() + except (OverflowError, OSError, TypeError, ValueError): + return None + + def _utf16_units(value: str) -> Iterator[int]: raw = value.encode("utf-16-le", "surrogatepass") for offset in range(0, len(raw), 2): @@ -158,7 +241,7 @@ def _is_within(root: str, candidate: str) -> bool: def _is_candidate_path(root: str, path: str) -> bool: """Accept only DSH's fixed root/project/session/log layout.""" - if os.path.basename(path) not in _LOG_NAMES: + if _parse_log_filename(os.path.basename(path)) is None: return False real_path = os.path.realpath(path) if not _is_within(root, real_path): @@ -167,7 +250,7 @@ def _is_candidate_path(root: str, path: str) -> bool: parts = os.path.relpath(real_path, root).split(os.sep) except ValueError: return False - return len(parts) == 3 and parts[-1] in _LOG_NAMES + return len(parts) == 3 and _parse_log_filename(parts[-1]) is not None def _iter_plain_lines(path: str) -> Iterator[str]: @@ -196,21 +279,22 @@ def _iter_zstd_lines(path: str) -> Iterator[str]: def _iter_records(path: str) -> Iterator[dict[str, Any]]: lines = _iter_zstd_lines(path) if path.endswith(".zstd") else _iter_plain_lines(path) saw_record = False - for line in lines: - if not line.strip(): - continue - # A DSH writer terminates every committed JSONL record. Do not use a - # possibly torn final line as a session event. - if not line.endswith(("\n", "\r")): - raise _DshFormatError("unterminated record") - try: - record = json.loads(line) - except (TypeError, ValueError) as exc: - raise _DshFormatError("invalid JSON record") from exc - if not isinstance(record, dict): - raise _DshFormatError("non-object record") - saw_record = True - yield record + with closing(lines): + for line in lines: + if not line.strip(): + continue + # A DSH writer terminates every committed JSONL record. Do not use + # a possibly torn final line as a session event. + if not line.endswith(("\n", "\r")): + raise _DshFormatError("unterminated record") + try: + record = json.loads(line) + except (TypeError, ValueError) as exc: + raise _DshFormatError("invalid JSON record") from exc + if not isinstance(record, dict): + raise _DshFormatError("non-object record") + saw_record = True + yield record if not saw_record: raise _DshFormatError("empty session") @@ -218,9 +302,17 @@ def _iter_records(path: str) -> Iterator[dict[str, Any]]: def _header_from_record(record: dict[str, Any], path: str, root: str) -> dict[str, Any]: if record.get("type") != "session": raise _DshFormatError("missing header") + parsed_name = _parse_log_filename(os.path.basename(path)) + if parsed_name is None: + raise _DshFormatError("non-canonical log name") + filename_version, _encoding = parsed_name version = record.get("version") - if version != 0: + if type(version) is not int or version != filename_version or version not in _SUPPORTED_FORMAT_VERSIONS: raise _DshFormatError("unsupported format") + if version >= 2 and ( + not _V2_HEADER_REQUIRED_KEYS.issubset(record) or not set(record).issubset(_V2_HEADER_ALLOWED_KEYS) + ): + raise _DshFormatError("invalid v2 header keys") session_id = record.get("id") created = record.get("createdAt") depth = record.get("delegationDepth") @@ -229,11 +321,15 @@ def _header_from_record(record: dict[str, Any], path: str, root: str) -> dict[st cwd = record.get("cwd") if cwd is not None and (not isinstance(cwd, str) or not cwd): raise _DshFormatError("invalid cwd") + if version >= 2 and cwd is not None and not _is_absolute_dsh_path(cwd): + raise _DshFormatError("invalid v2 cwd") parent = record.get("parentSession") if parent is not None and (not isinstance(parent, str) or not parent): raise _DshFormatError("invalid parent session") if record.get("origin") not in {None, "subagent"}: raise _DshFormatError("invalid origin") + if record.get("isSeeded") is not None and not isinstance(record.get("isSeeded"), bool): + raise _DshFormatError("invalid seeded flag") if record.get("agentPreset") is not None and not isinstance(record.get("agentPreset"), str): raise _DshFormatError("invalid agent preset") seed_length = record.get("seedLength") @@ -286,23 +382,6 @@ def _packed_count(record: dict[str, Any]) -> int: return len(values) -def _validate_event(record: dict[str, Any], expected_seq: int) -> None: - event_type = record.get("type") - if not isinstance(event_type, str) or event_type not in _KNOWN_EVENT_TYPES: - if record.get("ignorable") is True: - return - raise _DshFormatError("unknown event") - if record.get("seq") != expected_seq or not _is_safe_int(record.get("seq")): - raise _DshFormatError("non-contiguous sequence") - if not _is_safe_int(record.get("time")) or not isinstance(record.get("data"), dict): - raise _DshFormatError("invalid event envelope") - - -def _is_append_surface(record: dict[str, Any]) -> bool: - operation = record.get("surfaceOp") - return operation is None or operation == "append" - - def _text_blocks(content: Any) -> list[str]: if isinstance(content, str): return [_sanitize_text(content)] @@ -326,8 +405,6 @@ def _tool_names_from_content(content: Any) -> list[str]: def _human_user_text(record: dict[str, Any]) -> str: - if not _is_append_surface(record): - return "" data = record["data"] if data.get("role") != "user": raise _DshFormatError("invalid user message") @@ -339,8 +416,6 @@ def _human_user_text(record: dict[str, Any]) -> str: def _assistant_message(record: dict[str, Any]) -> tuple[str, list[str]]: - if not _is_append_surface(record): - return "", [] data = record["data"] message = data.get("message") if not isinstance(message, dict) or message.get("role") != "assistant": @@ -362,73 +437,207 @@ def _feedback_signals(record: dict[str, Any]) -> list[str]: return _detect_feedback(text) if text else [] +def _decode_source_event_seqs(value: Any, *, max_entries: int) -> list[int]: + """Expand DSH v2's compact ``number | [start, end]`` sequence ranges.""" + if not isinstance(value, list): + raise _DshFormatError("invalid surface sources") + decoded: list[int] = [] + has_range = False + for item in value: + if _is_safe_int(item): + if len(decoded) >= max_entries: + raise _DshFormatError("too many surface sources") + decoded.append(item) + continue + if ( + not isinstance(item, list) + or len(item) != 2 + or not _is_safe_int(item[0]) + or not _is_safe_int(item[1]) + or item[0] > item[1] + ): + raise _DshFormatError("invalid surface sources") + start, end = item + count = end - start + 1 + if count > max_entries - len(decoded): + raise _DshFormatError("too many surface sources") + decoded.extend(range(start, end + 1)) + has_range = True + if has_range and any(later <= earlier for earlier, later in zip(decoded, decoded[1:])): + raise _DshFormatError("non-increasing surface sources") + if len(set(decoded)) != len(decoded): + raise _DshFormatError("duplicate surface sources") + return decoded + + +def _surface_replacement(record: dict[str, Any], version: int) -> Optional[_Replacement]: + """Validate wire-format surface rules and normalize append/replace.""" + operation = record.get("surfaceOp") + if operation is None: + if version >= 2: + raise _DshFormatError("missing v2 surface operation") + operation = "append" + + # Preserve the currently supported v2 assistant contract explicitly. + # Broadening this to replacement requires an upstream format sample. + if version >= 2 and record["type"] == "assistant/message": + if operation != "append" or record.get("sourceEventSeqs") is not None: + raise _DshFormatError("v2 assistant surface must append without sources") + + source_seqs = record.get("sourceEventSeqs") + if source_seqs is not None: + source_seqs = ( + _decode_source_event_seqs(source_seqs, max_entries=record["seq"]) + if version >= 2 + else source_seqs + ) + if ( + not isinstance(source_seqs, list) + or any(not _is_safe_int(seq) for seq in source_seqs) + or len(set(source_seqs)) != len(source_seqs) + ): + raise _DshFormatError("invalid surface sources") + if operation == "append": + return None + if not isinstance(operation, dict) or operation.get("op") != "replace": + raise _DshFormatError("invalid surface operation") + start, end = operation.get("start"), operation.get("end") + if not _is_safe_int(start) or not _is_safe_int(end): + raise _DshFormatError("invalid surface replacement range") + if version >= 2 and not source_seqs: + raise _DshFormatError("missing v2 surface sources") + return _Replacement(start, end, frozenset(source_seqs) if version >= 2 else None) + + +def _decode_event(record: dict[str, Any], expected_seq: int, version: int) -> _DshEvent: + """Validate sequencing and harvest-relevant fields, then project safe data.""" + kind = record.get("type") + if not isinstance(kind, str) or not kind: + raise _DshFormatError("invalid event type") + if kind in _PACKED_TYPES: + if version >= 2: + raise _DshFormatError("packed row in v2 session") + count = _packed_count(record) + if record["seq0"] != expected_seq: + raise _DshFormatError("packed sequence gap") + end_time = record["time0"] + sum(record["data"]["dt"]) + if not _is_safe_int(end_time) or not _is_safe_int(expected_seq + count - 1): + raise _DshFormatError("packed event overflow") + return _DshEvent(expected_seq + count, end_time) + + # Ignorable extensions have exactly the same envelope and sequence rules. + if not _is_safe_int(record.get("seq")) or record["seq"] != expected_seq: + raise _DshFormatError("non-contiguous sequence") + if not _is_safe_int(record.get("time")) or not isinstance(record.get("data"), dict): + raise _DshFormatError("invalid event envelope") + if kind not in _KNOWN_EVENT_TYPES and record.get("ignorable") is not True: + raise _DshFormatError("unknown event") + + next_seq, timestamp = expected_seq + 1, record["time"] + if kind in _SURFACE_EVENT_TYPES: + replacement = _surface_replacement(record, version) + text, names = "", [] + if kind == "user/message": + text = _human_user_text(record) + elif kind == "assistant/message": + text, names = _assistant_message(record) + # Tool results and injected context still occupy the surface, but their + # raw content is unnecessary even when later replacements target them. + entry = _SurfaceEntry(expected_seq, kind, text, tuple(_dedup(names))) + return _DshEvent(next_seq, timestamp, entry, replacement) + if kind == "tool/call": + return _DshEvent(next_seq, timestamp, tool=_tool_call_name(record)) + if kind == "feedback/record": + return _DshEvent(next_seq, timestamp, feedback=tuple(_feedback_signals(record))) + if kind == "session/end-seed": + inherited = record["data"].get("inherited") + if inherited is not None and inherited is not True: + raise _DshFormatError("invalid inherited seed marker") + return _DshEvent(next_seq, timestamp, inherited_seed_marker=inherited is True) + return _DshEvent(next_seq, timestamp) + + +def _apply_surface( + surface: list[_SurfaceEntry], entry: _SurfaceEntry, replacement: Optional[_Replacement], +) -> None: + """Apply one normalized update without retaining raw or shadowed records.""" + if replacement is None: + surface.append(entry) + return + try: + start = next(index for index, item in enumerate(surface) if item.seq == replacement.start) + end = next(index for index, item in enumerate(surface) if item.seq == replacement.end) + except StopIteration as exc: + raise _DshFormatError("surface replacement range is not visible") from exc + if start > end: + raise _DshFormatError("invalid surface replacement range") + if replacement.sources is not None: + if any(surface[index].seq not in replacement.sources for index in range(start, end + 1)): + raise _DshFormatError("incomplete v2 surface sources") + surface[start : end + 1] = [entry] + + def _is_dsh_replay(digest: SessionDigest) -> bool: return bool(digest.user_prompts) and digest.user_prompts[0].lstrip().startswith(DSH_REPLAY_SENTINEL) -def digest_dsh_session(path: str, *, root: str) -> Optional[SessionDigest]: +def digest_dsh_session( + path: str, + *, + root: str, + progress: bool = False, + scope: Any = "all", + invoked_project: str = "", +) -> Optional[SessionDigest]: """Parse one complete DSH session file, returning ``None`` on any failure.""" + records = _iter_records(path) try: - records = _iter_records(path) header = _header_from_record(next(records), path, root) if header.get("origin") == "subagent": return None session_id = str(header["id"]) project = str(header.get("cwd") or "") + if (not project and scope != "all") or not _project_matches(project, scope, invoked_project): + return None started_at = _iso_timestamp(header["createdAt"]) ended_at = started_at user_prompts: list[str] = [] - assistant_finals: list[str] = [] - tools: list[str] = [] - feedback: list[str] = [] + assistant_finals: deque[str] = deque(maxlen=5) + tools: dict[str, None] = {} + feedback: dict[str, None] = {} + surface: list[_SurfaceEntry] = [] expected_seq = 0 n_user = 0 n_assistant = 0 + has_inherited_seed_marker = False for record in records: - row_type = record.get("type") - if row_type in _PACKED_TYPES: - count = _packed_count(record) - if record["seq0"] != expected_seq: - raise _DshFormatError("packed sequence gap") - expected_seq += count - # Packed chunks are intentionally not retained, but their final - # timestamp is still the best session-end timestamp. - deltas = record["data"]["dt"] - ended_at = _iso_timestamp(record["time0"] + sum(deltas)) - continue - - _validate_event(record, expected_seq) - if record.get("type") in _KNOWN_EVENT_TYPES: - expected_seq += 1 - ended_at = _iso_timestamp(record["time"]) - # An ignorable extension has a normal event envelope and therefore - # still occupies one sequence number. - elif record.get("ignorable") is True: - if record.get("seq") != expected_seq or not _is_safe_int(record.get("time")): - raise _DshFormatError("invalid ignorable event") - expected_seq += 1 - ended_at = _iso_timestamp(record["time"]) - - event_type = record.get("type") - if event_type == "user/message": - text = _human_user_text(record) + event = _decode_event(record, expected_seq, header["version"]) + expected_seq = event.next_seq + ended_at = _iso_timestamp(event.time) + if event.surface is not None: + _apply_surface(surface, event.surface, event.replacement) + if event.tool: + tools[event.tool] = None + feedback.update(dict.fromkeys(event.feedback)) + has_inherited_seed_marker = has_inherited_seed_marker or event.inherited_seed_marker + + if header["version"] >= 2 and bool(header["isSeeded"]) != has_inherited_seed_marker: + raise _DshFormatError("v2 seeded header and inherited seed marker disagree") + + for entry in surface: + if entry.kind == "user/message": + text = entry.text if text: user_prompts.append(text) - feedback.extend(_detect_feedback(text)) + feedback.update(dict.fromkeys(_detect_feedback(text))) n_user += 1 - elif event_type == "assistant/message": - text, names = _assistant_message(record) - tools.extend(names) + elif entry.kind == "assistant/message": + tools.update(dict.fromkeys(entry.tools)) n_assistant += 1 - if text: - assistant_finals.append(text) - elif event_type == "tool/call": - tools.append(_tool_call_name(record)) - elif event_type == "feedback/record": - feedback.extend(_feedback_signals(record)) - + if entry.text: + assistant_finals.append(entry.text) if not user_prompts and not assistant_finals: return None @@ -438,17 +647,20 @@ def digest_dsh_session(path: str, *, root: str) -> Optional[SessionDigest]: started_at=started_at, ended_at=ended_at, user_prompts=user_prompts, - assistant_finals=assistant_finals[-5:], - tools_used=_dedup(tools), + assistant_finals=list(assistant_finals), + tools_used=list(tools), files_touched=[], - feedback_signals=_dedup(feedback), + feedback_signals=list(feedback), n_user_turns=n_user, n_assistant_turns=n_assistant, raw_path=path, ) return None if _is_dsh_replay(digest) else digest - except (OSError, StopIteration, _DshFormatError, ValueError, TypeError, json.JSONDecodeError): + except (OSError, StopIteration, ValueError, TypeError) as exc: + _diagnostic("Skipping DSH session file %s: %s", path, exc, progress=progress) return None + finally: + records.close() def harvest_dsh( @@ -458,6 +670,7 @@ def harvest_dsh( invoked_project: str = "", since_iso: Optional[str] = None, limit: int = 0, + progress: bool = False, ) -> list[SessionDigest]: """Discover valid DSH session logs below one explicitly supplied root.""" if not session_root: @@ -466,33 +679,86 @@ def harvest_dsh( if not os.path.isdir(root): return [] - candidates: list[tuple[float, str]] = [] + grouped: dict[str, list[tuple[int, str, str, float]]] = {} for directory, _dirs, files in os.walk(root, followlinks=False): for filename in files: - if filename not in _LOG_NAMES: + parsed_name = _parse_log_filename(filename) + if parsed_name is None: continue path = os.path.join(directory, filename) if not _is_candidate_path(root, path): continue try: - candidates.append((os.path.getmtime(path), path)) + version, encoding = parsed_name + session_dir = os.path.dirname(path) + grouped.setdefault(session_dir, []).append( + (version, encoding, path, os.path.getmtime(path)) + ) except OSError: continue + + candidates: list[tuple[float, str]] = [] + for session_dir, entries in grouped.items(): + encodings = {entry[1] for entry in entries} + if len(encodings) != 1: + _diagnostic( + "Skipping DSH session directory %s: mixed log encodings", + session_dir, + progress=progress, + ) + continue + highest_version = max(entry[0] for entry in entries) + highest = [entry for entry in entries if entry[0] == highest_version] + # A canonical session directory has at most one file for a generation + # and encoding. Ambiguous duplicates are safer to skip than guess at. + if len(highest) != 1: + _diagnostic( + "Skipping DSH session directory %s: ambiguous files for highest generation v%d", + session_dir, + highest_version, + progress=progress, + ) + continue + _version, _encoding, path, mtime = highest[0] + if highest_version not in _SUPPORTED_FORMAT_VERSIONS: + _diagnostic( + "Skipping DSH session directory %s: selected file %s has unsupported highest " + "generation v%d; supported generations are %s. Upgrade SkillOpt to support " + "this DSH format.", + session_dir, + path, + highest_version, + ", ".join(str(version) for version in sorted(_SUPPORTED_FORMAT_VERSIONS)), + progress=progress, + ) + continue + _diagnostic( + "Selected DSH session file %s (highest generation v%d)", + path, + highest_version, + progress=progress, + ) + candidates.append((mtime, path)) candidates.sort(key=lambda item: (-item[0], item[1])) digests: list[SessionDigest] = [] seen_ids: set[str] = set() + since_epoch = _iso_epoch(since_iso) for _mtime, path in candidates: - digest = digest_dsh_session(path, root=root) + digest = digest_dsh_session( + path, root=root, progress=progress, scope=scope, invoked_project=invoked_project, + ) if digest is None or digest.session_id in seen_ids: continue seen_ids.add(digest.session_id) - if not digest.project and scope != "all": - continue - if not _project_matches(digest.project, scope, invoked_project): - continue - if since_iso and digest.ended_at and digest.ended_at < since_iso: - continue + if since_iso and digest.ended_at: + ended_epoch = _iso_epoch(digest.ended_at) + if since_epoch is not None and ended_epoch is not None: + if ended_epoch < since_epoch: + continue + elif digest.ended_at < since_iso: + # Preserve best-effort behavior for malformed legacy cutoffs. + continue digests.append(digest) if limit and len(digests) >= limit: break diff --git a/skillopt_sleep/harvest_sources.py b/skillopt_sleep/harvest_sources.py index 9bded4cd..d71c85a3 100644 --- a/skillopt_sleep/harvest_sources.py +++ b/skillopt_sleep/harvest_sources.py @@ -74,6 +74,7 @@ def harvest_for_config(cfg, *, since_iso: Optional[str] = None, limit: int = 0) invoked_project=invoked_project, since_iso=since_iso, limit=limit, + progress=bool(cfg.get("progress", False)), ) if source == "auto": codex_digests = harvest_codex( diff --git a/tests/fixtures/dsh/--fixture-project--/master-v2-fixture/session.v2.jsonl.zstd b/tests/fixtures/dsh/--fixture-project--/master-v2-fixture/session.v2.jsonl.zstd new file mode 100644 index 0000000000000000000000000000000000000000..4a54da40a42b58bfb5d8543d0bd9d2a6f52fac6b GIT binary patch literal 2032 zcmVwk|hwW)&YixqEwFQrjYm-(rm$*rtcKz3?_hJEPhG{j0yliieM0j z0EGa806w3RVP4n1R0G>#A1^#ycFeI@6tL*NaxTc1Rl$5VC8NJjVfG*3K5Vus9@J#D zeP0RL^?xVVha7DgyC@+u9`59LC>Xnr45t$9G3w)N>&L)!r9qVdnaoF&vy}XN<5H8gk>XcH%kLZ8W2Jq6pSs4#&g3@N z{ri$a?)0yCH&xmJ-;dhVTapMS$griZKc2nx0!E+P);zHUn>kL?wJ;(Ps}SIxf2D+? znx|00=4UIPw*+~bE@D*MXA1#2K(ZR#gJmlDl#$5Goe}Fw5}8!WeT=vtfy903%ob=X z(Qh5{aMWZ$pAU!u1JbF~tv|l@_}DGgNK(RdVOohGM{v4xyLeA5)fhm1uUTv+qo2Kx zStlwD5w+sChmZa#{0MP2L(vcE)d%B|_!XLZ*A5eVeLSQg2UrRh8} zki&5TDAMsr9Or?a=V>&KY8=QzYskd{W*g)M^ms%hlE}KiX<<*-Uq+w_a--=ctc59y zIQ@S7><5zkjF2bfNZ3)XtK~Qk^(9NG8OQ-a9!Kuab_B<>c+M>vjfrlM8yqJ?nVj{v zNwq}D&Z4UO(vieD7#!l_lYFx) zqOLbKJ}fbD>j&`XyagUHgHkqXOI6aJK6v9p7@%gW*>_TcF!K@@^!dX12w8wAa*RD# z5zh{9|1mqk^>Ue?iJ$2a37cx9`J&v8H3-k_0(%5F+|(2xf_VqjjHnF^l^0OfM4+#p ziM%s4D-jxux!(l&Q-s^V1OHlT+gl`A+!VOKl5QCn9a8$=$Ypi`#vHEXlM9{AW>0Fn`JX z322>~;M?6M%Pb~FC=SjHYDk7=#rLP>tQTO4gc}qX7qFJ9l=u9qoJ2^qT{8_$kCVpH z-w_4c%H>pxMTvGJE6<5Xuvd>62m?FgMNu3MC2as#&ABPFO(f^ePyyr+$3xCVWR#X! zOXwTJjJ$RkXs}uqYb-Q#P5<`nO14 zJ-vk2r=_;D^Qt}#5fiLEbb{gD&bIWfR5gzb+P->?Ak0=AMiys83c{sBVkcw;zyeSB z8$45qCk|@z51-5Ufe-=C z4KZ1sBJ+3Wpp2xte3E&%+zQ|yZX4#&UKNK8d@LN22T!I~82XlvCf%Zpk9hMCFq81Nu{<-L& ztROF`h8{U96OKO2mRq(cNr-e4*}4DSh=ho_-+2nVNDys+Vket$C&j@HXm-a&Cn#_^ z<#qxq;8fKJ&a8Ubj>OU&O$@!tXO#1c3}mYY@&L)O-&m{+2q2b*l#-t7n)@2oiA;pN z^Oe)uhz$IB{kEN*QlLvG>|xZo=RNF}pRSWB(|N9c>N$C_i!rUGO&s*pQ8dnB~X9jYSC&$AVb&nwSn(AVEMOR& zDm}L3ahSl#y5a{7WAqYSeO!5#2;NHvgjP#$FxBUr0VH0cV2+Q~Fr`|ia3&zQLzwq9 zr_vk9iz2?HL-5*wtfeApmRlp4qUh=!iz$Be29!ht3~%jZ)nqzoiSwqP3?-@)mNNbT znt)6YMzcGuHQxmU18%SSS|pmUV9_|{x{^Pz-8Hvv9d= 2: + # V2 makes the fork-lineage bit explicit. Keep it out of legacy + # headers, where the historical codec used seedLength instead. + value["isSeeded"] = False if cwd is not None: value["cwd"] = cwd value.update(extra) @@ -80,6 +88,13 @@ def _assistant(seq: int, text: str, *, tool_name="", replace=False): return event +def _replace_surface(event: dict, start: int, end: int, *source_event_seqs: int) -> dict: + event["surfaceOp"] = {"op": "replace", "start": start, "end": end} + if source_event_seqs: + event["sourceEventSeqs"] = list(source_event_seqs) + return event + + def _tool_call(seq: int, name: str): return { "type": "tool/call", @@ -107,11 +122,29 @@ def _metadata(seq: int, event_type: str): } -def _write_raw(root: Path, session_id: str, cwd: str | None, records: list[dict], **header_extra) -> Path: +def _end_seed(seq: int, *, inherited=True): + return { + "type": "session/end-seed", + "seq": seq, + "time": _BASE_TIME + 1000 * (seq + 1), + "data": {"inherited": inherited} if inherited else {}, + } + + +def _write_raw( + root: Path, + session_id: str, + cwd: str | None, + records: list[dict], + *, + version=0, + **header_extra, +) -> Path: project_dir = "_no-cwd" if cwd is None else _project_key(cwd) - path = root / project_dir / _encode_segment(session_id) / "session.jsonl" + filename = "session.jsonl" if version == 0 else f"session.v{version}.jsonl" + path = root / project_dir / _encode_segment(session_id) / filename path.parent.mkdir(parents=True, exist_ok=True) - rows = [_header(session_id, cwd, **header_extra), *records] + rows = [_header(session_id, cwd, version=version, **header_extra), *records] path.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") return path @@ -196,9 +229,10 @@ def test_fork_is_retained_but_subagent_and_replay_are_excluded(tmp_path: Path): tmp_path, "fork", project, - [_user(0, "inherited request"), _assistant(1, "superseded", replace=True), _assistant(2, "active final")], + [_user(0, "inherited request"), _end_seed(1), _assistant(2, "active final")], + version=2, parentSession="parent", - seedLength=1, + isSeeded=True, ) _write_raw( tmp_path, @@ -221,6 +255,65 @@ def test_fork_is_retained_but_subagent_and_replay_are_excluded(tmp_path: Path): assert digests[0].assistant_finals == ["active final"] +def test_surface_replacement_removes_shadowed_assistant_text(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + compacted = _replace_surface( + _user(2, "compacted context", source="system", append=False), + 0, + 1, + 0, + 1, + ) + _write_raw( + tmp_path, + "compacted", + project, + [ + _user(0, "request"), + _assistant(1, "obsolete assistant text"), + compacted, + _assistant(3, "current assistant text"), + ], + ) + + digest = harvest_dsh(str(tmp_path), scope="all")[0] + + assert digest.assistant_finals == ["current assistant text"] + assert "obsolete assistant text" not in json.dumps(digest.to_dict()) + + +def test_v2_range_encoded_replacement_provenance_is_replayed(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + tool_result = {**_metadata(2, "tool/result"), "surfaceOp": "append"} + tool_result["data"] = {"output": "private tool output"} + replacement = _replace_surface( + _user(4, "compacted context", source="system", append=False), 1, 3, 1, 2, 3, + ) + # This is the physical v2 representation written by the upstream codec: + # a run of three adjacent source sequence numbers becomes [start, end]. + replacement["sourceEventSeqs"] = [[1, 3]] + _write_raw( + tmp_path, + "range-provenance", + project, + [ + _user(0, "request"), + _assistant(1, "obsolete assistant text"), + tool_result, + _user(3, "injected context", source="plugin"), + replacement, + _assistant(5, "current assistant text"), + ], + version=2, + ) + + digest = harvest_dsh(str(tmp_path), scope="all")[0] + + assert digest.user_prompts == ["request"] + assert digest.assistant_finals == ["current assistant text"] + assert "obsolete assistant text" not in json.dumps(digest.to_dict()) + + def test_bad_session_is_silent_and_does_not_block_other_sessions(tmp_path: Path): project = str((tmp_path / "repo").resolve()) _write_raw(tmp_path, "good", project, [_user(0, "good request"), _assistant(1, "good final")]) @@ -233,6 +326,191 @@ def test_bad_session_is_silent_and_does_not_block_other_sessions(tmp_path: Path) assert [digest.session_id for digest in digests] == ["good"] +def test_v2_only_session_is_harvested(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + path = _write_raw( + tmp_path, + "v2-only", + project, + [_user(0, "current request"), _assistant(1, "current final")], + version=2, + ) + + digests = harvest_dsh(str(tmp_path), scope="all") + + assert [digest.session_id for digest in digests] == ["v2-only"] + assert digests[0].raw_path == str(path) + + +def test_v2_header_uses_the_published_required_and_allowed_keys(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + missing = _write_raw( + tmp_path, + "missing-seeded", + project, + [_user(0, "request"), _assistant(1, "final")], + version=2, + ) + rows = [json.loads(line) for line in missing.read_text(encoding="utf-8").splitlines()] + del rows[0]["isSeeded"] + missing.write_text("".join(json.dumps(row) + "\n" for row in rows), encoding="utf-8") + _write_raw( + tmp_path, + "retired-seed-length", + project, + [_user(0, "request"), _assistant(1, "final")], + version=2, + seedLength=0, + ) + _write_raw( + tmp_path, + "relative-cwd", + "relative/project", + [_user(0, "request"), _assistant(1, "final")], + version=2, + ) + + assert harvest_dsh(str(tmp_path), scope="all") == [] + + +def test_v2_seeded_header_must_match_end_seed_marker(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + _write_raw( + tmp_path, + "seeded-without-marker", + project, + [_user(0, "request"), _assistant(1, "final")], + version=2, + isSeeded=True, + ) + _write_raw( + tmp_path, + "unseeded-with-marker", + project, + [_user(0, "request"), _end_seed(1), _assistant(2, "final")], + version=2, + ) + invalid_marker = _end_seed(1) + invalid_marker["data"]["inherited"] = False + _write_raw( + tmp_path, + "invalid-marker", + project, + [_user(0, "request"), invalid_marker, _assistant(2, "final")], + version=2, + ) + + assert harvest_dsh(str(tmp_path), scope="all") == [] + + +def test_master_v2_fixture_is_harvested(): + pytest.importorskip("zstandard") + + digests = harvest_dsh(str(_MASTER_V2_FIXTURE_ROOT), scope="all") + + assert [digest.session_id for digest in digests] == ["master-v2-fixture"] + assert digests[0].project == "/fixture/project" + assert digests[0].user_prompts == ["fixture user request"] + assert digests[0].assistant_finals == ["fixture assistant response"] + assert digests[0].raw_path.endswith("session.v2.jsonl.zstd") + + +def test_v2_packed_rows_are_rejected(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + packed = { + "type": "text-chunks", + "seq0": 1, + "time0": _BASE_TIME + 2000, + "data": { + "turn": 1, + "step": 1, + "index": 0, + "dt": [0, 7, 9], + "texts": ["a", "b", "c"], + }, + } + _write_raw( + tmp_path, + "v2-packed", + project, + [_user(0, "request"), packed, _assistant(4, "final")], + version=2, + ) + + assert harvest_dsh(str(tmp_path), scope="all") == [] + + +def test_highest_generation_wins_over_retained_predecessors(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + _write_raw( + tmp_path, + "migrated", + project, + [_user(0, "old request"), _assistant(1, "old final")], + version=0, + ) + _write_raw( + tmp_path, + "migrated", + project, + [_user(0, "v1 request"), _assistant(1, "v1 final")], + version=1, + ) + current = _write_raw( + tmp_path, + "migrated", + project, + [_user(0, "current request"), _assistant(1, "current final")], + version=2, + ) + + digests = harvest_dsh(str(tmp_path), scope="all") + + assert [digest.session_id for digest in digests] == ["migrated"] + assert digests[0].user_prompts == ["current request"] + assert digests[0].raw_path == str(current) + + +def test_unsupported_highest_generation_does_not_fall_back(tmp_path: Path): + project = str((tmp_path / "repo").resolve()) + _write_raw( + tmp_path, + "future", + project, + [_user(0, "old request"), _assistant(1, "old final")], + version=0, + ) + _write_raw( + tmp_path, + "future", + project, + [_user(0, "future request"), _assistant(1, "future final")], + version=10, + ) + + assert harvest_dsh(str(tmp_path), scope="all") == [] + + +def test_unsupported_highest_generation_is_diagnosed(tmp_path: Path, caplog, capsys): + project = str((tmp_path / "repo").resolve()) + _write_raw( + tmp_path, + "future", + project, + [_user(0, "future request"), _assistant(1, "future final")], + version=10, + ) + + with caplog.at_level(logging.DEBUG, logger="skillopt_sleep.harvest_dsh"): + assert harvest_dsh(str(tmp_path), scope="all", progress=True) == [] + + diagnostic = " ".join(record.getMessage() for record in caplog.records) + assert "session.v10.jsonl" in diagnostic + assert "highest generation v10" in diagnostic + assert "Upgrade SkillOpt" in diagnostic + assert "session.v10.jsonl" in capsys.readouterr().err + + def test_unknown_required_event_rejects_but_ignorable_event_is_skipped(tmp_path: Path): project = str((tmp_path / "repo").resolve()) _write_raw( @@ -300,10 +578,10 @@ def test_scope_since_limit_and_identity_checks(tmp_path: Path): def test_zstd_concatenated_frames_are_read(tmp_path: Path): zstd = pytest.importorskip("zstandard") project = str((tmp_path / "repo").resolve()) - path = tmp_path / _project_key(project) / _encode_segment("compressed") / "session.jsonl.zstd" + path = tmp_path / _project_key(project) / _encode_segment("compressed") / "session.v2.jsonl.zstd" path.parent.mkdir(parents=True) compressor = zstd.ZstdCompressor(write_checksum=True) - header = json.dumps(_header("compressed", project)).encode() + b"\n" + header = json.dumps(_header("compressed", project, version=2)).encode() + b"\n" events = b"".join( json.dumps(row).encode() + b"\n" for row in [_user(0, "compressed request"), _assistant(1, "compressed final")] @@ -343,6 +621,7 @@ def test_cli_config_and_source_dispatch_for_dsh(monkeypatch, tmp_path: Path): invoked_project=project, since_iso="2026-01-01T00:00:00Z", limit=2, + progress=False, ) claude.assert_not_called() codex.assert_not_called() @@ -369,3 +648,148 @@ def test_auto_source_does_not_add_dsh_precedence(tmp_path: Path): ): assert harvest_for_config(cfg) == expected dsh.assert_not_called() + + +@pytest.mark.parametrize("offset_hours", [-7, 0, 8]) +@pytest.mark.parametrize("cutoff_delta_ms,keep", [(-1, True), (0, True), (1, False)]) +def test_since_compares_instants_with_millisecond_precision(tmp_path, offset_hours, cutoff_delta_ms, keep): + end_ms = _BASE_TIME + 2123 + answer = _assistant(1, "final") + answer["time"] = end_ms + _write_raw(tmp_path, "timestamp", str(tmp_path), [_user(0, "request"), answer], version=2) + cutoff = datetime.fromtimestamp((end_ms + cutoff_delta_ms) / 1000, timezone.utc) + since = cutoff.astimezone(timezone(timedelta(hours=offset_hours))).isoformat() + + assert bool(harvest_dsh(str(tmp_path), since_iso=since)) is keep + + +def test_since_accepts_local_sleep_checkpoint(tmp_path): + _write_raw(tmp_path, "local-time", str(tmp_path), [_user(0, "request"), _assistant(1, "final")]) + # Match state._now_iso's host-local, offset-free timestamps on any platform. + before = datetime.fromtimestamp((_BASE_TIME + 1000) / 1000).isoformat() + after = datetime.fromtimestamp((_BASE_TIME + 3000) / 1000).isoformat() + + assert len(harvest_dsh(str(tmp_path), since_iso=before)) == 1 + assert harvest_dsh(str(tmp_path), since_iso=after) == [] + + +def test_master_fixture_since_accepts_equivalent_offsets(): + pytest.importorskip("zstandard") + utc = harvest_dsh(str(_MASTER_V2_FIXTURE_ROOT), since_iso="2027-01-15T08:00:16Z") + offset = harvest_dsh(str(_MASTER_V2_FIXTURE_ROOT), since_iso="2027-01-15T16:00:16+08:00") + + assert len(utc) == 1 + assert offset == utc + + +@pytest.mark.parametrize("kind", ["turn/start", "plugin/info"]) +@pytest.mark.parametrize("bad_fields", [ + {"seq": True}, {"seq": 1.0}, {"seq": -1}, {"seq": 2}, + {"time": True}, {"time": -1}, {"data": []}, {"data": None}, + {"type": []}, {"type": ""}, {"type": None}, +]) +def test_known_and_ignorable_events_share_envelope_validation(tmp_path, kind, bad_fields): + invalid = {**_metadata(1, kind), "ignorable": True, **bad_fields} + _write_raw(tmp_path, "invalid", str(tmp_path), [_user(0, "request"), invalid, _assistant(2, "final")]) + _write_raw(tmp_path, "valid", str(tmp_path), [_user(0, "request"), _assistant(1, "final")]) + + assert [digest.session_id for digest in harvest_dsh(str(tmp_path))] == ["valid"] + + +@pytest.mark.parametrize("version", [0, 1, 2]) +def test_incremental_replacements_keep_result_and_injected_context_positions(tmp_path, version): + result = {**_metadata(2, "tool/result"), "surfaceOp": "append"} + result["data"] = {"output": "private tool output"} + records = [ + _user(0, "request"), + _assistant(1, "obsolete", tool_name="obsolete-tool"), + result, + _user(3, "injected context", source="plugin"), + _replace_surface(_user(4, "summary", source="system"), 1, 3, 1, 2, 3), + _assistant(5, "intermediate", tool_name="intermediate-tool"), + _replace_surface(_user(6, "new summary", source="system"), 4, 5, 4, 5), + _assistant(7, "current", tool_name="current-tool"), + ] + path = _write_raw(tmp_path, "replaced", str(tmp_path), records, version=version) + + digest = digest_dsh_session(str(path), root=str(tmp_path)) + + assert digest is not None + assert digest.user_prompts == ["request"] + assert digest.assistant_finals == ["current"] + assert digest.tools_used == ["current-tool"] + assert digest.n_assistant_turns == 1 + + +@pytest.mark.parametrize("replacement_fields", [ + {"sourceEventSeqs": [0]}, + {"sourceEventSeqs": [0, 1, 1]}, + {"sourceEventSeqs": [0, True]}, + {"sourceEventSeqs": [[1]]}, + {"sourceEventSeqs": [[1, 0]]}, + {"sourceEventSeqs": [[0, 2]]}, + {"sourceEventSeqs": [[0, 1], [1, 2]]}, + {"sourceEventSeqs": None}, + {"surfaceOp": {"op": "replace", "start": 1, "end": 0}}, + {"surfaceOp": {"op": "replace", "start": 0, "end": 99}}, +]) +def test_invalid_v2_replacement_discards_complete_session(tmp_path, replacement_fields): + replacement = _replace_surface(_user(2, "summary", source="system"), 0, 1, 0, 1) + replacement.update(replacement_fields) + path = _write_raw(tmp_path, "bad-replace", str(tmp_path), [ + _user(0, "request"), _assistant(1, "old"), replacement, _assistant(3, "new"), + ], version=2) + + assert digest_dsh_session(str(path), root=str(tmp_path)) is None + + +def test_digest_releases_raw_payloads_while_reading(tmp_path): + class Payload(str): + pass + + refs = [] + path = _write_raw(tmp_path, "streamed", str(tmp_path), [], version=2) + + def records(): + yield _header("streamed", str(tmp_path), version=2) + yield _user(0, "request") + for seq in range(1, 21): + # Allow the reader's current row; earlier payloads must be freed + # before EOF, even if their surface positions remain visible. + assert all(ref() is None for ref in refs[:-1]) + payload = Payload("private data " * 10000) + refs.append(weakref.ref(payload)) + if seq % 2: + row = _assistant(seq, "visible reply", tool_name="shell") + row["data"]["message"]["content"][0]["text"] = payload + row["data"]["message"]["content"][-1]["arguments"] = payload + else: + row = {**_metadata(seq, "tool/result"), "surfaceOp": "append", "data": {"output": payload}} + yield row + del row, payload + + with mock.patch("skillopt_sleep.harvest_dsh._iter_records", side_effect=lambda _path: records()): + digest = digest_dsh_session(str(path), root=str(tmp_path)) + + assert digest is not None + assert digest.assistant_finals == ["visible reply"] * 5 + assert digest.n_assistant_turns == 10 + assert digest.tools_used == ["shell"] + assert all(ref() is None for ref in refs) + + +def test_unrelated_project_is_skipped_before_events_are_read(tmp_path): + _write_raw(tmp_path, "other", str(tmp_path / "other"), [], version=2) + closed = [] + + def records(_path): + try: + yield _header("other", str(tmp_path / "other"), version=2) + raise AssertionError("unrelated session body should not be read") + finally: + closed.append(True) + + with mock.patch("skillopt_sleep.harvest_dsh._iter_records", side_effect=records): + assert harvest_dsh(str(tmp_path), scope="invoked", invoked_project=str(tmp_path / "wanted")) == [] + + assert closed == [True]