diff --git a/acp_adapter/session.py b/acp_adapter/session.py
index bbe34b067890..8c40d9bcb38c 100644
--- a/acp_adapter/session.py
+++ b/acp_adapter/session.py
@@ -461,10 +461,31 @@ class SessionManager:
                 except Exception:
                     logger.debug("Failed to update ACP session metadata", exc_info=True)
 
-            # Replace stored messages with current history atomically so a
-            # mid-rewrite failure rolls back and the previously persisted
-            # conversation is preserved (salvaged from #13675).
-            db.replace_messages(state.session_id, state.history)
+            # When the agent owns persistence to this same SessionDB it has
+            # already flushed the live transcript incrementally during
+            # run_conversation (append_message), and it preserves pre-compaction
+            # turns non-destructively via archive_and_compact() — keeping them on
+            # disk as searchable active=0/compacted=1 rows. Calling
+            # replace_messages() here would then be a redundant double-write that
+            # DELETEs exactly those archived rows (and, after a compression-driven
+            # id rotation where agent.session_id no longer equals
+            # state.session_id, clobbers the ended parent transcript) — silent
+            # data loss for any ACP conversation long enough to compress.
+            #
+            # Only fall back to the destructive atomic replace when the agent is
+            # NOT persisting itself to this DB (e.g. a test agent factory, or a
+            # fresh create/fork whose copied history the agent has not flushed
+            # yet). That path still rolls back on a mid-rewrite failure so the
+            # previously persisted conversation survives (salvaged from #13675).
+            agent = state.agent
+            agent_db = getattr(agent, "_session_db", None)
+            agent_owns_persistence = (
+                agent_db is not None
+                and agent_db is db
+                and bool(getattr(agent, "_session_db_created", False))
+            )
+            if not agent_owns_persistence:
+                db.replace_messages(state.session_id, state.history)
         except Exception:
             logger.warning("Failed to persist ACP session %s", state.session_id, exc_info=True)
 
diff --git a/tests/acp/test_session.py b/tests/acp/test_session.py
index 5ff5e08b8072..63d2871b6474 100644
--- a/tests/acp/test_session.py
+++ b/tests/acp/test_session.py
@@ -260,6 +260,76 @@ class TestListAndCleanup:
         assert messages[0]["content"] == "original"
         assert isinstance(messages[0].get("timestamp"), (int, float))
 
+    def test_save_session_preserves_agent_archived_history(self, tmp_path):
+        """Regression: ACP _persist must not destroy compression-archived rows.
+
+        When the agent owns persistence to the same SessionDB, it has already
+        flushed the transcript itself and used archive_and_compact() to keep
+        pre-compaction turns as searchable active=0/compacted=1 rows. A blind
+        replace_messages() here used to DELETE those archived rows (and the FTS
+        index entries with them) on every save — silent data loss for any ACP
+        conversation long enough to compress.
+        """
+        db = SessionDB(tmp_path / "state.db")
+
+        def factory():
+            # Mimic a live ACP agent: it persists to *this* db and has already
+            # created its session row / flushed at least one turn.
+            return SimpleNamespace(
+                model="test-model",
+                _session_db=db,
+                _session_db_created=True,
+            )
+
+        manager = SessionManager(agent_factory=factory, db=db)
+        state = manager.create_session(cwd="/work")
+
+        # Simulate the agent's own persistence: it flushed the live transcript,
+        # then compression archived the pre-compaction turns and inserted a
+        # compacted summary as the new active set.
+        db.append_message(
+            session_id=state.session_id, role="user", content="archived needle"
+        )
+        db.archive_and_compact(
+            state.session_id, [{"role": "user", "content": "compacted summary"}]
+        )
+
+        # ACP's in-memory history only tracks the post-compaction (active) set.
+        state.history = [{"role": "user", "content": "compacted summary"}]
+        manager.save_session(state.session_id)
+
+        # The archived pre-compaction turn must survive and stay discoverable.
+        contents = [
+            m["content"]
+            for m in db.get_messages(state.session_id, include_inactive=True)
+        ]
+        assert "archived needle" in contents
+        assert "compacted summary" in contents
+        hits = {r["session_id"] for r in db.search_messages("needle")}
+        assert state.session_id in hits
+
+    def test_save_session_still_replaces_when_agent_not_self_persisting(self, manager):
+        """Agents that don't own DB persistence keep ACP as the source of truth.
+
+        The default fixture's MagicMock agent has a ``_session_db`` that is *not*
+        the manager's db, so the destructive replace path stays active and ACP
+        history overwrites cleanly (no orphaned rows from a prior save).
+        """
+        state = manager.create_session()
+        db = manager._get_db()
+
+        state.history = [{"role": "user", "content": "v1"}]
+        manager.save_session(state.session_id)
+        assert [
+            m["content"] for m in db.get_messages_as_conversation(state.session_id)
+        ] == ["v1"]
+
+        state.history = [{"role": "user", "content": "v2 replaced"}]
+        manager.save_session(state.session_id)
+        assert [
+            m["content"] for m in db.get_messages_as_conversation(state.session_id)
+        ] == ["v2 replaced"]
+
     def test_cleanup_clears_all(self, manager):
         s1 = manager.create_session()
         s2 = manager.create_session()