diff --git a/docs/reference/cli.md b/docs/reference/cli.md index 1cf93f893..fdf8babf8 100644 --- a/docs/reference/cli.md +++ b/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`, diff --git a/docs/sleep/README.md b/docs/sleep/README.md index 08c0429fd..488b97919 100644 --- a/docs/sleep/README.md +++ b/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. diff --git a/skillopt_sleep/__main__.py b/skillopt_sleep/__main__.py index df1b4ae19..eebe5d457 100644 --- a/skillopt_sleep/__main__.py +++ b/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( diff --git a/skillopt_sleep/cycle.py b/skillopt_sleep/cycle.py index 4eb2bc813..9621694c2 100644 --- a/skillopt_sleep/cycle.py +++ b/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 diff --git a/skillopt_sleep/state.py b/skillopt_sleep/state.py index bb5f39bbe..ae8536be4 100644 --- a/skillopt_sleep/state.py +++ b/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 + # "::" -> iso timestamp of last harvested + # record (bare "" 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: diff --git a/tests/test_handoff_backend.py b/tests/test_handoff_backend.py index b4b850228..6f1d1a9e3 100644 --- a/tests/test_handoff_backend.py +++ b/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__": diff --git a/tests/test_harvest_checkpoint_source.py b/tests/test_harvest_checkpoint_source.py new file mode 100644 index 000000000..6df852f3d --- /dev/null +++ b/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