Exclude SkillOpt-generated Codex replay sessions from harvest

skillopt_sleep/harvest_codex.py

@@ -12,6 +12,7 @@
 
 from skillopt_sleep.harvest import (
     _detect_feedback,
+    _is_agent_session,
     _is_meta_prompt,
     _iter_jsonl,
     _project_matches,
@@ -20,6 +21,37 @@
 from skillopt_sleep.types import SessionDigest
 
 
+_CODEX_REPLAY_PREFIXES = (
+    "Complete the task. Apply the skill and memory rules exactly,",
+    "Complete the task. Apply the skill and memory rules EXACTLY,",
+)
+_CODEX_REPLAY_MARKERS = (
+    "## CURRENT SKILL",
+    "## FAILED TASKS",
+    "## SUCCESSFUL TASKS",
+    "You are a strict grader",
+    "## TASK\n",
+    "## SKILL\n",
+    "## Skill\n",
+)
+
+
+def _is_codex_replay(digest: SessionDigest) -> bool:
+    """Detect the prompt shape emitted by SkillOpt's Codex replay backend."""
+    if not digest.user_prompts:
+        return False
+    prompt = digest.user_prompts[0]
+    return (
+        any(marker in prompt for marker in _CODEX_REPLAY_MARKERS)
+        or (
+            any(prompt.startswith(prefix) for prefix in _CODEX_REPLAY_PREFIXES)
+            and "\n# Skill\n" in prompt
+            and "\n# Memory\n" in prompt
+            and "\n# Task\n" in prompt
+        )
+    )
+
+
 def _payload(rec: Dict[str, Any]) -> Dict[str, Any]:
     payload = rec.get("payload")
     return payload if isinstance(payload, dict) else {}
@@ -225,6 +257,8 @@ def harvest_codex(
             continue
         if not _project_matches(digest.project or "", scope, invoked_project):
             continue
+        if _is_agent_session(digest) or _is_codex_replay(digest):
+            continue
         if since_iso and digest.ended_at and digest.ended_at < since_iso:
             continue
         digests.append(digest)

tests/test_harvest_codex_replay.py

@@ -0,0 +1,53 @@
+"""Regression tests for excluding SkillOpt-generated Codex sessions."""
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+
+from skillopt_sleep.harvest_codex import harvest_codex
+
+
+def _write_session(path: str, prompt: str) -> None:
+    with open(path, "w", encoding="utf-8") as handle:
+        for record in (
+            {
+                "timestamp": "2026-09-20T00:00:00Z",
+                "payload": {"type": "turn_context", "cwd": "/repo"},
+            },
+            {
+                "timestamp": "2026-09-20T00:00:01Z",
+                "payload": {"type": "user_message", "message": prompt},
+            },
+            {
+                "timestamp": "2026-09-20T00:00:10Z",
+                "payload": {"type": "agent_message", "message": "Completed."},
+            },
+        ):
+            handle.write(json.dumps(record) + "\n")
+
+
+class TestCodexReplayHarvest(unittest.TestCase):
+    def test_skillopt_replay_session_is_excluded(self):
+        prompt = (
+            "Complete the task. Apply the skill and memory rules EXACTLY, including "
+            "any rule about searching before answering.\n\n"
+            "# Skill\nlearned rules\n\n# Memory\nprior notes\n\n"
+            "# Task\nanswer the task\n\nReturn ONLY the final answer."
+        )
+        with tempfile.TemporaryDirectory() as tmp:
+            _write_session(os.path.join(tmp, "replay.jsonl"), prompt)
+            self.assertEqual(harvest_codex(tmp, scope="all"), [])
+
+    def test_real_user_session_is_preserved(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            _write_session(os.path.join(tmp, "real.jsonl"), "Please update the parser.")
+            digests = harvest_codex(tmp, scope="all")
+
+        self.assertEqual(len(digests), 1)
+        self.assertEqual(digests[0].session_id, "real")
+
+
+if __name__ == "__main__":
+    unittest.main()