From 84d64abf2c3dbad3e5ac94541de77b28cda0c7fc Mon Sep 17 00:00:00 2001 From: TaoRunguo Date: Thu, 3 Sep 2026 16:08:11 +0800 Subject: [PATCH] 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 6aefbb20..9b50d2e7 100644 --- a/docs/reference/cli.md +++ b/docs/reference/cli.md @@ -134,7 +134,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 | @@ -144,6 +144,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()