url/SkillOpt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Codex Desktop transcript harvesting

Browse Source
This commit is contained in:
Kirill Kostarev
2026-06-12 16:37:23 +03:00
committed by carpedkm
parent e8c3e10b30
commit 31715a8b43
12 changed files with 479 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skillopt_sleep/__main__.py
+13 -9
View File
@@ -9,7 +9,8 @@
Common flags:
--project PATH project to evolve (default: cwd)
--scope all|invoked harvest scope (default: invoked)
--backend mock|anthropic
--backend mock|claude|codex
--source claude|codex|auto
--model NAME
--lookback-hours N
--auto-adopt
@@ -25,10 +26,11 @@ from typing import Any, Dict
from skillopt_sleep.config import load_config
from skillopt_sleep.cycle import run_sleep_cycle
from skillopt_sleep.harvest import harvest
from skillopt_sleep.harvest_sources import harvest_for_config
from skillopt_sleep.mine import mine
from skillopt_sleep.staging import adopt as adopt_staging
from skillopt_sleep.staging import latest_staging
from skillopt_sleep.state import SleepState
from skillopt_sleep.staging import latest_staging, adopt as adopt_staging
def _add_common(p: argparse.ArgumentParser) -> None:
@@ -38,6 +40,9 @@ def _add_common(p: argparse.ArgumentParser) -> None:
p.add_argument("--model", default="")
p.add_argument("--codex-path", default="", help="path to the real @openai/codex binary")
p.add_argument("--claude-home", default="", help="override ~/.claude (also isolates state)")
p.add_argument("--codex-home", default="", help="override ~/.codex for archived session harvest")
p.add_argument("--source", default="", choices=["", "claude", "codex", "auto"],
help="session transcript source")
p.add_argument("--lookback-hours", type=int, default=0)
p.add_argument("--edit-budget", type=int, default=0)
p.add_argument("--auto-adopt", action="store_true")
@@ -59,6 +64,10 @@ def _cfg_from_args(args) -> Any:
overrides["codex_path"] = os.path.abspath(args.codex_path)
if getattr(args, "claude_home", ""):
overrides["claude_home"] = os.path.abspath(args.claude_home)
if getattr(args, "codex_home", ""):
overrides["codex_home"] = os.path.abspath(args.codex_home)
if getattr(args, "source", ""):
overrides["transcript_source"] = args.source
if getattr(args, "lookback_hours", 0):
overrides["lookback_hours"] = args.lookback_hours
if getattr(args, "edit_budget", 0):
@@ -143,12 +152,7 @@ def cmd_adopt(args) -> int:
def cmd_harvest(args) -> int:
cfg = _cfg_from_args(args)
digests = harvest(
cfg.transcripts_dir,
scope=cfg.get("projects", "invoked"),
invoked_project=cfg.get("invoked_project", ""),
limit=cfg.get("max_tasks_per_night", 40) * 3,
)
digests = harvest_for_config(cfg, limit=cfg.get("max_tasks_per_night", 40) * 3)
tasks = mine(digests, max_tasks=cfg.get("max_tasks_per_night", 40),
holdout_fraction=cfg.get("holdout_fraction", 0.34), seed=cfg.get("seed", 42))
if args.json:
skillopt_sleep/config.py
+9 -3
View File
@@ -13,17 +13,19 @@ from __future__ import annotations
import json
import os
from dataclasses import dataclass, field, asdict
from typing import Any, Dict, List, Optional
from dataclasses import dataclass, field
from typing import Any, Dict, Optional
HOME_STATE_DIR = os.path.expanduser("~/.skillopt-sleep")
CLAUDE_HOME = os.path.expanduser("~/.claude")
CODEX_HOME = os.path.expanduser("~/.codex")
DEFAULTS: Dict[str, Any] = {
# ── scope ──────────────────────────────────────────────────────────────
"claude_home": CLAUDE_HOME,
"codex_home": CODEX_HOME,
"transcript_source": "claude", # "claude" | "codex" | "auto"
"projects": "invoked", # "invoked" | "all" | [list of abs paths]
"invoked_project": "", # filled at runtime (cwd) when projects == "invoked"
"lookback_hours": 72, # harvest window when no prior sleep recorded
@@ -94,6 +96,10 @@ class SleepConfig:
def transcripts_dir(self) -> str:
return os.path.join(self.data["claude_home"], "projects")
@property
def codex_archived_sessions_dir(self) -> str:
return os.path.join(self.data["codex_home"], "archived_sessions")
@property
def history_path(self) -> str:
return os.path.join(self.data["claude_home"], "history.jsonl")
skillopt_sleep/cycle.py
+7 -9
View File
@@ -10,18 +10,18 @@ CI use. With backend="anthropic" it spends the user's budget for real lift.
from __future__ import annotations
import os
import time
from dataclasses import dataclass
from typing import Any, Dict, List, Optional
from typing import List, Optional
from skillopt_sleep.backend import get_backend
from skillopt_sleep.config import SleepConfig, load_config
from skillopt_sleep.consolidate import consolidate
from skillopt_sleep.harvest import harvest
from skillopt_sleep.harvest_sources import harvest_for_config
from skillopt_sleep.memory import ensure_skill_scaffold
from skillopt_sleep.mine import mine
from skillopt_sleep.staging import adopt as adopt_staging
from skillopt_sleep.staging import write_staging
from skillopt_sleep.state import SleepState, _now_iso
from skillopt_sleep.staging import write_staging, adopt as adopt_staging
from skillopt_sleep.types import SessionDigest, SleepReport, TaskRecord
@@ -117,10 +117,8 @@ def run_sleep_cycle(
n_sessions = 0
else:
since = state.last_harvest_for(project)
digests = harvest(
cfg.transcripts_dir,
scope=cfg.get("projects", "invoked"),
invoked_project=cfg.get("invoked_project", ""),
digests = harvest_for_config(
cfg,
since_iso=since,
limit=cfg.get("max_tasks_per_night", 40) * 3,
)
@@ -151,7 +149,7 @@ def run_sleep_cycle(
if not skill:
skill = ensure_skill_scaffold(
"", name=cfg.get("managed_skill_name", "skillopt-sleep-learned"),
description="Preferences and procedures learned from past Claude Code sessions.",
description="Preferences and procedures learned from past local agent sessions.",
)
report = SleepReport(
skillopt_sleep/harvest_codex.py
+253
View File
@@ -0,0 +1,253 @@
"""SkillOpt-Sleep Codex Desktop session harvesting.
Reads Codex Desktop archived session JSONL files and normalizes them into
``SessionDigest`` records without copying developer/system instructions, tool
arguments, or raw tool outputs.
"""
from __future__ import annotations
import os
import re
from typing import Any, Dict, Iterable, List, Optional
from skillopt_sleep.harvest import (
_detect_feedback,
_is_meta_prompt,
_iter_jsonl,
_project_matches,
)
from skillopt_sleep.types import SessionDigest
_SECRET_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
(re.compile(r"sk-[A-Za-z0-9_-]{10,}"), "[REDACTED_OPENAI_KEY]"),
(re.compile(r"(?i)(Authorization:\s*Bearer\s+)[^\s\"']+"), r"\1[REDACTED]"),
(re.compile(r"(?i)(Authorization:\s*Basic\s+)[^\s\"']+"), r"\1[REDACTED]"),
(
re.compile(r"(?i)\b(api[_-]?key|token|password|secret)\b(\s*[:=]\s*)[^\s\"']+"),
r"\1\2[REDACTED]",
),
(
re.compile(r"(?i)\b(api[_-]?key|token|password|secret)\b(\s+)[^\s\"']+"),
r"\1\2[REDACTED]",
),
(
re.compile(
r"-----BEGIN [A-Z ]*PRIVATE KEY-----.*?-----END [A-Z ]*PRIVATE KEY-----",
re.DOTALL,
),
"[REDACTED_PRIVATE_KEY]",
),
)
def _payload(rec: Dict[str, Any]) -> Dict[str, Any]:
payload = rec.get("payload")
return payload if isinstance(payload, dict) else {}
def _timestamp(rec: Dict[str, Any], payload: Dict[str, Any]) -> str:
for value in (
payload.get("timestamp"),
rec.get("timestamp"),
payload.get("started_at"),
payload.get("completed_at"),
):
if isinstance(value, str) and value:
return value
return ""
def _text_from_any(content: Any) -> str:
if isinstance(content, str):
return content
if isinstance(content, list):
parts: List[str] = []
for item in content:
if isinstance(item, str):
parts.append(item)
elif isinstance(item, dict):
if item.get("type") == "text" and item.get("text"):
parts.append(str(item["text"]))
elif item.get("text"):
parts.append(str(item["text"]))
return "\n".join(parts)
if isinstance(content, dict):
if content.get("text"):
return str(content["text"])
if content.get("content"):
return _text_from_any(content["content"])
return ""
def _strip_codex_meta(text: str) -> str:
stripped = text.strip()
if not stripped:
return ""
if stripped.startswith("<codex_internal_context"):
return ""
if stripped.startswith("<environment_context"):
return ""
if stripped.startswith("# AGENTS.md instructions") or "--- project-doc ---" in stripped:
for marker in ("</environment_context>", "</INSTRUCTIONS>"):
idx = stripped.rfind(marker)
if idx == -1:
continue
tail = stripped[idx + len(marker):].strip()
if tail and not tail.startswith("<"):
return tail
return ""
return stripped
def _sanitize_text(text: str) -> str:
sanitized = _strip_codex_meta(text).replace("\x00", "").strip()
if not sanitized or _is_meta_prompt(sanitized):
return ""
for pattern, replacement in _SECRET_PATTERNS:
sanitized = pattern.sub(replacement, sanitized)
return sanitized
def _sanitize_tool_name(name: str) -> str:
return re.sub(r"[^A-Za-z0-9_.:-]+", "_", name)[:80]
def _tool_name(payload: Dict[str, Any]) -> str:
payload_type = payload.get("type")
name = payload.get("name")
if isinstance(name, str) and name:
return _sanitize_tool_name(name)
if payload_type == "exec_command_end":
return "exec_command"
if payload_type == "patch_apply_end":
return "apply_patch"
if payload_type == "web_search_call":
return "web_search"
if payload_type == "tool_search_call":
return "tool_search"
if isinstance(payload_type, str) and payload_type.endswith("_tool_call"):
return _sanitize_tool_name(payload_type)
return ""
def _dedup(xs: Iterable[str]) -> List[str]:
seen = set()
out: List[str] = []
for x in xs:
if x not in seen:
seen.add(x)
out.append(x)
return out
def digest_codex_archived_session(path: str, project: str = "") -> Optional[SessionDigest]:
"""Build a ``SessionDigest`` from one Codex Desktop archived session."""
session_id = os.path.splitext(os.path.basename(path))[0]
started = ""
ended = ""
session_project = ""
user_prompts: List[str] = []
assistant_finals: List[str] = []
tools: List[str] = []
feedback: List[str] = []
n_user = 0
n_asst = 0
for rec in _iter_jsonl(path):
payload = _payload(rec)
payload_type = payload.get("type")
ts = _timestamp(rec, payload)
if ts:
if not started:
started = ts
ended = ts
cwd = payload.get("cwd")
if isinstance(cwd, str) and cwd:
if not session_project:
session_project = cwd
if project and _project_matches(cwd, "invoked", project):
session_project = cwd
role = payload.get("role")
text = ""
output_role = ""
if payload_type == "user_message":
text = _text_from_any(payload.get("message"))
output_role = "user"
elif payload_type == "agent_message":
text = _text_from_any(payload.get("message"))
output_role = "assistant"
elif payload_type == "message" and role in {"user", "assistant"}:
text = _text_from_any(payload.get("content"))
output_role = str(role)
else:
tool = _tool_name(payload)
if tool:
tools.append(tool)
continue
sanitized = _sanitize_text(text)
if not sanitized:
continue
if output_role == "user":
n_user += 1
user_prompts.append(sanitized)
feedback.extend(_detect_feedback(sanitized))
elif output_role == "assistant":
n_asst += 1
assistant_finals.append(sanitized)
if project and not _project_matches(session_project or "", "invoked", project):
return None
if n_user == 0 and n_asst == 0:
return None
return SessionDigest(
session_id=session_id,
project=session_project,
started_at=started,
ended_at=ended,
user_prompts=user_prompts,
assistant_finals=assistant_finals[-5:],
tools_used=_dedup(tools),
files_touched=[],
feedback_signals=feedback,
n_user_turns=n_user,
n_assistant_turns=n_asst,
raw_path=path,
)
def harvest_codex(
archived_sessions_dir: str,
*,
scope: Any = "all",
invoked_project: str = "",
since_iso: Optional[str] = None,
limit: int = 0,
) -> List[SessionDigest]:
"""Walk ``~/.codex/archived_sessions`` and return matching digests."""
digests: List[SessionDigest] = []
if not os.path.isdir(archived_sessions_dir):
return digests
paths = [
os.path.join(archived_sessions_dir, fn)
for fn in os.listdir(archived_sessions_dir)
if fn.endswith(".jsonl")
]
paths.sort(key=lambda p: os.path.getmtime(p), reverse=True)
project_hint = invoked_project if scope == "invoked" else ""
for path in paths:
digest = digest_codex_archived_session(path, project=project_hint)
if digest is None:
continue
if not _project_matches(digest.project or "", 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
skillopt_sleep/harvest_sources.py
+41
View File
@@ -0,0 +1,41 @@
"""Source selection for SkillOpt-Sleep transcript harvesting."""
from __future__ import annotations
from typing import Optional
from skillopt_sleep.harvest import harvest
from skillopt_sleep.harvest_codex import harvest_codex
from skillopt_sleep.types import SessionDigest
def harvest_for_config(cfg, *, since_iso: Optional[str] = None, limit: int = 0) -> list[SessionDigest]:
source = cfg.get("transcript_source", "claude")
scope = cfg.get("projects", "invoked")
invoked_project = cfg.get("invoked_project", "")
if source == "codex":
return harvest_codex(
cfg.codex_archived_sessions_dir,
scope=scope,
invoked_project=invoked_project,
since_iso=since_iso,
limit=limit,
)
if source == "auto":
codex_digests = harvest_codex(
cfg.codex_archived_sessions_dir,
scope=scope,
invoked_project=invoked_project,
since_iso=since_iso,
limit=limit,
)
if codex_digests:
return codex_digests
return harvest(
cfg.transcripts_dir,
scope=scope,
invoked_project=invoked_project,
since_iso=since_iso,
limit=limit,
)
skillopt_sleep/memory.py
+9 -10
View File
@@ -12,7 +12,6 @@ from typing import List, Tuple
from skillopt_sleep.types import EditRecord
LEARNED_START = "<!-- SKILLOPT-SLEEP:LEARNED START -->"
LEARNED_END = "<!-- SKILLOPT-SLEEP:LEARNED END -->"
_BANNER = (
@@ -79,7 +78,7 @@ def apply_edits(doc: str, edits: List[EditRecord]) -> Tuple[str, List[EditRecord
anchor substring.
"""
lines = current_learned_lines(doc)
norm_set = {_norm(l) for l in lines}
norm_set = {_norm(line) for line in lines}
applied: List[EditRecord] = []
for e in edits:
@@ -92,31 +91,31 @@ def apply_edits(doc: str, edits: List[EditRecord]) -> Tuple[str, List[EditRecord
applied.append(e)
elif op == "delete":
anchor = _norm(e.anchor or e.content)
keep = [l for l in lines if anchor not in _norm(l)]
keep = [line for line in lines if anchor not in _norm(line)]
if len(keep) != len(lines):
lines = keep
norm_set = {_norm(l) for l in lines}
norm_set = {_norm(line) for line in lines}
applied.append(e)
elif op == "replace":
anchor = _norm(e.anchor)
new_lines = []
changed = False
for l in lines:
if anchor and anchor in _norm(l):
for line in lines:
if anchor and anchor in _norm(line):
new_lines.append(e.content.strip())
changed = True
else:
new_lines.append(l)
new_lines.append(line)
if changed:
lines = new_lines
norm_set = {_norm(l) for l in lines}
norm_set = {_norm(line) for line in lines}
applied.append(e)
return set_learned(doc, lines), applied
def ensure_skill_scaffold(doc: str, *, name: str, description: str) -> str:
"""Ensure a SKILL.md has YAML frontmatter so Claude Code loads it."""
"""Ensure a SKILL.md has YAML frontmatter so local agents load it."""
if doc.lstrip().startswith("---"):
return doc
fm = (
@@ -125,6 +124,6 @@ def ensure_skill_scaffold(doc: str, *, name: str, description: str) -> str:
f"description: {description}\n"
"---\n\n"
f"# {name}\n\n"
"Preferences and procedures learned from your past Claude Code sessions.\n"
"Preferences and procedures learned from your past local agent sessions.\n"
)
return fm + doc
skillopt_sleep/types.py
+5 -6
View File
@@ -8,18 +8,17 @@ external dependencies.
"""
from __future__ import annotations
from dataclasses import dataclass, field, asdict
from typing import Any, Dict, List, Optional
from dataclasses import asdict, dataclass, field
from typing import Any, Dict, List
# ── Stage 1: harvest ──────────────────────────────────────────────────────────
@dataclass
class SessionDigest:
"""A normalized summary of one Claude Code session transcript.
"""A normalized summary of one local agent session transcript.
Produced by :mod:`skillopt_sleep.harvest` from a ``<sessionId>.jsonl``
transcript plus ``history.jsonl`` entries.
Produced by source-specific harvesters from Claude Code transcripts or
Codex Desktop archived sessions.
"""
session_id: str