feat(storage): add sqlite durable fencing

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

@@ -7,11 +7,14 @@
 import importlib
 import logging
 from typing import Any, Protocol, runtime_checkable
+from uuid import uuid4
 
 from coding_agent.adapter import PipelineAdapter
 from coding_agent.core.config import Config
 from coding_agent.local_storage import local_sqlite_storage_config
+from coding_agent.local_storage import local_sqlite_path_from_storage_config
 from coding_agent.server.session_manager import SessionManager
+from coding_agent.server.stores.session_owner_store import SQLiteSessionOwnerStore
 
 logger = logging.getLogger(__name__)
 
@@ -114,7 +117,7 @@ async def submit_approval_response(
 
     async def resume_session(self, session_id: str, **kwargs: Any) -> Any: ...
 
-    def attach_runtime(
+    async def attach_runtime(
         self,
         managed_session: LocalCliManagedSession,
         *,
@@ -146,11 +149,25 @@ class ServerBackedLocalCliSessionManager:
     def __post_init__(self) -> None:
         if self.owner_renew_interval_seconds <= 0:
             raise ValueError("owner_renew_interval_seconds must be positive")
+        owner_store = self.owner_store
+        owner_id = self.owner_id
+        fencing_token = self.fencing_token
+        if owner_store is None:
+            owner_store = SQLiteSessionOwnerStore(
+                local_sqlite_path_from_storage_config(
+                    dict(self.storage_config),
+                )
+            )
+            owner_id = owner_id or f"local-cli:{uuid4().hex}"
+            fencing_token = fencing_token or 1
+        self.owner_store = owner_store
+        self.owner_id = owner_id
+        self.fencing_token = fencing_token
         self._delegate = SessionManager(
             storage_config=dict(self.storage_config),
-            owner_store=self.owner_store,
-            owner_id=self.owner_id,
-            fencing_token=self.fencing_token,
+            owner_store=owner_store,
+            owner_id=owner_id,
+            fencing_token=fencing_token,
         )
 
     async def create_session(self, **kwargs: Any) -> str:
@@ -269,7 +286,7 @@ async def submit_approval_response(
     async def resume_session(self, session_id: str, **kwargs: Any) -> Any:
         return await self._delegate.resume_session(session_id, **kwargs)
 
-    def attach_runtime(
+    async def attach_runtime(
         self,
         managed_session: LocalCliManagedSession,
         *,
@@ -283,7 +300,7 @@ def attach_runtime(
             adapter=pipeline_adapter,
         )
         managed_session.tape_id = pipeline_ctx.tape.tape_id
-        self._delegate._persist_session(managed_session)
+        await self._delegate._persist_session_async(managed_session)
 
     async def close(self) -> None:
         if self._owner_renew_task is not None:

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

@@ -476,6 +476,8 @@ async def _resume_managed_session(
                 raise click.UsageError("Pass --session <id> or --last.")
             resolved_session_id = await _latest_managed_session_id(session_manager)
         session = await session_manager.get_session_async(resolved_session_id)
+        await session_manager.acquire_session_owner(resolved_session_id)
+        await session_manager.start_owner_lease_renewal()
         consumer = HeadlessConsumer(auto_approve=config.approval_mode == "yolo")
         task = asyncio.create_task(
             session_manager.resume_session(
@@ -612,6 +614,7 @@ async def _run_managed_one_shot(config: Config, goal: str, consumer) -> None:
             base_url=config.base_url,
             max_steps=config.max_steps,
         )
+        await session_manager.start_owner_lease_renewal()
         session = await session_manager.get_session_async(session_id)
         task = asyncio.create_task(session_manager.run_agent(session_id, goal))
         await _stream_managed_session_wire(session.wire, task, consumer)

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

@@ -175,12 +175,13 @@ async def _initialize_managed_session(self) -> None:
         )
         self.context["session_id"] = session_id
         managed_session = self._session_manager.get_session(session_id)
-        self._session_manager.attach_runtime(
+        await self._session_manager.attach_runtime(
             managed_session,
             pipeline=self._pipeline,
             pipeline_ctx=self._pipeline_ctx,
             pipeline_adapter=self._pipeline_adapter,
         )
+        await self._session_manager.start_owner_lease_renewal()
         self._managed_session_initialized = True
 
     async def _create_managed_session(self) -> str:
@@ -386,6 +387,7 @@ async def run(self):
             self._footer.disable()
             if self._pipeline_adapter is not None:
                 await self._pipeline_adapter.close()
+            await self._session_manager.close()
 
         print_pt("\nSession ended.\n", output=prompt_output)