diff --git a/coding-agent/docs/adr/0067-local-sqlite-durable-tape-runtime.md b/coding-agent/docs/adr/0067-local-sqlite-durable-tape-runtime.md new file mode 100644 index 000000000..e5d23827f --- /dev/null +++ b/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` diff --git a/coding-agent/src/coding_agent/acp/server.py b/coding-agent/src/coding_agent/acp/server.py index 557222e4f..530387c05 100644 --- a/coding-agent/src/coding_agent/acp/server.py +++ b/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, diff --git a/coding-agent/src/coding_agent/cli/acp_command.py b/coding-agent/src/coding_agent/cli/acp_command.py index 7d0caeede..b60f04478 100644 --- a/coding-agent/src/coding_agent/cli/acp_command.py +++ b/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( diff --git a/coding-agent/src/coding_agent/cli/local_runtime.py b/coding-agent/src/coding_agent/cli/local_runtime.py index 6244d65a5..a83886196 100644 --- a/coding-agent/src/coding_agent/cli/local_runtime.py +++ b/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__ = [ diff --git a/coding-agent/src/coding_agent/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index 25dcc7364..8004386c7 100644 --- a/coding-agent/src/coding_agent/server/session_manager.py +++ b/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. diff --git a/coding-agent/src/coding_agent/server/stores/session_owner_store.py b/coding-agent/src/coding_agent/server/stores/session_owner_store.py index 380da56f7..e70a9cdc1 100644 --- a/coding-agent/src/coding_agent/server/stores/session_owner_store.py +++ b/coding-agent/src/coding_agent/server/stores/session_owner_store.py @@ -1,8 +1,11 @@ from __future__ import annotations from dataclasses import dataclass -from datetime import datetime +from datetime import UTC, datetime, timedelta from enum import StrEnum +from pathlib import Path +import sqlite3 +import threading from typing import Protocol from agentkit.storage.pg import PGPool, PGSessionOwnerStore @@ -166,3 +169,218 @@ async def get_owner(self, session_id: str) -> SessionOwnerRecord | None: lease_expires_at=lease_expires_at, fencing_token=fencing_token, ) + + +class SQLiteSessionOwnerStore: + _CREATE_SCHEMA_SQL = """ + CREATE TABLE IF NOT EXISTS session_owners ( + session_id TEXT PRIMARY KEY, + owner_id TEXT NOT NULL, + lease_expires_at TEXT NOT NULL, + fencing_token INTEGER NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """ + + def __init__(self, path: Path) -> None: + self._path = path + self._lock = threading.Lock() + self._path.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as connection: + connection.execute(self._CREATE_SCHEMA_SQL) + + def _connect(self) -> sqlite3.Connection: + connection = sqlite3.connect(self._path, timeout=5.0) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA busy_timeout = 5000") + return connection + + async def acquire( + self, + session_id: str, + owner_id: str, + lease_seconds: float = 30.0, + fencing_token: int = 1, + ) -> bool: + _require_owner_input( + session_id=session_id, + owner_id=owner_id, + lease_seconds=lease_seconds, + fencing_token=fencing_token, + ) + now = datetime.now(UTC) + expires_at = now + timedelta(seconds=lease_seconds) + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + """ + SELECT lease_expires_at, fencing_token + FROM session_owners + WHERE session_id = ? + """, + (session_id,), + ).fetchone() + if row is not None: + current_expiry = _datetime_from_sqlite_text(row["lease_expires_at"]) + if current_expiry > now: + return False + current_token = _required_sqlite_int(row, "fencing_token") + if fencing_token <= current_token: + return False + connection.execute( + """ + INSERT INTO session_owners ( + session_id, owner_id, lease_expires_at, fencing_token, + created_at, updated_at + ) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(session_id) + DO UPDATE SET + owner_id = excluded.owner_id, + lease_expires_at = excluded.lease_expires_at, + fencing_token = excluded.fencing_token, + updated_at = excluded.updated_at + """, + ( + session_id, + owner_id, + _datetime_to_sqlite_text(expires_at), + fencing_token, + _datetime_to_sqlite_text(now), + _datetime_to_sqlite_text(now), + ), + ) + return True + + async def renew( + self, + session_id: str, + owner_id: str, + lease_seconds: float = 30.0, + new_fencing_token: int = 2, + current_fencing_token: int = 1, + ) -> bool: + _require_owner_input( + session_id=session_id, + owner_id=owner_id, + lease_seconds=lease_seconds, + fencing_token=new_fencing_token, + ) + if current_fencing_token <= 0: + raise ValueError("current_fencing_token must be positive") + now = datetime.now(UTC) + expires_at = now + timedelta(seconds=lease_seconds) + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + cursor = connection.execute( + """ + UPDATE session_owners + SET lease_expires_at = ?, + fencing_token = ?, + updated_at = ? + WHERE session_id = ? + AND owner_id = ? + AND lease_expires_at > ? + AND fencing_token = ? + """, + ( + _datetime_to_sqlite_text(expires_at), + new_fencing_token, + _datetime_to_sqlite_text(now), + session_id, + owner_id, + _datetime_to_sqlite_text(now), + current_fencing_token, + ), + ) + return cursor.rowcount == 1 + + async def release( + self, + session_id: str, + owner_id: str, + fencing_token: int, + ) -> bool: + _require_owner_input( + session_id=session_id, + owner_id=owner_id, + lease_seconds=1.0, + fencing_token=fencing_token, + ) + with self._lock, self._connect() as connection: + connection.execute("BEGIN IMMEDIATE") + cursor = connection.execute( + """ + DELETE FROM session_owners + WHERE session_id = ? + AND owner_id = ? + AND fencing_token = ? + """, + (session_id, owner_id, fencing_token), + ) + return cursor.rowcount == 1 + + async def get_owner(self, session_id: str) -> SessionOwnerRecord | None: + if not session_id.strip(): + raise ValueError("session_id must be non-empty") + with self._lock, self._connect() as connection: + row = connection.execute( + """ + SELECT owner_id, lease_expires_at, fencing_token + FROM session_owners + WHERE session_id = ? + """, + (session_id,), + ).fetchone() + if row is None: + return None + return SessionOwnerRecord( + owner_id=_required_sqlite_str(row, "owner_id"), + lease_expires_at=_datetime_from_sqlite_text(row["lease_expires_at"]), + fencing_token=_required_sqlite_int(row, "fencing_token"), + ) + + +def _require_owner_input( + *, + session_id: str, + owner_id: str, + lease_seconds: float, + fencing_token: int, +) -> None: + if not session_id.strip(): + raise ValueError("session_id must be non-empty") + if not owner_id.strip(): + raise ValueError("owner_id must be non-empty") + if lease_seconds <= 0: + raise ValueError("lease_seconds must be positive") + if fencing_token <= 0: + raise ValueError("fencing_token must be positive") + + +def _datetime_to_sqlite_text(value: datetime) -> str: + return value.astimezone(UTC).isoformat() + + +def _datetime_from_sqlite_text(value: object) -> datetime: + if not isinstance(value, str): + raise TypeError("sqlite session owner datetime must be text") + parsed = datetime.fromisoformat(value) + if parsed.tzinfo is None: + return parsed.replace(tzinfo=UTC) + return parsed.astimezone(UTC) + + +def _required_sqlite_str(row: sqlite3.Row, key: str) -> str: + value = row[key] + if not isinstance(value, str): + raise TypeError(f"sqlite session owner row missing string {key}") + return value + + +def _required_sqlite_int(row: sqlite3.Row, key: str) -> int: + value = row[key] + if not isinstance(value, int): + raise TypeError(f"sqlite session owner row missing int {key}") + return value diff --git a/coding-agent/tests/acp/test_compat_harness.py b/coding-agent/tests/acp/test_compat_harness.py index a7df5a0ba..ab3d27fc0 100644 --- a/coding-agent/tests/acp/test_compat_harness.py +++ b/coding-agent/tests/acp/test_compat_harness.py @@ -143,6 +143,9 @@ async def get_session_async(self, session_id: str) -> Any: default_run_target=None, ) + async def acquire_session_owner(self, session_id: str) -> None: + self.calls.append(("acquire_session_owner", session_id)) + async def update_session_mcp_servers( self, session_id: str, diff --git a/coding-agent/tests/acp/test_server.py b/coding-agent/tests/acp/test_server.py index 4dea9bb63..048638d58 100644 --- a/coding-agent/tests/acp/test_server.py +++ b/coding-agent/tests/acp/test_server.py @@ -12,6 +12,7 @@ from coding_agent.acp.server import AcpServer, run_stdio from coding_agent.approval import ApprovalPolicy from coding_agent.events import DisplayEvent +from coding_agent.server.stores.session_owner_store import SessionOwnershipConflictError from coding_agent.wire.local import LocalWire from coding_agent.wire.protocol import ( ApprovalRequest, @@ -44,6 +45,7 @@ def __init__( ) ] self.additional_directories: list[str] = [] + self.raise_owner_conflict = False async def create_session(self, **kwargs: Any) -> str: self.calls.append(("create_session", kwargs)) @@ -60,6 +62,11 @@ async def get_session_async(self, session_id: str) -> Any: default_run_target=None, ) + async def acquire_session_owner(self, session_id: str) -> None: + self.calls.append(("acquire_session_owner", session_id)) + if self.raise_owner_conflict: + raise SessionOwnershipConflictError("stale owner or fencing token rejected") + async def list_sessions_async(self) -> list[str]: self.calls.append(("list_sessions_async", None)) return ["sess-1", "sess-2"] @@ -749,6 +756,40 @@ async def test_session_load_updates_additional_directories_from_params( ) in manager.calls +@pytest.mark.asyncio +async def test_session_load_rejects_owner_conflict_before_param_updates( + tmp_path: Path, +) -> None: + manager = FakeManager(repo_path=tmp_path) + manager.raise_owner_conflict = True + server = AcpServer(manager) + + response = await server.handle_message( + { + "jsonrpc": "2.0", + "id": 7, + "method": "session/load", + "params": { + "sessionId": "sess-1", + "cwd": str(tmp_path), + "mcpServers": [], + }, + } + ) + + assert response == { + "jsonrpc": "2.0", + "id": 7, + "error": { + "code": -32000, + "message": "stale owner or fencing token rejected", + }, + } + assert ("get_session_async", "sess-1") in manager.calls + assert ("acquire_session_owner", "sess-1") in manager.calls + assert not any(call[0].startswith("update_session_") for call in manager.calls) + + @pytest.mark.asyncio async def test_session_resume_updates_session_params_without_replay( tmp_path: Path, @@ -1113,8 +1154,9 @@ async def test_session_load_replays_display_events_before_response( }, }, ] - assert manager.calls[:5] == [ + assert manager.calls[:6] == [ ("get_session_async", "sess-1"), + ("acquire_session_owner", "sess-1"), ( "update_session_mcp_servers", {"session_id": "sess-1", "mcp_servers": {}}, @@ -1607,6 +1649,47 @@ async def test_stdio_server_writes_jsonrpc_to_stdout_only() -> None: assert stderr == [] +@pytest.mark.asyncio +async def test_stdio_server_drops_configured_tracing_logs_from_stdout( + capsys: pytest.CaptureFixture[str], +) -> None: + from agentkit.runtime.pipeline import logger as pipeline_logger + from agentkit.tracing import configure_tracing + + manager = FakeManager() + stdout: list[str] = [] + stderr: list[str] = [] + stdin = [ + '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":1}}\n', + ] + + configure_tracing(enabled=True) + try: + await run_stdio( + AcpServer(manager), + stdin=stdin, + stdout=stdout.append, + stderr=stderr.append, + ) + pipeline_logger.info("stage_start", stage="after_stdio", entry_count=1) + finally: + configure_tracing(enabled=False) + + assert stdout == [ + '{"jsonrpc":"2.0","id":1,"result":{"protocolVersion":1,' + '"agentCapabilities":{"loadSession":true,' + '"mcpCapabilities":{"http":false,"sse":false},' + '"promptCapabilities":{},' + '"sessionCapabilities":{"close":{},"list":{},"resume":{},' + '"additionalDirectories":{}}},' + '"agentInfo":{"name":"coding-agent","title":"Coding Agent",' + '"version":"0.1.0"},"authMethods":[]}}\n' + ] + assert stderr == [] + captured = capsys.readouterr() + assert captured.out == "" + + @pytest.mark.asyncio async def test_stdio_session_load_writes_replay_updates_before_response( tmp_path: Path, diff --git a/coding-agent/tests/cli/test_entrypoint_contract.py b/coding-agent/tests/cli/test_entrypoint_contract.py index 4e650c78a..6b15e45ac 100644 --- a/coding-agent/tests/cli/test_entrypoint_contract.py +++ b/coding-agent/tests/cli/test_entrypoint_contract.py @@ -146,6 +146,81 @@ async def fake_run_acp_stdio(ctx, approval_policy: str, max_steps: int) -> None: ] +def test_acp_stdio_disables_agentkit_tracing( + monkeypatch: pytest.MonkeyPatch, +) -> None: + from click.testing import CliRunner + + from coding_agent.cli import acp_command + from coding_agent.cli.main import main + + calls: list[tuple[str, object]] = [] + + class FakeManager: + async def close(self) -> None: + calls.append(("close", None)) + + def fake_configure_tracing(*, enabled: bool | None = None, level: str = "INFO"): + calls.append(("configure_tracing", {"enabled": enabled, "level": level})) + + def fake_create_local_cli_session_manager( + *, + storage_config, + owner_store, + owner_id, + fencing_token, + ): + calls.append( + ( + "create_manager", + { + "storage_config": storage_config, + "owner_store_type": type(owner_store).__name__, + "owner_id": owner_id, + "fencing_token_type": type(fencing_token).__name__, + }, + ) + ) + return FakeManager() + + async def fake_run_stdio(server): + calls.append(("run_stdio", type(server).__name__)) + + monkeypatch.setattr(acp_command, "configure_tracing", fake_configure_tracing) + monkeypatch.setattr( + acp_command, + "create_local_cli_session_manager", + fake_create_local_cli_session_manager, + ) + monkeypatch.setattr(acp_command, "run_stdio", fake_run_stdio) + + result = CliRunner(env=_click_credential_free_env()).invoke( + main, + ["acp"], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert calls[0] == ("configure_tracing", {"enabled": False, "level": "INFO"}) + assert calls[1] == ( + "create_manager", + { + "storage_config": { + "http_session_backend": "fs", + "tape_backend": "sqlite", + "checkpoint_backend": "sqlite", + "runtime_backend": "sqlite", + }, + "owner_store_type": "SQLiteSessionOwnerStore", + "owner_id": calls[1][1]["owner_id"], + "fencing_token_type": "int", + }, + ) + assert str(calls[1][1]["owner_id"]).startswith("acp-stdio:") + assert ("run_stdio", "AcpServer") in calls + assert calls[-1] == ("close", None) + + def test_daemon_run_help_is_daemon_backed_client_surface() -> None: runner = CliRunner(env=_click_credential_free_env()) diff --git a/coding-agent/tests/cli/test_local_runtime.py b/coding-agent/tests/cli/test_local_runtime.py index 70c520510..554f8f97d 100644 --- a/coding-agent/tests/cli/test_local_runtime.py +++ b/coding-agent/tests/cli/test_local_runtime.py @@ -19,10 +19,30 @@ def attach_runtime_binding(self, *, pipeline, ctx, adapter) -> None: class FakeDelegate: def __init__(self) -> None: self.persisted = [] + self.additional_directory_updates = [] + self.owner_acquires = [] + self.release_owned_sessions_called = False + self.close_called = False def _persist_session(self, session) -> None: self.persisted.append(session) + async def acquire_session_owner(self, session_id: str) -> None: + self.owner_acquires.append(session_id) + + async def update_session_additional_directories( + self, + session_id: str, + additional_directories: list[str], + ) -> None: + self.additional_directory_updates.append((session_id, additional_directories)) + + async def release_owned_sessions(self) -> None: + self.release_owned_sessions_called = True + + async def close(self) -> None: + self.close_called = True + def test_server_backed_local_cli_attach_runtime_uses_session_binding_delegate() -> None: manager = object.__new__(ServerBackedLocalCliSessionManager) @@ -43,3 +63,39 @@ def test_server_backed_local_cli_attach_runtime_uses_session_binding_delegate() assert session.attached == (pipeline, ctx, adapter) assert session.tape_id == "repl-tape" assert delegate.persisted == [session] + + +async def test_server_backed_local_cli_delegates_additional_directory_updates() -> None: + manager = object.__new__(ServerBackedLocalCliSessionManager) + delegate = FakeDelegate() + manager._delegate = delegate + + await manager.update_session_additional_directories( + "session-1", + ["/workspace/extra"], + ) + + assert delegate.additional_directory_updates == [ + ("session-1", ["/workspace/extra"]) + ] + + +async def test_server_backed_local_cli_delegates_owner_acquire() -> None: + manager = object.__new__(ServerBackedLocalCliSessionManager) + delegate = FakeDelegate() + manager._delegate = delegate + + await manager.acquire_session_owner("session-1") + + assert delegate.owner_acquires == ["session-1"] + + +async def test_server_backed_local_cli_releases_owner_leases_before_close() -> None: + manager = object.__new__(ServerBackedLocalCliSessionManager) + delegate = FakeDelegate() + manager._delegate = delegate + + await manager.close() + + assert delegate.release_owned_sessions_called is True + assert delegate.close_called is True diff --git a/coding-agent/tests/ui/test_session_owner_store.py b/coding-agent/tests/ui/test_session_owner_store.py index 16984d7b9..ecbfca0bf 100644 --- a/coding-agent/tests/ui/test_session_owner_store.py +++ b/coding-agent/tests/ui/test_session_owner_store.py @@ -1,10 +1,13 @@ from __future__ import annotations +import asyncio from datetime import UTC, datetime, timedelta +from pathlib import Path import pytest from coding_agent.server.stores.session_owner_store import ( SessionOwnerRecord, + SQLiteSessionOwnerStore, SessionOwnerStore, ) @@ -116,3 +119,118 @@ async def test_session_owner_get_owner_returns_typed_record( lease_expires_at=owner.lease_expires_at, fencing_token=1, ) + + +@pytest.mark.asyncio +async def test_sqlite_session_owner_store_acquire_renew_release_and_takeover( + tmp_path: Path, +) -> None: + store = SQLiteSessionOwnerStore(tmp_path / "owners.sqlite3") + + assert await store.acquire("s1", "owner-a", lease_seconds=30.0, fencing_token=1) + assert not await store.acquire( + "s1", + "owner-b", + lease_seconds=30.0, + fencing_token=2, + ) + + renewed = await store.renew( + "s1", + "owner-a", + lease_seconds=30.0, + new_fencing_token=2, + current_fencing_token=1, + ) + owner = await store.get_owner("s1") + + assert renewed is True + assert owner is not None + assert owner.owner_id == "owner-a" + assert owner.fencing_token == 2 + + assert await store.release("s1", "owner-a", fencing_token=2) + assert await store.get_owner("s1") is None + + assert await store.acquire( + "s1", + "owner-dead", + lease_seconds=0.001, + fencing_token=3, + ) + await asyncio.sleep(0.01) + + assert await store.acquire("s1", "owner-b", lease_seconds=30.0, fencing_token=4) + taken = await store.get_owner("s1") + + assert taken == SessionOwnerRecord( + owner_id="owner-b", + lease_expires_at=taken.lease_expires_at, + fencing_token=4, + ) + + +@pytest.mark.asyncio +async def test_sqlite_session_owner_store_rejects_stale_owner( + tmp_path: Path, +) -> None: + store = SQLiteSessionOwnerStore(tmp_path / "owners.sqlite3") + assert await store.acquire("s1", "owner-a", lease_seconds=30.0, fencing_token=1) + + assert not await store.renew( + "s1", + "owner-a", + lease_seconds=30.0, + new_fencing_token=2, + current_fencing_token=99, + ) + assert not await store.release("s1", "owner-b", fencing_token=1) + + owner = await store.get_owner("s1") + + assert owner == SessionOwnerRecord( + owner_id="owner-a", + lease_expires_at=owner.lease_expires_at, + fencing_token=1, + ) + + +@pytest.mark.asyncio +async def test_sqlite_session_owner_store_rejects_non_increasing_takeover_token( + tmp_path: Path, +) -> None: + store = SQLiteSessionOwnerStore(tmp_path / "owners.sqlite3") + assert await store.acquire( + "s1", + "owner-dead", + lease_seconds=0.001, + fencing_token=10, + ) + await asyncio.sleep(0.01) + + assert not await store.acquire( + "s1", + "owner-lower", + lease_seconds=30.0, + fencing_token=7, + ) + + owner = await store.get_owner("s1") + assert owner == SessionOwnerRecord( + owner_id="owner-dead", + lease_expires_at=owner.lease_expires_at, + fencing_token=10, + ) + + assert await store.acquire( + "s1", + "owner-next", + lease_seconds=30.0, + fencing_token=11, + ) + taken = await store.get_owner("s1") + assert taken == SessionOwnerRecord( + owner_id="owner-next", + lease_expires_at=taken.lease_expires_at, + fencing_token=11, + )