From ffadb69694bfe6857135d061fc5d1d3cdaa6c07b Mon Sep 17 00:00:00 2001 From: jheronimus Date: Mon, 13 Jul 2026 18:31:34 +0300 Subject: [PATCH 1/2] fix: make session index append-only to prevent concurrent-write data loss MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _upsert read the entire index, filtered out the record, and rewrote the whole file. Two interleaved calls (two Tau processes in the same folder) both read the same state, both append their record, and the second write_text overwrites the first — silently dropping the loser's entry. Switch _upsert to append-only: one write() syscall per record, no truncation, no rewrite. Duplicate lines for the same session ID (from touch_session updates) are already resolved at read time by _deduplicate_records (last writer wins), so read-side semantics are unchanged. _write_index is removed as dead code. --- src/tau_coding/session_manager.py | 22 +++++---- tests/test_session_manager.py | 77 +++++++++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 10 deletions(-) diff --git a/src/tau_coding/session_manager.py b/src/tau_coding/session_manager.py index 1cccf2296..860d926aa 100644 --- a/src/tau_coding/session_manager.py +++ b/src/tau_coding/session_manager.py @@ -234,18 +234,20 @@ def _read_all_records(self) -> list[CodingSessionRecord]: records.extend(self._read_index(index_path)) return _deduplicate_records(records) - def _write_index(self, path: Path, records: list[CodingSessionRecord]) -> None: - path.parent.mkdir(parents=True, exist_ok=True) - content = "\n".join(record.to_model().model_dump_json() for record in records) - if content: - content += "\n" - path.write_text(content, encoding="utf-8") - def _upsert(self, record: CodingSessionRecord) -> None: + """Append a session record to the project index. + + Append-only — never truncates, never rewrites — so a crash during + write cannot lose previously indexed sessions, and concurrent writers + cannot overwrite each other. Duplicate entries for the same session + ID are resolved at read time by ``_deduplicate_records`` (last writer + wins). + """ path = self.project_index_path(record.cwd) - records = [item for item in self._read_index(path) if item.id != record.id] - records.append(record) - self._write_index(path, records) + path.parent.mkdir(parents=True, exist_ok=True) + line = record.to_model().model_dump_json() + "\n" + with open(path, "a", encoding="utf-8") as f: + f.write(line) def _deduplicate_records(records: list[CodingSessionRecord]) -> list[CodingSessionRecord]: diff --git a/tests/test_session_manager.py b/tests/test_session_manager.py index 491944c59..685d80fa4 100644 --- a/tests/test_session_manager.py +++ b/tests/test_session_manager.py @@ -161,3 +161,80 @@ def test_session_manager_sorts_newest_updated_first(tmp_path: Path) -> None: assert [session.id for session in sessions] == ["older", "newer"] assert newer in sessions + + +def test_concurrent_upserts_dont_lose_sessions(tmp_path: Path) -> None: + """Two interleaved _upsert calls must not lose either record. + + The old read-modify-write _upsert truncated index.jsonl and rewrote it. + Two threads that read before either writes lose one record. Uses a + barrier inside a monkey-patched _read_index to force both threads into + the read before either writes. + """ + import threading + + manager = SessionManager(TauPaths(home=tmp_path / ".tau", agents_home=tmp_path / ".agents")) + cwd = tmp_path / "project" + cwd.mkdir() + + manager.create_session(cwd=cwd, model="existing", session_id="existing") + + read_barrier = threading.Barrier(2, timeout=5) + original_read = manager._read_index + + def patched_read(path: Path) -> list: + read_barrier.wait() + return original_read(path) + + manager._read_index = patched_read + + errors: list[Exception] = [] + + def upsert(session_id: str, model: str) -> None: + try: + manager.create_session(cwd=cwd, model=model, session_id=session_id) + except Exception as e: + errors.append(e) + + t1 = threading.Thread(target=upsert, args=("session-a", "a")) + t2 = threading.Thread(target=upsert, args=("session-b", "b")) + t1.start() + t2.start() + t1.join() + t2.join() + + manager._read_index = original_read + + assert not errors + assert {s.id for s in manager.list_sessions(cwd)} == { + "existing", + "session-a", + "session-b", + } + + +def test_index_deduplicates_on_read(tmp_path: Path) -> None: + """Append-only _upsert writes duplicate lines for updated sessions; + the read path deduplicates them via _deduplicate_records. + """ + manager = SessionManager(TauPaths(home=tmp_path / ".tau", agents_home=tmp_path / ".agents")) + cwd = tmp_path / "project" + cwd.mkdir() + + manager.create_session(cwd=cwd, model="a", session_id="session-a") + + updated = manager.touch_session("session-a", title="Updated") + assert updated is not None + assert updated.title == "Updated" + + index_path = manager.project_index_path(cwd) + raw_lines = [ + line for line in index_path.read_text(encoding="utf-8").splitlines() if line.strip() + ] + session_a_lines = [line for line in raw_lines if '"session-a"' in line] + assert len(session_a_lines) == 2 # original + update + + fetched = manager.get_session("session-a") + assert fetched is not None + assert fetched.title == "Updated" + assert [s.id for s in manager.list_sessions(cwd)] == ["session-a"] From 63d9bca89a223a68ab88f68c3bcd133b174125ce Mon Sep 17 00:00:00 2001 From: jheronimus Date: Mon, 13 Jul 2026 18:31:43 +0300 Subject: [PATCH 2/2] fix: re-index orphaned sessions on next persist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _refresh_persisted_state called touch_session, which silently no-ops when get_session returns None. A session whose index entry was lost (or never written — e.g. created by an older Tau version that didn't emit the session_info/model_change/thinking_level_change trio, or imported with dangling parent_ids) stayed invisible to /resume and tau -c forever, even while actively used. Replace the touch_session call with _touch_or_index_session, which re-creates the index entry when missing and touches it otherwise. _index_current_session now delegates to the same helper with update_timestamp=False so the first-persist index entry keeps its original created_at. This is the indexing portion of upstream PR #308 (closed unmerged); issue #306 was closed as 'already fixed' but the _touch_or_index_session fix is not in upstream/main. --- src/tau_coding/session.py | 31 ++++++++++++++---------- tests/test_coding_session.py | 46 ++++++++++++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 13 deletions(-) diff --git a/src/tau_coding/session.py b/src/tau_coding/session.py index d0316c792..1a14cc732 100644 --- a/src/tau_coding/session.py +++ b/src/tau_coding/session.py @@ -1746,12 +1746,7 @@ def _invalidate_context_usage_cache(self) -> None: async def _refresh_persisted_state(self, *, leaf_id: str | None) -> None: entries = await self._read_session_entries() self._state = SessionState.from_entries(entries, leaf_id=leaf_id) - if self._config.session_id is not None and self._config.session_manager is not None: - self._config.session_manager.touch_session( - self._config.session_id, - model=self.model, - provider_name=self.provider_name, - ) + self._touch_or_index_session() async def _read_session_entries(self) -> list[SessionEntry]: """Read stored entries, detaching roots imported from external history.""" @@ -1782,17 +1777,27 @@ def _ensure_session_file_initialized(self) -> None: self._pending_initial_entries = () def _index_current_session(self) -> None: + self._touch_or_index_session(update_timestamp=False) + self._config = replace(self._config, index_on_first_persist=False) + + def _touch_or_index_session(self, *, update_timestamp: bool = True) -> None: if self._config.session_id is None or self._config.session_manager is None: return existing = self._config.session_manager.get_session(self._config.session_id) - if existing is not None: + if existing is None: + self._config.session_manager.create_session( + cwd=self.cwd, + model=self.model, + provider_name=self.provider_name, + session_id=self._config.session_id, + ) return - self._config.session_manager.create_session( - cwd=self.cwd, - model=self.model, - provider_name=self.provider_name, - session_id=self._config.session_id, - ) + if update_timestamp: + self._config.session_manager.touch_session( + self._config.session_id, + model=self.model, + provider_name=self.provider_name, + ) async def _try_auto_compact( self, diff --git a/tests/test_coding_session.py b/tests/test_coding_session.py index b74f0ea99..cfd074b44 100644 --- a/tests/test_coding_session.py +++ b/tests/test_coding_session.py @@ -1794,6 +1794,52 @@ async def test_session_touches_session_manager_after_persisting_messages(tmp_pat assert updated.updated_at >= record.updated_at +@pytest.mark.anyio +async def test_refresh_persisted_state_re_indexes_missing_session(tmp_path: Path) -> None: + """A session whose index entry is missing gets re-indexed on the next persist. + + Reproduces the orphan condition: a session .jsonl file exists on disk but + its index entry was lost (or never written). Before the fix, every + touch_session call silently no-op'd because get_session returned None, + so the session stayed invisible to /resume and tau -c forever. After the + fix, _refresh_persisted_state re-creates the index entry via + _touch_or_index_session. + """ + storage = JsonlSessionStorage(tmp_path / "session.jsonl") + manager = SessionManager(TauPaths(home=tmp_path / ".tau", agents_home=tmp_path / ".agents")) + record = manager.create_session(cwd=tmp_path, model="fake") + provider = FakeProvider( + [ + [ + ProviderResponseStartEvent(model="fake"), + ProviderResponseEndEvent(message=AssistantMessage(content="Hi")), + ] + ] + ) + config = CodingSessionConfig( + provider=provider, + model="fake", + system="You are Tau.", + storage=storage, + cwd=tmp_path, + session_id=record.id, + session_manager=manager, + resource_paths=TauResourcePaths(root=tmp_path / "resources", agents_root=None), + ) + session = await CodingSession.load(config) + + # Simulate index corruption: wipe the index after session creation + index_path = manager.project_index_path(tmp_path) + index_path.write_text("") + assert manager.get_session(record.id) is None + + _events = await _collect_session_events(session.prompt("Hello")) + + reindexed = manager.get_session(record.id) + assert reindexed is not None + assert reindexed.id == record.id + + @pytest.mark.anyio async def test_session_auto_names_first_unnamed_managed_session(tmp_path: Path) -> None: storage = JsonlSessionStorage(tmp_path / "session.jsonl")