diff --git a/src/tau_coding/session.py b/src/tau_coding/session.py index d0316c79..1a14cc73 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/src/tau_coding/session_manager.py b/src/tau_coding/session_manager.py index 1cccf229..860d926a 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_coding_session.py b/tests/test_coding_session.py index b74f0ea9..cfd074b4 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") diff --git a/tests/test_session_manager.py b/tests/test_session_manager.py index 491944c5..685d80fa 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"]