diff --git a/docs/decisions/2026-07-31-metadocs-living-project-memory.md b/docs/decisions/2026-07-31-metadocs-living-project-memory.md index 62f4d52..08c73e4 100644 --- a/docs/decisions/2026-07-31-metadocs-living-project-memory.md +++ b/docs/decisions/2026-07-31-metadocs-living-project-memory.md @@ -82,3 +82,21 @@ share-composer: `claude -p --tools "" --strict-mcp-config`, промпт на ar - **Свой демон-планировщик** — launchd уже умеет догонять проспанные запуски. 2026-07-31 · PR: feat/metadocs + +## Пересмотр 2026-08-02: подача по сессиям, бюджет прогона отменён + +Максим: «давал бы ему обновления по сессии — он сам разберётся; бюджет не +нужен». Он прав: 40K-бюджет лечил симптом — таймаут первого прогона случился +из-за упаковки нескольких сессий в один вызов, а не из-за размера сессий. + +Теперь единица работы — **сессия** (её непрочитанный хвост по watermark), от +старой к новой, один вызов на сессию. Фейл сессии останавливает проект до +следующего прогона: поздние сессии продолжают истории ранних, дистиллировать +их вне порядка — записать конец раньше начала. Коммит — на проект. Прогон +без spillover: один `run` переваривает весь накопившийся бэклог, прерывание +безопасно (watermark двигается по главам). + +Единственный размерный порог остался на уровне одного **вызова**: +сессия-марафон режется на последовательные главы (`MAX_CALL_CHARS`, 60K), +поздние главы видят документы, обновлённые ранними. Это не бюджет прогона, +а честный предел одного обращения к модели. diff --git a/src/session_recall/metadocs/collect.py b/src/session_recall/metadocs/collect.py index 61ff4dc..e953c02 100644 --- a/src/session_recall/metadocs/collect.py +++ b/src/session_recall/metadocs/collect.py @@ -1,4 +1,13 @@ -"""Pull the new dialogue out of the index — user words and final answers only. +"""Pull the new dialogue out of the index — user words and final answers only, +handed over SESSION BY SESSION. + +The session is the unit of work on purpose: one session is one work arc (a bug +hunt, a feature, a decision), and the per-session watermark makes it the +natural increment — a run gives the distiller each session's unseen tail, +oldest session first, until nothing is pending. There is no run-level budget: +runs take as long as the backlog demands. The only size threshold lives at the +level of ONE model call — a marathon session is split into sequential chapters, +because a single call has practical limits even though the run does not. Reads the existing index DB directly and read-only: the indexer already stores exactly the surface layer meta docs wants (``user`` messages and the @@ -14,25 +23,16 @@ from .config import PROJECT_ALL, PROJECT_GIT, Watermarks -# One run reads at most this much text per project. Whole sessions are taken -# oldest-first until the budget is hit; the rest simply waits for the next run -# (watermarks only advance over what was actually taken), so a giant day is -# processed across several runs instead of overflowing one model call. -# 40K measured as the practical ceiling: distilling one batch into several -# complete documents already takes minutes; bigger batches hit the timeout. -BUDGET_CHARS = 40_000 -MAX_TURN_CHARS = 4_000 # one pasted log must not eat the whole budget +MAX_CALL_CHARS = 60_000 # per-CALL ceiling: chapter split for marathon sessions +MAX_TURN_CHARS = 4_000 # one pasted log must not eat a whole chapter @dataclass -class ProjectBatch: - project: str - turns: list = field(default_factory=list) # {source, session_id, role, text, ts} - spillover: bool = False # true when the budget cut sessions off this run - - @property - def chars(self) -> int: - return sum(len(t["text"]) for t in self.turns) +class SessionUpdate: + """One session's dialogue the distiller has not seen yet.""" + source: str + session_id: str + turns: list = field(default_factory=list) # {source, session_id, role, text, ts} def open_index(db_path: Path) -> sqlite3.Connection: @@ -70,38 +70,49 @@ def select_projects(db: sqlite3.Connection, selector: list, return sorted(p for p, _ in rows if p in wanted) -def new_dialogue(db: sqlite3.Connection, project: str, - marks: Watermarks) -> ProjectBatch: - """Everything said in this project since the last run, oldest first, - grouped so a session is either taken whole or left whole for next time.""" +def pending_sessions(db: sqlite3.Connection, project: str, + marks: Watermarks) -> list[SessionUpdate]: + """Every session with dialogue newer than its watermark, oldest first. + A late-indexed old session has no mark and is picked up whole; an appended + session contributes only its tail — «обновления по сессии».""" rows = db.execute( "SELECT source, session_id, role, text, ts FROM chunks " "WHERE project = ? AND role IN ('user', 'assistant') " "ORDER BY ts, turn_index", (project,)).fetchall() - sessions: dict[str, list] = {} + per: dict[str, SessionUpdate] = {} for source, sid, role, text, ts in rows: if not text or not text.strip(): continue ts = int(ts or 0) if ts <= marks.last_ts(source, sid): continue - sessions.setdefault(f"{source}:{sid}", []).append( - {"source": source, "session_id": sid, "role": role, - "text": text[:MAX_TURN_CHARS], "ts": ts}) - - batch = ProjectBatch(project=project) - for _, turns in sorted(sessions.items(), key=lambda kv: kv[1][0]["ts"]): - size = sum(len(t["text"]) for t in turns) - if batch.turns and batch.chars + size > BUDGET_CHARS: - batch.spillover = True - break - batch.turns.extend(turns) - return batch - - -def advance_marks(marks: Watermarks, batch: ProjectBatch) -> None: - """Only after the batch was distilled AND written — a crash in between + upd = per.setdefault(f"{source}:{sid}", + SessionUpdate(source=source, session_id=sid)) + upd.turns.append({"source": source, "session_id": sid, "role": role, + "text": text[:MAX_TURN_CHARS], "ts": ts}) + return sorted(per.values(), key=lambda u: u.turns[0]["ts"]) + + +def chapters(turns: list, ceiling: int | None = None) -> list[list]: + """A session almost always fits one call. A marathon session is split into + sequential chapters, in order; each later chapter's call sees the documents + the earlier chapters already updated, so the story stays continuous.""" + ceiling = MAX_CALL_CHARS if ceiling is None else ceiling + out, cur, size = [], [], 0 + for t in turns: + if cur and size + len(t["text"]) > ceiling: + out.append(cur) + cur, size = [], 0 + cur.append(t) + size += len(t["text"]) + if cur: + out.append(cur) + return out + + +def advance_marks(marks: Watermarks, turns: list) -> None: + """Only after these turns were distilled AND written — a crash in between must re-process, never silently skip.""" - for t in batch.turns: + for t in turns: marks.advance(t["source"], t["session_id"], t["ts"]) diff --git a/src/session_recall/metadocs/distill.py b/src/session_recall/metadocs/distill.py index b90dffd..1ac9a1a 100644 --- a/src/session_recall/metadocs/distill.py +++ b/src/session_recall/metadocs/distill.py @@ -36,7 +36,8 @@ _SYSTEM = """\ You maintain a project's living memory: a few markdown documents distilled \ from what was said in work sessions. You receive the current documents and the \ -new dialogue since the last run. Return updated documents. +new dialogue of ONE work session (or one chapter of a long session). Return \ +updated documents. Track exactly these entities: @@ -54,9 +55,11 @@ ONLY — never copy the stored values themselves into the map. Rules: -- Update existing entries instead of duplicating them; merge when the new \ -dialogue continues an old story. Keep entries that the new dialogue does not \ -touch. +- Before adding ANYTHING, search the current documents for an entry about the \ +same bug, action, decision, or storage location — and update or extend that \ +entry instead of adding a twin. A new entry is for a genuinely new story only; \ +merge when the new dialogue continues an old one. Keep entries the new \ +dialogue does not touch. - Every entry ends with a `sources:` line listing session ids it came from. \ Cite artifacts that exist outside this machine (PR numbers, commits, paths) \ so an entry can be checked without the transcripts. diff --git a/src/session_recall/metadocs/run.py b/src/session_recall/metadocs/run.py index d1eefd9..e044ad5 100644 --- a/src/session_recall/metadocs/run.py +++ b/src/session_recall/metadocs/run.py @@ -1,9 +1,13 @@ -"""One meta docs run: collect → distill → scan → write → commit → advance. +"""One meta docs run: for every project, feed the distiller its pending +sessions one at a time — распаковка «обновлений по сессии». -Order is the safety property. Watermarks move only after the batch's documents -are safely in the repo, so a crash or a garbled model answer re-processes the -same dialogue next run instead of silently swallowing it. A project whose -distillation fails is skipped without advancing; the others still run. +Order is the safety property, twice over. Watermarks move only after a +chapter's documents are safely in the repo, so a crash or a garbled model +answer re-processes the same dialogue next run instead of silently swallowing +it. And sessions are strictly oldest-first: when one session fails, the +project HALTS for this run — later sessions build on earlier stories, and +distilling them out of order would write the ending before the beginning. +Other projects still run; each project that changed gets its own commit. """ from dataclasses import dataclass, field @@ -17,25 +21,24 @@ @dataclass class RunReport: - projects: list = field(default_factory=list) # (project, files_written, note) + projects: list = field(default_factory=list) # (project, done, pending, note) blocked: list = field(default_factory=list) # (path, kinds) scanner stops - committed: str = "" - spillover: bool = False # some dialogue waits for the next run + commits: list = field(default_factory=list) # (project, short sha) def summary(self) -> str: if not self.projects and not self.blocked: return "meta docs: nothing new" lines = [] - for project, files, note in self.projects: - what = ", ".join(files) if files else "no doc changes" - lines.append(f"{project}: {what}{f' ({note})' if note else ''}") + for project, done, pending, note in self.projects: + line = f"{project}: {done}/{pending} session(s)" + if note: + line += f" — {note}" + lines.append(line) for path, kinds in self.blocked: lines.append(f"BLOCKED by secret scanner: {path} [{', '.join(kinds)}] " "— fix the source, the entry will retry next run") - if self.committed: - lines.append(f"committed {self.committed}") - if self.spillover: - lines.append("more dialogue than one run's budget — the rest runs next time") + for project, sha in self.commits: + lines.append(f"committed {sha} ({project})") return "\n".join(lines) @@ -47,27 +50,32 @@ def run_once(cfg: MetaConfig, db, distiller: Distiller, report = RunReport() for project in collect.select_projects(db, cfg.projects): - batch = collect.new_dialogue(db, project, marks) - if not batch.turns: + sessions = collect.pending_sessions(db, project, marks) + if not sessions: continue - report.spillover = report.spillover or batch.spillover - updates = distiller(project, batch.turns, current_docs(repo, project)) - if updates is None: - # engine down or unparseable output: do not advance, try again later - report.projects.append((project, [], "distill failed, will retry")) - continue - wr = write_docs(repo, project, updates) - report.blocked.extend(wr.blocked) - if wr.blocked: - # something in this batch trips the scanner; keep the marks put so - # a human can fix the doc source and the next run reprocesses - report.projects.append((project, wr.written, "partially blocked")) - continue - collect.advance_marks(marks, batch) - marks.save() - report.projects.append((project, wr.written, "")) - - if any(files for _, files, _ in report.projects): - report.committed = commit( - repo, "meta docs: distill new sessions", push=cfg.push) + done, note, touched = 0, "", False + for session in sessions: + ok = True + for chapter in collect.chapters(session.turns): + updates = distiller(project, chapter, current_docs(repo, project)) + if updates is None: + note, ok = "distill failed, will retry", False + break + wr = write_docs(repo, project, updates) + report.blocked.extend(wr.blocked) + if wr.blocked: + note, ok = "blocked by secret scanner", False + break + touched = touched or bool(wr.written) + collect.advance_marks(marks, chapter) + marks.save() + if not ok: + break # story order: later sessions wait for the retry + done += 1 + if touched: + sha = commit(repo, f"meta docs: {project} — {done} session(s)", + push=cfg.push) + if sha: + report.commits.append((project, sha)) + report.projects.append((project, done, len(sessions), note)) return report diff --git a/tests/test_metadocs.py b/tests/test_metadocs.py index 6860a3e..88243e4 100644 --- a/tests/test_metadocs.py +++ b/tests/test_metadocs.py @@ -53,33 +53,41 @@ def repo(tmp_path): def test_collect_only_dialogue_roles(db, marks): add_turn(db, "p", "user", "как чинили CI?", 100) add_turn(db, "p", "assistant", "запинили mcp<2", 101) - batch = collect.new_dialogue(db, "p", marks) - assert [t["role"] for t in batch.turns] == ["user", "assistant"] + (upd,) = collect.pending_sessions(db, "p", marks) + assert [t["role"] for t in upd.turns] == ["user", "assistant"] -def test_collect_respects_watermark(db, marks): +def test_collect_session_tail_after_watermark(db, marks): + """«Обновления по сессии»: an appended session contributes only its tail.""" add_turn(db, "p", "user", "old", 100) add_turn(db, "p", "user", "new", 200) marks.advance("claude", "sess-1", 100) - batch = collect.new_dialogue(db, "p", marks) - assert [t["text"] for t in batch.turns] == ["new"] + (upd,) = collect.pending_sessions(db, "p", marks) + assert [t["text"] for t in upd.turns] == ["new"] def test_collect_late_indexed_session_still_taken(db, marks): """An OLD session indexed for the first time has no mark → taken whole.""" marks.advance("claude", "sess-1", 500) add_turn(db, "p", "user", "ancient but never distilled", 100, sid="sess-2") - batch = collect.new_dialogue(db, "p", marks) - assert len(batch.turns) == 1 + (upd,) = collect.pending_sessions(db, "p", marks) + assert upd.session_id == "sess-2" and len(upd.turns) == 1 -def test_collect_budget_takes_whole_sessions(db, marks, monkeypatch): - monkeypatch.setattr(collect, "BUDGET_CHARS", 100) - add_turn(db, "p", "user", "a" * 80, 100, sid="s1") - add_turn(db, "p", "user", "b" * 80, 200, sid="s2") - batch = collect.new_dialogue(db, "p", marks) - assert batch.spillover is True - assert {t["session_id"] for t in batch.turns} == {"s1"} # s2 waits, whole +def test_collect_sessions_ordered_oldest_first(db, marks): + add_turn(db, "p", "user", "later story", 300, sid="s2") + add_turn(db, "p", "user", "earlier story", 100, sid="s1") + got = collect.pending_sessions(db, "p", marks) + assert [u.session_id for u in got] == ["s1", "s2"] + + +def test_chapters_split_only_marathon_sessions(): + small = [{"text": "x" * 10, "ts": i} for i in range(3)] + assert len(collect.chapters(small, ceiling=100)) == 1 + big = [{"text": "x" * 40, "ts": i} for i in range(5)] + parts = collect.chapters(big, ceiling=100) + assert len(parts) == 3 # 2+2+1, order preserved + assert [t["ts"] for p in parts for t in p] == [0, 1, 2, 3, 4] def test_select_projects_git_probes_cwd(db, marks): @@ -121,6 +129,13 @@ def test_prompt_marks_dialogue_as_data(): "never copy the stored values themselves" in distill._SYSTEM +def test_prompt_demands_update_before_add(): + """Maxim's rule: актуализировать — look for an existing entry first, + adding a twin is the failure mode.""" + assert "Before adding ANYTHING" in distill._SYSTEM + assert "instead of adding a twin" in distill._SYSTEM + + def test_cli_distiller_runs_in_empty_cwd_no_tools(): seen = {} @@ -159,18 +174,16 @@ def test_commit_only_on_change(repo): # -- the whole run ------------------------------------------------------------ -def _world(db, tmp_path, repo): +def _world(db, tmp_path, repo, monkeypatch): + monkeypatch.setattr("session_recall.metadocs.run.state_path", + lambda d=None: tmp_path / "st.json") add_turn(db, "proj", "user", "почему выбрали пин mcp<2?", 100) add_turn(db, "proj", "assistant", "потому что 2.0 удалил fastmcp, PR #13", 101) return MetaConfig(repo=str(repo), projects=[PROJECT_ALL]) def test_run_distills_writes_commits_advances(db, tmp_path, repo, monkeypatch): - monkeypatch.setattr("session_recall.metadocs.config.state_path", - lambda d=None: tmp_path / "st.json") - monkeypatch.setattr("session_recall.metadocs.run.state_path", - lambda d=None: tmp_path / "st.json") - cfg = _world(db, tmp_path, repo) + cfg = _world(db, tmp_path, repo, monkeypatch) calls = [] def distiller(project, turns, docs): @@ -178,34 +191,72 @@ def distiller(project, turns, docs): return {"decisions.md": "## пин mcp<2\nпочему: fastmcp удалили. sources: sess-1\n"} report = run_once(cfg, db, distiller) - assert calls == [("proj", 2)] - assert report.committed + assert calls == [("proj", 2)] # one session → one call + assert report.commits and report.commits[0][0] == "proj" + assert report.projects == [("proj", 1, 1, "")] assert (repo / "proj" / "decisions.md").exists() # second run: watermark advanced, nothing new, no commit report2 = run_once(cfg, db, distiller) - assert report2.committed == "" and len(calls) == 1 + assert report2.commits == [] and len(calls) == 1 -def test_run_failed_distill_does_not_advance(db, tmp_path, repo, monkeypatch): - monkeypatch.setattr("session_recall.metadocs.run.state_path", - lambda d=None: tmp_path / "st.json") - cfg = _world(db, tmp_path, repo) +def test_run_one_call_per_session(db, tmp_path, repo, monkeypatch): + cfg = _world(db, tmp_path, repo, monkeypatch) + add_turn(db, "proj", "user", "другая история", 300, sid="sess-2") + seen = [] + distiller = lambda p, t, d: seen.append({x["session_id"] for x in t}) or {} + report = run_once(cfg, db, distiller) + assert seen == [{"sess-1"}, {"sess-2"}] # never mixed in one call + assert report.projects == [("proj", 2, 2, "")] + + +def test_run_marathon_session_processed_in_chapters(db, tmp_path, repo, monkeypatch): + cfg = _world(db, tmp_path, repo, monkeypatch) + monkeypatch.setattr(collect, "MAX_CALL_CHARS", 50) + add_turn(db, "proj", "user", "x" * 40, 102) + add_turn(db, "proj", "user", "y" * 40, 103) + calls = [] + distiller = lambda p, t, d: calls.append(len(t)) or {} + report = run_once(cfg, db, distiller) + assert len(calls) >= 2 # split, but all of it processed + assert report.projects == [("proj", 1, 1, "")] + marks = Watermarks(tmp_path / "st.json") + assert marks.last_ts("claude", "sess-1") == 103 + + +def test_run_failed_session_halts_project_in_order(db, tmp_path, repo, monkeypatch): + """Later sessions build on earlier stories: a failure stops the project so + nothing is distilled out of order, and the watermark stays put.""" + cfg = _world(db, tmp_path, repo, monkeypatch) + add_turn(db, "proj", "user", "продолжение истории", 300, sid="sess-2") report = run_once(cfg, db, lambda p, t, d: None) - assert report.committed == "" + assert report.commits == [] + assert report.projects == [("proj", 0, 2, "distill failed, will retry")] marks = Watermarks(tmp_path / "st.json") - assert marks.last_ts("claude", "sess-1") == 0 # retry next run + assert marks.last_ts("claude", "sess-1") == 0 + assert marks.last_ts("claude", "sess-2") == 0 # never attempted def test_run_scanner_block_freezes_watermark(db, tmp_path, repo, monkeypatch): - monkeypatch.setattr("session_recall.metadocs.run.state_path", - lambda d=None: tmp_path / "st.json") - cfg = _world(db, tmp_path, repo) + cfg = _world(db, tmp_path, repo, monkeypatch) leaky = {"bugs.md": "fix used AKIAIOSFODNN7EXAMPLE\n"} report = run_once(cfg, db, lambda p, t, d: leaky) assert report.blocked assert Watermarks(tmp_path / "st.json").last_ts("claude", "sess-1") == 0 +def test_run_commit_per_project(db, tmp_path, repo, monkeypatch): + cfg = _world(db, tmp_path, repo, monkeypatch) + add_turn(db, "other", "user", "второй проект", 100, sid="sess-9") + distiller = lambda p, t, d: {"bugs.md": f"# {p}\n"} + report = run_once(cfg, db, distiller) + assert [c[0] for c in report.commits] == ["other", "proj"] + log = subprocess.run(["git", "-C", str(repo), "log", "--format=%s"], + capture_output=True, text=True).stdout + assert "meta docs: other — 1 session(s)" in log + assert "meta docs: proj — 1 session(s)" in log + + # -- schedule ----------------------------------------------------------------- def test_plist_shape(tmp_path): plist = schedule.build_plist("21:30", tmp_path / "m.log")