fix(sleep): key the harvest checkpoint by project and transcript source

docs/reference/cli.md

@@ -418,6 +418,11 @@ a new harvest checkpoint; subsequent runs use that checkpoint rather than the
 initial lookback. Use `harvest` or `dry-run` to verify counts before the first
 stateful run.
 
+There is one checkpoint per project **and transcript source**, because each
+source reads a different store with its own timestamps. Feeding one project from
+several sources therefore keeps a separate window per source, and a run of one
+source does not advance another source's window.
+
 The managed `schedule` command persists the project, backend, time, and optional
 auto-adopt setting only. It does not copy source, Cursor paths, model, or target
 skill flags into the scheduled command. Put `transcript_source`, `cursor_home`,

docs/sleep/README.md

@@ -337,7 +337,9 @@ adopt, or advance state. Use another backend for those tasks.
 The initial harvest window is 72 hours. Set `--lookback-hours N` explicitly when
 older sessions should be considered; `0` scans all history subject to the
 session limit. A stateful `run`, even with no mined tasks, advances the harvest
-checkpoint. Use `harvest` or `dry-run` to inspect counts first. A real-backend
+checkpoint for that project and transcript source — each source keeps its own
+window, so a run of one source never hides another source's sessions. Use
+`harvest` or `dry-run` to inspect counts first. A real-backend
 `dry-run` still incurs provider calls and spend, and session/task limits are not
 hard call, token, time, or monetary budgets.
 

skillopt_sleep/__main__.py

@@ -424,7 +424,7 @@ def _handoff_mine_and_pin(cfg, args, backend, snapshot: str, dry: bool):
                   file=sys.stderr)
             digests = None
     if digests is None:
-        since = state.last_harvest_for(project)
+        since = state.last_harvest_for(project, cfg.get("transcript_source", ""))
         lookback_hours = cfg.get("lookback_hours", 72)
         if since is None and lookback_hours and lookback_hours > 0:
             since = _now_iso(time.time() - lookback_hours * 3600)
@@ -476,7 +476,7 @@ def _handoff_mine_and_pin(cfg, args, backend, snapshot: str, dry: bool):
         if not dry:
             # Advance the harvest window like run_sleep_cycle's no-tasks
             # branch, or every later run re-scans the same stale window.
-            state.set_last_harvest(project, started)
+            state.set_last_harvest(project, started, cfg.get("transcript_source", ""))
             state.save()
         return 0, None
     payload = make_tasks_payload(

skillopt_sleep/cycle.py

@@ -788,13 +788,16 @@ def budget_exhausted() -> bool:
     )
 
     # ── 1+2. harvest + mine (unless seed_tasks injected) ─────────────────
+    # The harvest cursor is per (project, source): each source reads its own
+    # store, so one shared cursor would make a source skip its sessions (#312).
+    harvest_source = cfg.get("transcript_source", "")
     digests: List[SessionDigest] = []
     if seed_tasks is not None:
         tasks = seed_tasks
         n_sessions = 0
         _progress(cfg, f"using {len(tasks)} seeded tasks")
     else:
-        since = state.last_harvest_for(project)
+        since = state.last_harvest_for(project, harvest_source)
         # On first run (no prior harvest), apply lookback_hours so we don't
         # scan the entire transcript history and trigger massive LLM mining.
         if since is None:
@@ -889,7 +892,7 @@ def budget_exhausted() -> bool:
     if not tasks:
         report.ended_at = _now_iso(clock)
         report.notes.append("no tasks mined — nothing to consolidate")
-        state.set_last_harvest(project, started)
+        state.set_last_harvest(project, started, harvest_source)
         state.record_night({"night": night, "accepted": False, "n_tasks": 0})
         if not dry_run:
             state.save()
@@ -1176,7 +1179,7 @@ def budget_exhausted() -> bool:
                 )
         except Exception:
             pass
-        state.set_last_harvest(project, started)
+        state.set_last_harvest(project, started, harvest_source)
         night_scores = (
             {"baseline": result.baseline_score, "candidate": result.candidate_score}
             if scores_measured

skillopt_sleep/state.py

@@ -5,7 +5,8 @@
 short-term -> long-term transfer). It records:
 
   - night counter
-  - last harvest timestamp per project (so each night only sees new data)
+  - last harvest timestamp per project and transcript source (so each night
+    only sees new data from the store that source reads)
   - cross-night "slow/meta" memory (lessons that persisted across nights)
   - per-night history (scores, accept/reject) for trend reporting
 """
@@ -92,10 +93,28 @@ def _now_iso(clock: Optional[float] = None) -> str:
     return _t.strftime("%Y-%m-%dT%H:%M:%S", _t.localtime(clock if clock is not None else _t.time()))
 
 
+HARVEST_KEY_SEP = "::"
+
+
+def harvest_cursor_key(project: str, source: str = "") -> str:
+    """Key of one harvest checkpoint: one cursor per project AND source.
+
+    Every transcript source reads a different store with its own timestamps,
+    so a single project-wide cursor makes the second source of a project skip
+    its older, never-harvested sessions for good (#312).
+    """
+    project = str(project or "")
+    source = str(source or "")
+    return f"{project}{HARVEST_KEY_SEP}{source}" if source else project
+
+
 DEFAULT_STATE: Dict[str, Any] = {
     "version": 1,
     "night": 0,
-    "last_harvest": {},     # project -> iso timestamp of last harvested record
+    # "<project>::<transcript_source>" -> iso timestamp of last harvested
+    # record (bare "<project>" keys written before checkpoints were split per
+    # source are still read, see last_harvest_for)
+    "last_harvest": {},
     "slow_memory": "",      # cross-night consolidated lessons (meta-skill analogue)
     "history": [],          # list of per-night summaries
     "task_archive": [],     # capped list of past mined tasks (for associative recall)
@@ -139,11 +158,25 @@ def save(self) -> None:
     def night(self) -> int:
         return int(self.data.get("night", 0))
 
-    def last_harvest_for(self, project: str) -> Optional[str]:
-        return self.data.get("last_harvest", {}).get(project)
+    def last_harvest_for(self, project: str, source: str = "") -> Optional[str]:
+        """Last harvest time for ``project`` from transcript ``source``.
 
-    def set_last_harvest(self, project: str, iso_ts: str) -> None:
-        self.data.setdefault("last_harvest", {})[project] = iso_ts
+        A bare ``project`` key (written before checkpoints were split per
+        source) is used as the fallback, so an upgrade does not re-scan the
+        whole transcript history; it is never overwritten, because it is the
+        only cursor older state files have.
+        """
+        cursors = self.data.get("last_harvest", {})
+        if source:
+            key = harvest_cursor_key(project, source)
+            if key in cursors:
+                return cursors[key]
+        return cursors.get(str(project))
+
+    def set_last_harvest(self, project: str, iso_ts: str, source: str = "") -> None:
+        self.data.setdefault("last_harvest", {})[
+            harvest_cursor_key(project, source)
+        ] = iso_ts
 
     @property
     def slow_memory(self) -> str:

tests/test_handoff_backend.py

@@ -301,10 +301,13 @@ def test_run_with_no_tasks_exits_0_and_advances_harvest_window(self):
             ])
             self.assertEqual(rc, 0)
             # the no-tasks branch must persist last-harvest, otherwise every
-            # later run re-scans the same stale window forever
+            # later run re-scans the same stale window forever. The cursor is
+            # keyed by (project, transcript source) — #312.
             cfg = load_config(invoked_project=proj, claude_home=claude_home)
             state = SleepState.load(cfg.state_path)
-            self.assertIsNotNone(state.last_harvest_for(proj))
+            self.assertIsNotNone(
+                state.last_harvest_for(proj, cfg.get("transcript_source", ""))
+            )
 
 
 if __name__ == "__main__":

tests/test_harvest_checkpoint_source.py

@@ -0,0 +1,145 @@
+"""Harvest checkpoints are per (project, transcript source) — issue #312.
+
+``skillopt-sleep run`` used to store one harvest cursor per project, whatever
+``--source`` the run used. When one project is fed from several sources, every
+source after the first only saw sessions newer than the first run's start, so
+its older, never-harvested sessions were skipped for good.
+
+Offline: mock backend, no model calls, no network.
+Run:  python -m pytest tests/test_harvest_checkpoint_source.py
+"""
+from __future__ import annotations
+
+import json
+import os
+
+from skillopt_sleep.config import load_config
+from skillopt_sleep.state import SleepState, harvest_cursor_key
+
+PROJECT = "/work/proj"
+
+
+def _write_claude_session(transcripts_dir: str, session_id: str, project: str) -> None:
+    """Two-turn interactive session, ended a few minutes ago (inside lookback)."""
+    from datetime import datetime, timedelta
+
+    folder = os.path.join(transcripts_dir, "projects", "demo")
+    os.makedirs(folder, exist_ok=True)
+    path = os.path.join(folder, session_id + ".jsonl")
+    base = datetime.now() - timedelta(minutes=5)
+    with open(path, "w", encoding="utf-8") as f:
+        for i, (role, text) in enumerate((
+            ("user", "Fix the failing date parser test in module B."),
+            ("assistant", "Fixed."),
+            ("user", "Thanks, now add a regression test."),
+            ("assistant", "Added."),
+        )):
+            timestamp = (base + timedelta(minutes=i)).strftime("%Y-%m-%dT%H:%M:%S.000Z")
+            f.write(json.dumps({
+                "type": role, "cwd": project, "sessionId": session_id,
+                "timestamp": timestamp,
+                "message": {"role": role, "content": text},
+            }) + "\n")
+
+
+# ── state level ─────────────────────────────────────────────────────────────
+
+def test_cursors_are_keyed_per_project_and_source(tmp_path):
+    state = SleepState.load(str(tmp_path / "state.json"))
+    state.set_last_harvest(PROJECT, "2026-01-02T03:00:00", "codex")
+
+    assert state.last_harvest_for(PROJECT, "codex") == "2026-01-02T03:00:00"
+    # another source of the same project has no cursor of its own
+    assert state.last_harvest_for(PROJECT, "cursor") is None
+    # and no project-wide cursor was written for it to inherit
+    assert state.last_harvest_for(PROJECT) is None
+    assert state.data["last_harvest"] == {harvest_cursor_key(PROJECT, "codex"):
+                                          "2026-01-02T03:00:00"}
+
+
+def test_sources_do_not_share_one_cursor(tmp_path):
+    state = SleepState.load(str(tmp_path / "state.json"))
+    state.set_last_harvest(PROJECT, "2026-01-02T03:00:00", "codex")
+    state.set_last_harvest(PROJECT, "2026-01-05T03:00:00", "cursor")
+
+    assert state.last_harvest_for(PROJECT, "codex") == "2026-01-02T03:00:00"
+    assert state.last_harvest_for(PROJECT, "cursor") == "2026-01-05T03:00:00"
+
+
+def test_legacy_bare_project_cursor_is_still_read_and_kept(tmp_path):
+    # Older state files key the cursor by project alone; an upgrade must not
+    # re-scan the whole transcript history, so it is read as the fallback.
+    state = SleepState.load(str(tmp_path / "state.json"))
+    state.set_last_harvest(PROJECT, "2026-01-02T03:00:00")
+
+    assert state.last_harvest_for(PROJECT, "codex") == "2026-01-02T03:00:00"
+    assert state.last_harvest_for(PROJECT) == "2026-01-02T03:00:00"
+
+    # A per-source write must not overwrite the legacy cursor: other sources
+    # still have only that one to fall back on.
+    state.set_last_harvest(PROJECT, "2026-01-05T03:00:00", "codex")
+    assert state.last_harvest_for(PROJECT, "cursor") == "2026-01-02T03:00:00"
+    assert state.last_harvest_for(PROJECT, "codex") == "2026-01-05T03:00:00"
+    assert state.last_harvest_for(PROJECT) == "2026-01-02T03:00:00"
+
+
+# ── end to end (issue #312 repro) ───────────────────────────────────────────
+
+def _cfg(project: str, claude_home: str, cursor_home: str, source: str):
+    return load_config(
+        invoked_project=project, projects="all", backend="mock",
+        claude_home=claude_home, cursor_home=cursor_home,
+        transcript_source=source, lookback_hours=24,
+    )
+
+
+def test_second_source_run_does_not_hide_the_first_sources_sessions(tmp_path):
+    from skillopt_sleep.cycle import run_sleep_cycle
+
+    proj = str(tmp_path / "proj")
+    os.makedirs(proj)
+    claude_home = str(tmp_path / ".claude")
+    os.makedirs(claude_home)
+    # an empty Cursor store: that run harvests nothing but persists state
+    cursor_home = str(tmp_path / ".cursor")
+    os.makedirs(os.path.join(cursor_home, "projects"))
+    _write_claude_session(os.path.join(claude_home, "projects"), "real-1", proj)
+    _write_claude_session(os.path.join(claude_home, "projects"), "real-2", proj)
+
+    claude = _cfg(proj, claude_home, cursor_home, "claude")
+    cursor = _cfg(proj, claude_home, cursor_home, "cursor")
+
+    first = run_sleep_cycle(claude, dry_run=True)
+    assert first.report.n_sessions == 2
+
+    # stateful run of a DIFFERENT source in the same project: it persists a
+    # checkpoint for itself only (#312: this used to move the project cursor)
+    other = run_sleep_cycle(cursor, dry_run=False)
+    assert other.report.n_sessions == 0
+
+    # the claude sessions were never harvested, so they must still be there
+    again = run_sleep_cycle(claude, dry_run=True)
+    assert again.report.n_sessions == 2
+
+    state = SleepState.load(claude.state_path)
+    assert state.last_harvest_for(proj, "cursor") is not None
+    assert state.last_harvest_for(proj, "claude") is None
+
+
+def test_a_source_still_skips_what_it_already_harvested(tmp_path):
+    from skillopt_sleep.cycle import run_sleep_cycle
+
+    proj = str(tmp_path / "proj")
+    os.makedirs(proj)
+    claude_home = str(tmp_path / ".claude")
+    os.makedirs(claude_home)
+    cursor_home = str(tmp_path / ".cursor")
+    os.makedirs(os.path.join(cursor_home, "projects"))
+    _write_claude_session(os.path.join(claude_home, "projects"), "real-1", proj)
+
+    claude = _cfg(proj, claude_home, cursor_home, "claude")
+    first = run_sleep_cycle(claude, dry_run=False)
+    assert first.report.n_sessions == 1
+
+    # its own cursor advanced: the same session is not harvested twice
+    assert run_sleep_cycle(claude, dry_run=True).report.n_sessions == 0