fix: Tau doesn't add a conversation to the index on concurrent writes and orphaned sessions

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,

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]:

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")

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"]