feat(acp): use sqlite local durability

coding-agent/docs/adr/0067-local-sqlite-durable-tape-runtime.md

@@ -0,0 +1,124 @@
+# ADR-0067: Use SQLite-first durable tape and runtime stores for local ACP
+
+**Status**: Proposed
+**Date**: 2026-06-05
+
+## Context
+
+ACP `session/load` now replays durable display events, but local ACP still
+starts its session manager with filesystem session metadata and JSONL runtime
+events. Real stdio validation showed that durable run display events can
+survive process restart while the result endpoint projection can still lose the
+final answer. That is a source-of-truth problem: replayable event history,
+runtime result projections, and local multi-process access are not yet using one
+transactional local store.
+
+The repository already has the storage pieces needed for a narrower fix.
+`SQLiteTapeStore` persists append-only tape entries with indexes, and
+`SQLiteRuntimeStore` persists runs, runtime events, message snapshots, and
+interactions. PostgreSQL-backed session ownership already models owner leases
+with fencing tokens for multi-instance HTTP sessions. Bub's tape store design
+reinforces the same boundary: tape remains the logical append-only model, while
+SQLite or SQLAlchemy are replaceable store backends.
+
+This ADR scopes the next refactor to local ACP/local CLI durability. It must not
+change cloud production semantics, require multiple pods to share a SQLite file,
+or replace tape with mutable result tables.
+
+## Decision
+
+Make local ACP/local CLI storage SQLite-first while preserving tape as the
+authoritative append-only logical model.
+
+The local default for ACP should move from JSONL runtime storage to a coherent
+SQLite-backed bundle:
+
+- filesystem session metadata for local session records unless a later ADR
+  replaces local session metadata storage;
+- `SQLiteTapeStore` for authoritative tape entries;
+- `SQLiteCheckpointStore` for local checkpoints;
+- `SQLiteRuntimeStore` for runs, runtime events, interactions, and message
+  snapshots.
+
+`session/list`, `session/load`, `session/result`, and ACP replay must be derived
+from durable store state. Result records, display event replay, message
+snapshots, topic views, and last-message summaries are projections or
+checkpoints. They must remain reconstructible from tape plus durable runtime
+metadata and must not become independent sources of truth.
+
+JSONL remains available only for legacy import, export, debugging, and a
+short-term compatibility window. Do not keep JSONL and SQLite as dual
+authoritative stores for the same local session.
+
+Add a local SQLite implementation of the existing `SessionOwnerStoreProtocol`
+for ACP/local multi-process ownership. The owner record must include
+`session_id`, `owner_id`, `lease_expires_at`, and a monotonic fencing token.
+Mutation paths that already call `SessionManager._assert_owner` continue to use
+that boundary. When a lease expires, takeover must advance the fencing token so
+a suspended old owner cannot resume and append or mutate state with stale
+authority.
+
+Cloud/server production remains PostgreSQL-first for multi-instance durability.
+SQLite is a single-host local store. Do not use a shared SQLite file as the
+cloud multi-pod coordination mechanism. Future cloud changes should keep the
+same repository/store interfaces and use PostgreSQL or another server database.
+
+Implementation order:
+
+1. Add this ADR and contract tests that describe the SQLite-first local ACP
+   behavior.
+2. Add `SQLiteSessionOwnerStore` under the existing session owner store
+   protocol and run the current owner/fencing tests against it.
+3. Switch ACP/local CLI construction to configure SQLite tape, checkpoint, and
+   runtime stores, plus SQLite owner leases.
+4. Make ACP `session/list`, `session/load`, and result queries rely on durable
+   store replay/projections after process restart.
+5. Add JSONL import/export/backcompat tooling for existing local sessions.
+6. Remove JSONL as an authoritative local ACP runtime path after compatibility
+   tests and migration gates pass.
+
+## Alternatives Rejected
+
+- Keep JSONL as the local authoritative store — rejected because local
+  multi-process recovery would require hand-written locking, indexes, replay
+  repair, and ownership semantics that duplicate database behavior.
+- Replace tape with mutable session/result tables — rejected because tape is
+  the durable append-only conversation and event-history model. Result/list/load
+  data are projections.
+- Directly depend on Bub tapestore packages — rejected for now because this
+  repository already has Python `SQLiteTapeStore` and PostgreSQL stores with
+  product-specific indexing and runtime integration. Bub remains design
+  guidance for keeping store backends behind a stable tape abstraction.
+- Use SQLite for cloud multi-pod session ownership — rejected because SQLite is
+  a local single-host store and should not be treated as a distributed
+  coordination service.
+- Add a second ACP-only lease system — rejected because `SessionManager` already
+  has owner/fencing enforcement points. Local ACP should implement the existing
+  protocol, not bypass it.
+
+## Acceptance Criteria
+
+- [ ] `test_acp_uses_sqlite_local_storage_bundle_by_default`
+- [ ] `test_sqlite_session_owner_store_acquire_renew_release_and_takeover`
+- [ ] `test_acp_restart_load_replays_final_answer_from_sqlite_runtime_store`
+- [ ] `test_acp_concurrent_prompt_on_same_session_returns_busy_or_conflict`
+- [ ] `test_stale_sqlite_owner_fencing_token_cannot_mutate_session`
+- [ ] `test_jsonl_local_tape_import_preserves_sqlite_replay_projection`
+- [ ] `uv run pytest tests/acp tests/cli/test_entrypoint_contract.py tests/ui/test_session_manager_owner_checks.py tests/coding_agent/test_sqlite_runtime_store.py tests/agentkit/storage/test_sqlite.py -v`
+
+## References
+
+- `docs/adr/0003-http-sessions-use-one-stable-tape-timeline.md`
+- `docs/adr/0013-define-phase2-multi-instance-session-ownership-for-pg-http-sessions.md`
+- `docs/adr/0030-postgresql-durable-runtime-store.md`
+- `docs/adr/0033-postgresql-tape-debug-queries.md`
+- `docs/adr/0055-session-resume-and-interrupted-run-semantics.md`
+- `docs/adr/0061-acp-stdio-adapter.md`
+- `docs/adr/0063-acp-session-load-replay.md`
+- `src/agentkit/storage/sqlite.py`
+- `src/coding_agent/runtime_store.py`
+- `src/coding_agent/server/session_manager.py`
+- `src/coding_agent/server/stores/session_owner_store.py`
+- `src/coding_agent/cli/acp_command.py`
+- `https://github.com/bubbuild/bub-contrib/tree/main/packages/bub-tapestore-sqlite`
+- `https://github.com/bubbuild/bub-contrib/tree/main/packages/bub-tapestore-sqlalchemy`

coding-agent/src/coding_agent/acp/server.py

@@ -16,6 +16,7 @@
     wire_message_to_session_update,
 )
 from coding_agent.approval import ApprovalPolicy
+from coding_agent.server.stores.session_owner_store import SessionOwnershipConflictError
 from coding_agent.wire.protocol import ApprovalRequest, TurnEnd
 
 JSONObject = dict[str, Any]
@@ -30,6 +31,8 @@ async def create_session(self, **kwargs: Any) -> str: ...
 
     async def get_session_async(self, session_id: str) -> Any: ...
 
+    async def acquire_session_owner(self, session_id: str) -> None: ...
+
     async def run_agent(self, session_id: str, prompt: str) -> None: ...
 
     async def cancel_session_turn(self, session_id: str) -> Any: ...
@@ -136,6 +139,14 @@ async def handle_message(self, message: object) -> JSONObject | None:
                 "id": request_id,
                 "error": {"code": exc.code, "message": exc.message},
             }
+        except SessionOwnershipConflictError as exc:
+            if isinstance(message, dict) and "id" not in message:
+                return None
+            return {
+                "jsonrpc": "2.0",
+                "id": request_id,
+                "error": {"code": -32000, "message": str(exc)},
+            }
         except RuntimeError as exc:
             if str(exc) == "turn already in progress":
                 message_text = "Turn already in progress"
@@ -315,6 +326,7 @@ async def _session_load(self, params: JSONObject) -> None:
         )
 
         await self._session_manager.get_session_async(session_id)
+        await self._session_manager.acquire_session_owner(session_id)
         await self._session_manager.update_session_mcp_servers(
             session_id,
             mcp_servers,
@@ -370,6 +382,7 @@ async def _session_resume(self, params: JSONObject) -> None:
         )
 
         await self._session_manager.get_session_async(session_id)
+        await self._session_manager.acquire_session_owner(session_id)
         await self._session_manager.update_session_mcp_servers(
             session_id,
             mcp_servers,

coding-agent/src/coding_agent/cli/acp_command.py

@@ -1,12 +1,18 @@
 from __future__ import annotations
 
 import asyncio
+import os
+from pathlib import Path
+import time
+import uuid
 
 import click
 
+from agentkit.tracing import configure_tracing
 from coding_agent.acp import AcpServer, run_stdio
 from coding_agent.approval import ApprovalPolicy
 from coding_agent.cli.local_runtime import create_local_cli_session_manager
+from coding_agent.server.stores.session_owner_store import SQLiteSessionOwnerStore
 
 
 def _shared_cli_arg(ctx: click.Context, name: str) -> str | None:
@@ -40,11 +46,18 @@ async def _run_acp_stdio(
     approval_policy: str,
     max_steps: int,
 ) -> None:
+    configure_tracing(enabled=False)
+    data_dir = Path(os.environ.get("AGENT_DATA_DIR", "./data"))
     manager = create_local_cli_session_manager(
         storage_config={
             "http_session_backend": "fs",
-            "runtime_backend": "jsonl",
-        }
+            "tape_backend": "sqlite",
+            "checkpoint_backend": "sqlite",
+            "runtime_backend": "sqlite",
+        },
+        owner_store=SQLiteSessionOwnerStore(data_dir / "session_owners.sqlite3"),
+        owner_id=f"acp-stdio:{uuid.uuid4().hex}",
+        fencing_token=time.time_ns(),
     )
     try:
         server = AcpServer(

coding-agent/src/coding_agent/cli/local_runtime.py

@@ -55,6 +55,8 @@ def get_session(self, session_id: str) -> Any: ...
 
     async def get_session_async(self, session_id: str) -> Any: ...
 
+    async def acquire_session_owner(self, session_id: str) -> None: ...
+
     async def list_sessions_async(self) -> list[str]: ...
 
     async def close_session(self, session_id: str) -> None: ...
@@ -127,9 +129,17 @@ class ServerBackedLocalCliSessionManager:
             "runtime_backend": "jsonl",
         }
     )
+    owner_store: Any | None = None
+    owner_id: str | None = None
+    fencing_token: int | None = None
 
     def __post_init__(self) -> None:
-        self._delegate = SessionManager(storage_config=dict(self.storage_config))
+        self._delegate = SessionManager(
+            storage_config=dict(self.storage_config),
+            owner_store=self.owner_store,
+            owner_id=self.owner_id,
+            fencing_token=self.fencing_token,
+        )
 
     async def create_session(self, **kwargs: Any) -> str:
         return await self._delegate.create_session(**kwargs)
@@ -140,6 +150,9 @@ def get_session(self, session_id: str) -> Any:
     async def get_session_async(self, session_id: str) -> Any:
         return await self._delegate.get_session_async(session_id)
 
+    async def acquire_session_owner(self, session_id: str) -> None:
+        await self._delegate.acquire_session_owner(session_id)
+
     async def list_sessions_async(self) -> list[str]:
         return await self._delegate.list_sessions_async()
 
@@ -153,6 +166,16 @@ async def update_session_mcp_servers(
     ) -> None:
         await self._delegate.update_session_mcp_servers(session_id, mcp_servers)
 
+    async def update_session_additional_directories(
+        self,
+        session_id: str,
+        additional_directories: list[str],
+    ) -> None:
+        await self._delegate.update_session_additional_directories(
+            session_id,
+            additional_directories,
+        )
+
     async def list_runtime_runs(self, session_id: str) -> Any:
         return await self._delegate.list_runtime_runs(session_id)
 
@@ -234,12 +257,16 @@ def attach_runtime(
         self._delegate._persist_session(managed_session)
 
     async def close(self) -> None:
+        await self._delegate.release_owned_sessions()
         await self._delegate.close()
 
 
 def create_local_cli_session_manager(
     *,
     storage_config: dict[str, object] | None = None,
+    owner_store: Any | None = None,
+    owner_id: str | None = None,
+    fencing_token: int | None = None,
 ) -> LocalCliSessionManager:
     """Create the local REPL session manager.
 
@@ -248,9 +275,17 @@ def create_local_cli_session_manager(
     importing server/control-plane types directly.
     """
 
+    kwargs = {
+        "owner_store": owner_store,
+        "owner_id": owner_id,
+        "fencing_token": fencing_token,
+    }
     if storage_config is None:
-        return ServerBackedLocalCliSessionManager()
-    return ServerBackedLocalCliSessionManager(storage_config=storage_config)
+        return ServerBackedLocalCliSessionManager(**kwargs)
+    return ServerBackedLocalCliSessionManager(
+        storage_config=storage_config,
+        **kwargs,
+    )
 
 
 __all__ = [

coding-agent/src/coding_agent/server/session_manager.py

@@ -2356,6 +2356,7 @@ async def update_session_mcp_servers(
         mcp_servers: dict[str, dict[str, Any]],
     ) -> None:
         session = await self.get_session_async(session_id)
+        await self._assert_owner(session_id)
         resolved_mcp_servers = _session_mcp_servers_from_store(mcp_servers)
         if session.mcp_servers == resolved_mcp_servers:
             return
@@ -2370,6 +2371,7 @@ async def update_session_additional_directories(
         additional_directories: list[str],
     ) -> None:
         session = await self.get_session_async(session_id)
+        await self._assert_owner(session_id)
         resolved_additional_directories = _session_additional_directories_from_store(
             additional_directories
         )
@@ -2402,6 +2404,9 @@ def get_session(self, session_id: str) -> Session:
             Session.from_store_data(cast(dict[str, Any], loaded))
         )
 
+    async def acquire_session_owner(self, session_id: str) -> None:
+        await self._acquire_owner_for_session(session_id)
+
     def has_session(self, session_id: str) -> bool:
         """Check if a session exists.