diff --git a/tests/test_recall_compaction_scope.py b/tests/test_recall_compaction_scope.py new file mode 100644 index 00000000..b8cc18e7 --- /dev/null +++ b/tests/test_recall_compaction_scope.py @@ -0,0 +1,185 @@ +"""Live-context exclusion for the model-facing recall tool. + +After a compaction the summary is a cache over the originals, not their +replacement — recall is the re-derivation path back into them. Scoping it: + +- ``get_compaction_checkpoint`` reads the latest persisted marker's watermark + (distinct from ``get_compaction_watermark``, which computes what a NEW + compaction would use). +- ``search_history(exclude_ws_id=…, exclude_after=…)`` drops the excluded + workstream's rows ABOVE the boundary — the live segment already in the + model's context — while rows at or below it (the summarized-away past) + stay searchable. ``exclude_after=None`` excludes the whole workstream: + never compacted means everything is live. +- ``_exec_recall`` passes its own workstream with a boundary read fresh at + execution time, and labels own-conversation hits so the model knows it is + re-reading its compacted past. +- The exclusion composes with the #745 tenancy scope, and the resume nudge + teaches the model the path exists. + +Other workstreams are untouched — recall remains the cross-conversation +search tool. The /history command deliberately has no exclusion: a human +browsing history has no "context" to duplicate. +""" + +from __future__ import annotations + +import json + +from tests._session_helpers import make_session +from turnstone.core.metacognition import NUDGE_COMPACTION_RESUME +from turnstone.core.session import COMPACTION_SOURCE + +NEEDLE = "quillfeather" + + +def _fill(st, ws: str, owner: str = "u1") -> list[int]: + """Register ``ws`` and write four searchable rows; return their ids.""" + st.register_workstream(ws, user_id=owner, title="t", kind="interactive") + return [st.save_message(ws, "user", f"{NEEDLE} row{i} in {ws}") for i in range(4)] + + +def _mark(st, ws: str, watermark: int | None, content: str = "SUMMARY") -> int: + """Write a compaction marker with ``watermark`` (None = malformed/legacy meta).""" + meta = json.dumps({"watermark": watermark}) if watermark is not None else None + return st.save_message(ws, "assistant", content, source=COMPACTION_SOURCE, meta=meta) + + +def _hits(st, **kwargs) -> set[str]: + return {r[3] for r in st.search_history(NEEDLE, limit=50, **kwargs)} + + +# --------------------------------------------------------------------------- +# get_compaction_checkpoint +# --------------------------------------------------------------------------- + + +class TestGetCompactionCheckpoint: + def test_none_when_never_compacted(self, storage_backend): + _fill(storage_backend, "ws1") + assert storage_backend.get_compaction_checkpoint("ws1") is None + + def test_reads_marker_watermark(self, storage_backend): + st = storage_backend + ids = _fill(st, "ws1") + _mark(st, "ws1", ids[1]) + assert st.get_compaction_checkpoint("ws1") == ids[1] + + def test_latest_marker_wins(self, storage_backend): + st = storage_backend + ids = _fill(st, "ws1") + _mark(st, "ws1", ids[0]) + _mark(st, "ws1", ids[2]) + assert st.get_compaction_checkpoint("ws1") == ids[2] + + def test_malformed_meta_reads_none(self, storage_backend): + """A legacy/corrupt marker must read as 'whole ws live' (exclude all), + never as a garbage boundary.""" + st = storage_backend + _fill(st, "ws1") + _mark(st, "ws1", None) + assert st.get_compaction_checkpoint("ws1") is None + + +# --------------------------------------------------------------------------- +# search_history live-context exclusion +# --------------------------------------------------------------------------- + + +class TestLiveContextExclusion: + def test_excludes_live_segment_keeps_compacted_past(self, storage_backend): + st = storage_backend + ids = _fill(st, "ws1") # rows 0..3 + boundary = ids[1] # rows 0-1 compacted away; 2-3 live + found = _hits(st, exclude_ws_id="ws1", exclude_after=boundary) + assert found == {f"{NEEDLE} row0 in ws1", f"{NEEDLE} row1 in ws1"} + + def test_never_compacted_ws_fully_excluded(self, storage_backend): + st = storage_backend + _fill(st, "ws1") + assert _hits(st, exclude_ws_id="ws1", exclude_after=None) == set() + + def test_other_workstreams_unaffected(self, storage_backend): + st = storage_backend + _fill(st, "ws1") + _fill(st, "ws2") + found = _hits(st, exclude_ws_id="ws1", exclude_after=None) + assert found == {f"{NEEDLE} row{i} in ws2" for i in range(4)} + + def test_no_exclusion_without_ws(self, storage_backend): + """The /history command path: no exclude args → everything searchable.""" + st = storage_backend + _fill(st, "ws1") + assert len(_hits(st)) == 4 + + def test_composes_with_tenancy_scope(self, storage_backend): + """Exclusion and the #745 private-project predicate BOTH drop rows in + one query: a mid-conversation boundary leaves ws_mine rows 2-3 live + (excluded) and 0-1 compacted (kept), while the tenancy predicate + hides dave's private-project row from carol — deleting either + fragment fails this test.""" + st = storage_backend + st.create_project("P", "P", owner_id="alice", visibility="private") + ids = _fill(st, "ws_mine", owner="alice") + st.register_workstream("ws_priv", user_id="dave", title="t", project_id="P") + st.save_message("ws_priv", "user", f"{NEEDLE} private row") + boundary = ids[1] # rows 0-1 compacted past; rows 2-3 live context + _mark(st, "ws_mine", boundary) + found = _hits(st, user_id="carol", exclude_ws_id="ws_mine", exclude_after=boundary) + assert found == {f"{NEEDLE} row0 in ws_mine", f"{NEEDLE} row1 in ws_mine"} + + +# --------------------------------------------------------------------------- +# _exec_recall plumbing + labeling +# --------------------------------------------------------------------------- + + +class TestRecallExecScope: + def _run_recall(self, session, rows, monkeypatch, checkpoint=7): + calls: dict = {} + + def fake_search_history(query, limit=20, offset=0, **kwargs): + calls.update(kwargs) + return rows + + monkeypatch.setattr("turnstone.core.session.search_history", fake_search_history) + monkeypatch.setattr( + "turnstone.core.session.get_compaction_checkpoint", lambda ws: checkpoint + ) + item = session._prepare_recall("c1", {"query": "x"}) + _, output = session._exec_recall(item) + return calls, output + + def test_passes_own_ws_and_fresh_boundary(self, monkeypatch): + session = make_session(user_id="owner") + session._ws_id = "ws-self" + calls, _ = self._run_recall(session, [], monkeypatch, checkpoint=42) + assert calls["exclude_ws_id"] == "ws-self" + assert calls["exclude_after"] == 42 + + def test_no_exclusion_without_registered_ws(self, monkeypatch): + session = make_session(user_id="owner") + session._ws_id = "" + calls, _ = self._run_recall(session, [], monkeypatch) + assert calls["exclude_ws_id"] is None + assert calls["exclude_after"] is None + + def test_own_conversation_hits_are_labeled(self, monkeypatch): + session = make_session(user_id="owner") + session._ws_id = "ws-self" + rows = [ + ("2026-07-02T10:00:00", "ws-self", "user", "old detail", None), + ("2026-07-02T11:00:00", "ws-other", "user", "other detail", None), + ] + _, output = self._run_recall(session, rows, monkeypatch) + own_line = next(line for line in output.splitlines() if "old detail" in line) + other_line = next(line for line in output.splitlines() if "other detail" in line) + assert "(earlier in this conversation, compacted)" in own_line + assert "(earlier in this conversation, compacted)" not in other_line + + +def test_resume_nudge_teaches_recall(): + """The model is told the summary is a digest and recall reaches the + compacted portion — the pointer that makes the instrumented form usable.""" + assert "recall tool" in NUDGE_COMPACTION_RESUME + assert "compacted portion" in NUDGE_COMPACTION_RESUME diff --git a/tests/test_search_history_visibility.py b/tests/test_search_history_visibility.py index 7e0e7b1a..d46f7074 100644 --- a/tests/test_search_history_visibility.py +++ b/tests/test_search_history_visibility.py @@ -163,7 +163,7 @@ class TestParityWithWsVisible: class TestRecallScopePlumbing: def _recorder(self, calls): - def fake_search_history(query, limit=20, offset=0, *, user_id=None): + def fake_search_history(query, limit=20, offset=0, *, user_id=None, **kwargs): calls.append(user_id) return [] diff --git a/turnstone/core/memory.py b/turnstone/core/memory.py index f499df00..6c47dab9 100644 --- a/turnstone/core/memory.py +++ b/turnstone/core/memory.py @@ -642,22 +642,52 @@ def update_workstream_title(ws_id: str, title: str) -> None: def search_history( - query: str, limit: int = 20, offset: int = 0, *, user_id: str | None = None + query: str, + limit: int = 20, + offset: int = 0, + *, + user_id: str | None = None, + exclude_ws_id: str | None = None, + exclude_after: int | None = None, ) -> list[Any]: """Search conversation history. ``user_id`` scopes rows by project tenancy (private-project workstreams hidden unless creator/owner/member — see :meth:`StorageBackend.search_history`); ``None`` = unscoped, for - single-user lanes only. + single-user lanes only. ``exclude_ws_id``/``exclude_after`` drop the + excluded workstream's live segment (rows above its compaction + checkpoint; the whole workstream when ``exclude_after`` is ``None``) — + the model-facing recall path passes its own ws so results never + duplicate what is already in context. """ try: - return get_storage().search_history(query, limit, offset, user_id=user_id) + return get_storage().search_history( + query, + limit, + offset, + user_id=user_id, + exclude_ws_id=exclude_ws_id, + exclude_after=exclude_after, + ) except Exception: log.warning("Failed to search history", exc_info=True) return [] +def get_compaction_checkpoint(ws_id: str) -> int | None: + """Latest persisted compaction marker's watermark for ``ws_id`` — see + :meth:`StorageBackend.get_compaction_checkpoint`. Returns ``None`` on any + storage error, which callers must read as "the whole workstream is live" + (recall then excludes it entirely — degraded to less information, never + to duplicated or leaked rows).""" + try: + return get_storage().get_compaction_checkpoint(ws_id) + except Exception: + log.warning("Failed to get compaction checkpoint for ws=%s", ws_id, exc_info=True) + return None + + def search_history_recent(limit: int = 20, *, user_id: str | None = None) -> list[Any]: """Return most recent conversation messages, tenancy-scoped like :func:`search_history`.""" diff --git a/turnstone/core/metacognition.py b/turnstone/core/metacognition.py index 7a00b38d..28488b4b 100644 --- a/turnstone/core/metacognition.py +++ b/turnstone/core/metacognition.py @@ -123,7 +123,9 @@ NUDGE_COMPACTION_RESUME = ( "The conversation was just compacted to free context. If there is remaining " "work, continue from the summary above — pick up the open tasks and next " "steps you recorded and keep going without waiting for further instructions. " - "If the task is already complete, give your final answer." + "The summary is a digest, not the record: if it is missing a detail you " + "need, the recall tool can search the compacted portion of this " + "conversation. If the task is already complete, give your final answer." ) _NUDGE_MAP: dict[str, str] = { diff --git a/turnstone/core/session.py b/turnstone/core/session.py index bbad47c0..9d9948d8 100644 --- a/turnstone/core/session.py +++ b/turnstone/core/session.py @@ -68,6 +68,7 @@ from turnstone.core.memory import ( delete_structured_memory_by_id, delete_workstream, get_attachments, + get_compaction_checkpoint, get_compaction_floor, get_compaction_watermark, get_skill_by_name, @@ -13576,13 +13577,35 @@ class ChatSession: return call_id, msg def _exec_recall(self, item: dict[str, Any]) -> tuple[str, str]: - """Search conversation history, scoped to the prepare-time user.""" + """Search conversation history, scoped to the prepare-time user. + + The session's own workstream is searchable only BELOW its compaction + checkpoint: rows above it are the live segment, already in context — + returning them would spend result slots on duplicates. Below it is + the summarized-away past, exactly what recall exists to re-derive + (the summary is a cache over the originals, not their replacement). + The boundary is read fresh at execution, not pinned at prepare, so a + compaction that ran while the item was queued is respected. + + Known limit: a FORKED session excludes only its own ws — the parent + rows it inherited into context remain searchable under the parent's + id (and unlabeled, since they aren't this ws). Harmless duplication + bounded by tenancy, and precise dedup needs a fork-time row cursor; + not worth carrying until forks matter here. + """ call_id = item["call_id"] query, limit, offset = item["query"], item["limit"], item.get("offset", 0) # KeyError on a missing pin is deliberate — an unpinned item must # fail loudly, not fall back to an unscoped (tenant-wide) search. - conv_rows = search_history(query, limit, offset, user_id=item["scope_user_id"]) + conv_rows = search_history( + query, + limit, + offset, + user_id=item["scope_user_id"], + exclude_ws_id=self._ws_id or None, + exclude_after=(get_compaction_checkpoint(self._ws_id) if self._ws_id else None), + ) if conv_rows: lines = [] for ts, sid, role, content, tool_name in conv_rows: @@ -13590,7 +13613,8 @@ class ChatSession: text = (content or "")[:2000] if content and len(content) > 2000: text += f"... ({len(content)} chars total)" - lines.append(f"[{ts} {sid}] {label}: {text}") + own = " (earlier in this conversation, compacted)" if sid == self._ws_id else "" + lines.append(f"[{ts} {sid}]{own} {label}: {text}") header = f"Conversations ({len(conv_rows)} matches" if offset: header += f", offset {offset}" diff --git a/turnstone/core/storage/_postgresql.py b/turnstone/core/storage/_postgresql.py index bdc2147f..51582725 100644 --- a/turnstone/core/storage/_postgresql.py +++ b/turnstone/core/storage/_postgresql.py @@ -80,6 +80,9 @@ from turnstone.core.storage._utils import ( from turnstone.core.storage._utils import ( HEURISTIC_RULE_MUTABLE as _HEURISTIC_RULE_MUTABLE, ) +from turnstone.core.storage._utils import ( + HISTORY_CONTEXT_EXCLUSION_SQL as _HISTORY_EXCL_SQL, +) from turnstone.core.storage._utils import ( HISTORY_VISIBILITY_SCOPE_SQL as _HISTORY_SCOPE_SQL, ) @@ -124,6 +127,7 @@ from turnstone.core.storage._utils import ( ) from turnstone.core.storage._utils import ( find_orphan_conversations, + parse_checkpoint_watermark, prepare_provider_data_for_save, purge_orphan_conversations, release_attachment_refs, @@ -547,6 +551,23 @@ class PostgreSQLBackend: ).scalar() return int(n or 0) + def get_compaction_checkpoint(self, ws_id: str) -> int | None: + """Latest persisted marker's watermark — see the protocol docstring. + ``None`` = never compacted / malformed meta (whole ws is live).""" + with self._conn() as conn: + row = conn.execute( + sa.select(conversations.c.meta) + .where( + sa.and_( + conversations.c.ws_id == ws_id, + conversations.c._source == _COMPACTION_SOURCE, + ) + ) + .order_by(conversations.c.id.desc()) + .limit(1) + ).fetchone() + return parse_checkpoint_watermark(row[0]) if row is not None else None + def delete_messages_after(self, ws_id: str, keep_count: int) -> int: with self._conn() as conn: cutoff_row = conn.execute( @@ -1203,18 +1224,32 @@ class PostgreSQLBackend: # -- Conversation search --------------------------------------------------- def search_history( - self, query: str, limit: int = 20, offset: int = 0, *, user_id: str | None = None + self, + query: str, + limit: int = 20, + offset: int = 0, + *, + user_id: str | None = None, + exclude_ws_id: str | None = None, + exclude_after: int | None = None, ) -> list[Any]: if not query or not query.strip(): return [] capped = min(int(limit), 100) capped_offset = max(0, int(offset)) - # Project-tenancy scope (see HISTORY_VISIBILITY_SCOPE_SQL): applied in + # Project-tenancy scope (see HISTORY_VISIBILITY_SCOPE_SQL) and the + # live-context exclusion (HISTORY_CONTEXT_EXCLUSION_SQL): applied in # SQL, not post-filtered in Python, so limit/offset pagination stays # honest — a page never silently shrinks because hidden rows were # fetched then dropped. scope_sql = _HISTORY_SCOPE_SQL if user_id is not None else "" - scope_params = {"scope_user": user_id} if user_id is not None else {} + scope_params: dict[str, Any] = {"scope_user": user_id} if user_id is not None else {} + if exclude_ws_id is not None: + scope_sql += _HISTORY_EXCL_SQL + # exclude_after=None → never compacted → the whole ws is live + # context; ids start at 1, so -1 excludes every row. + scope_params["excl_ws"] = exclude_ws_id + scope_params["excl_after"] = -1 if exclude_after is None else exclude_after with self._conn() as conn: # Use PostgreSQL full-text search if search_vector column exists try: diff --git a/turnstone/core/storage/_protocol.py b/turnstone/core/storage/_protocol.py index 366d4252..7ae91b40 100644 --- a/turnstone/core/storage/_protocol.py +++ b/turnstone/core/storage/_protocol.py @@ -278,6 +278,19 @@ class StorageBackend(Protocol): """ ... + def get_compaction_checkpoint(self, ws_id: str) -> int | None: + """The latest persisted compaction marker's watermark for ``ws_id``. + + Every row with ``id <=`` the returned boundary was folded into the + summary the live session now holds — the summarized-away past; rows + above it are the live segment still in the model's context. Distinct + from :meth:`get_compaction_watermark`, which computes the boundary a + NEW compaction would use; this reads the one already persisted. + ``None`` when the workstream never compacted or the marker's meta is + malformed (callers must then treat the WHOLE workstream as live). + """ + ... + # -- Workstream attachments (content-addressed, refcounted) --------------- # # Pending (uploaded-but-unsent) bytes live in the per-node in-memory @@ -778,7 +791,14 @@ class StorageBackend(Protocol): # -- Conversation search --------------------------------------------------- def search_history( - self, query: str, limit: int = 20, offset: int = 0, *, user_id: str | None = None + self, + query: str, + limit: int = 20, + offset: int = 0, + *, + user_id: str | None = None, + exclude_ws_id: str | None = None, + exclude_after: int | None = None, ) -> list[Any]: """Search conversation history. Returns (timestamp, ws_id, role, content, tool_name). @@ -791,6 +811,14 @@ class StorageBackend(Protocol): rule); ``tests/test_search_history_visibility.py`` pins the parity. ``None`` (default) applies no scoping — correct only for single-user lanes (local CLI); authenticated surfaces MUST pass the acting user. + + ``exclude_ws_id`` + ``exclude_after`` drop *exclude_ws_id*'s rows with + ``id > exclude_after`` — the live-context exclusion for the + model-facing recall tool (rows the model can already see; see + ``HISTORY_CONTEXT_EXCLUSION_SQL``). ``exclude_after=None`` with an + ``exclude_ws_id`` set excludes the entire workstream (never + compacted → all live). Both applied in SQL so pagination stays + honest. """ ... diff --git a/turnstone/core/storage/_sqlite.py b/turnstone/core/storage/_sqlite.py index a4aaa549..5dd1f1d8 100644 --- a/turnstone/core/storage/_sqlite.py +++ b/turnstone/core/storage/_sqlite.py @@ -80,6 +80,9 @@ from turnstone.core.storage._utils import ( from turnstone.core.storage._utils import ( HEURISTIC_RULE_MUTABLE as _HEURISTIC_RULE_MUTABLE, ) +from turnstone.core.storage._utils import ( + HISTORY_CONTEXT_EXCLUSION_SQL as _HISTORY_EXCL_SQL, +) from turnstone.core.storage._utils import ( HISTORY_VISIBILITY_SCOPE_SQL as _HISTORY_SCOPE_SQL, ) @@ -124,6 +127,7 @@ from turnstone.core.storage._utils import ( ) from turnstone.core.storage._utils import ( find_orphan_conversations, + parse_checkpoint_watermark, prepare_provider_data_for_save, purge_orphan_conversations, release_attachment_refs, @@ -626,6 +630,23 @@ class SQLiteBackend: ).scalar() return int(n or 0) + def get_compaction_checkpoint(self, ws_id: str) -> int | None: + """Latest persisted marker's watermark — see the protocol docstring. + ``None`` = never compacted / malformed meta (whole ws is live).""" + with self._conn() as conn: + row = conn.execute( + sa.select(conversations.c.meta) + .where( + sa.and_( + conversations.c.ws_id == ws_id, + conversations.c._source == _COMPACTION_SOURCE, + ) + ) + .order_by(conversations.c.id.desc()) + .limit(1) + ).fetchone() + return parse_checkpoint_watermark(row[0]) if row is not None else None + def delete_messages_after(self, ws_id: str, keep_count: int) -> int: with self._conn() as conn: # Find the id of the first row to delete (the row at offset keep_count) @@ -1378,18 +1399,32 @@ class SQLiteBackend: # -- Conversation search --------------------------------------------------- def search_history( - self, query: str, limit: int = 20, offset: int = 0, *, user_id: str | None = None + self, + query: str, + limit: int = 20, + offset: int = 0, + *, + user_id: str | None = None, + exclude_ws_id: str | None = None, + exclude_after: int | None = None, ) -> list[Any]: if not query or not query.strip(): return [] capped = min(int(limit), 100) capped_offset = max(0, int(offset)) - # Project-tenancy scope (see HISTORY_VISIBILITY_SCOPE_SQL): applied in + # Project-tenancy scope (see HISTORY_VISIBILITY_SCOPE_SQL) and the + # live-context exclusion (HISTORY_CONTEXT_EXCLUSION_SQL): applied in # SQL, not post-filtered in Python, so limit/offset pagination stays # honest — a page never silently shrinks because hidden rows were # fetched then dropped. scope_sql = _HISTORY_SCOPE_SQL if user_id is not None else "" - scope_params = {"scope_user": user_id} if user_id is not None else {} + scope_params: dict[str, Any] = {"scope_user": user_id} if user_id is not None else {} + if exclude_ws_id is not None: + scope_sql += _HISTORY_EXCL_SQL + # exclude_after=None → never compacted → the whole ws is live + # context; ids start at 1, so -1 excludes every row. + scope_params["excl_ws"] = exclude_ws_id + scope_params["excl_after"] = -1 if exclude_after is None else exclude_after with self._conn() as conn: if self._fts5_available: return list( diff --git a/turnstone/core/storage/_utils.py b/turnstone/core/storage/_utils.py index dcb3b2c7..341cb098 100644 --- a/turnstone/core/storage/_utils.py +++ b/turnstone/core/storage/_utils.py @@ -1034,23 +1034,44 @@ HISTORY_VISIBILITY_SCOPE_SQL = ( ") " ) +# Live-context exclusion for the model-facing recall tool: drop rows of ONE +# workstream (the caller's own) above its compaction checkpoint — those rows +# are the live segment, already in the model's context, and returning them +# wastes result slots on duplicates. Rows at or below the checkpoint are the +# summarized-away past: exactly what recall exists to re-derive. +# ``:excl_after`` = the checkpoint boundary, or ``-1`` for a never-compacted +# workstream — the whole conversation is live then, so the whole workstream +# is excluded. Human-facing surfaces (the /history command) deliberately do +# NOT apply this: a person browsing history has no "context" to duplicate. + +HISTORY_CONTEXT_EXCLUSION_SQL = "AND NOT (c.ws_id = :excl_ws AND c.id > :excl_after) " + def _is_compaction_marker(row: Any) -> bool: """True when a stored row is a compaction checkpoint marker (``_source`` = row index 7).""" return len(row) > 7 and row[7] == COMPACTION_SOURCE -def _compaction_watermark(row: Any) -> int | None: - """Read a marker row's checkpoint watermark from its ``meta`` column (row index 10). +def parse_checkpoint_watermark(meta_json: str | None) -> int | None: + """Parse a compaction marker's ``meta`` JSON into its watermark id. - Returns the boundary conversation id, or ``None`` for a marker that predates - the watermark field or whose meta is malformed (caller falls back to the full - transcript — never load *less* than is safe).""" - meta = _source_meta_from_json(row[10] if len(row) > 10 else None) + The single decoder for the checkpoint boundary — shared by the resume + slice (:func:`reconstruct_turns_checkpointed` via + :func:`_compaction_watermark`) and the backends' + ``get_compaction_checkpoint``. Returns ``None`` for a marker that + predates the watermark field or whose meta is malformed (callers fall + back to safe behavior: resume loads the full transcript, recall excludes + the whole workstream — never *less* safe than the honest answer).""" + meta = _source_meta_from_json(meta_json) wm = meta.get("watermark") if meta else None return wm if isinstance(wm, int) and not isinstance(wm, bool) else None +def _compaction_watermark(row: Any) -> int | None: + """Read a marker row's checkpoint watermark from its ``meta`` column (row index 10).""" + return parse_checkpoint_watermark(row[10] if len(row) > 10 else None) + + def reconstruct_turns_checkpointed( rows: list[Any], ws_id: str,