diff --git a/coding-agent/docs/adr/0015-enforce-owner-routed-http-event-streams.md b/coding-agent/docs/adr/0015-enforce-owner-routed-http-event-streams.md index 3f43a8f61..2334c9477 100644 --- a/coding-agent/docs/adr/0015-enforce-owner-routed-http-event-streams.md +++ b/coding-agent/docs/adr/0015-enforce-owner-routed-http-event-streams.md @@ -1,6 +1,6 @@ # ADR-0015: Enforce owner-routed HTTP event streams for Phase 2 sticky routing -**Status**: Proposed +**Status**: Accepted **Date**: 2026-04-21 ## Context @@ -25,6 +25,22 @@ Specifically: This ADR intentionally keeps approval metadata persistence out of scope. It closes the narrowest, clearest routing gap first: only the current owner may serve the live SSE stream for a session. +## Implementation Status + +The sticky-owner first cut is implemented. `GET /sessions/{session_id}/events` +and `/sessions/{session_id}/display-events` authorize ownership before queue +registration, register through the owned queue path, and re-check ownership +inside the stream before delivering queued events or keepalive pings. + +Covered regressions live in `tests/ui/test_http_server_failover.py`, including +stale-owner rejection before queue registration, owner-change stream shutdown, +queued-event drop after owner change, disappearing-session races, owner change +after queue registration, the append-to-recheck race window, and current-owner +keepalive behavior. + +Brokered cross-instance stream routing, redirect payloads, and queue transfer +remain intentionally out of scope for this ADR. + ## Alternatives Rejected - Keep `/events` unfenced and rely on load-balancer stickiness alone — rejected because it leaves stale instances free to attach live streams even when PostgreSQL ownership already says they are not authoritative. @@ -34,17 +50,18 @@ This ADR intentionally keeps approval metadata persistence out of scope. It clos ## Acceptance Criteria -- [ ] `test_get_events_rejects_stale_owner_before_stream_registration` -- [ ] `test_get_events_stops_stream_after_owner_change` -- [ ] `test_get_events_keeps_stream_alive_for_current_owner` -- [ ] `test_get_events_returns_404_before_owner_check_for_missing_session` -- [ ] `uv run pytest tests/ui/test_http_server_failover.py -v` -- [ ] `uv run pytest tests/ui/test_http_server.py tests/ui/test_http_server_failover.py tests/ui/test_session_manager_owner_checks.py -k "events or owner or failover" -v` +- [x] `test_get_events_rejects_stale_owner_before_stream_registration` +- [x] `test_get_events_stops_stream_after_owner_change` +- [x] `test_get_events_keeps_stream_alive_for_current_owner` +- [x] `test_get_events_returns_404_before_owner_check_for_missing_session` +- [x] `uv run pytest tests/ui/test_http_server_failover.py -v` +- [x] `uv run pytest tests/ui/test_http_server.py tests/ui/test_http_server_failover.py tests/ui/test_session_manager_owner_checks.py -k "events or owner or failover" -v` ## References - `docs/adr/0013-define-phase2-multi-instance-session-ownership-for-pg-http-sessions.md` - `docs/pg-phase2-multi-instance-checklist.md` -- `src/coding_agent/ui/http_server.py` -- `src/coding_agent/ui/session_manager.py` +- `src/coding_agent/server/http_server.py` +- `src/coding_agent/server/session_manager.py` - `tests/ui/test_http_server.py` +- `tests/ui/test_http_server_failover.py` diff --git a/coding-agent/docs/adr/0060-native-platform-sandbox-backends.md b/coding-agent/docs/adr/0060-native-platform-sandbox-backends.md index ce3625d5b..9b8ac0f9d 100644 --- a/coding-agent/docs/adr/0060-native-platform-sandbox-backends.md +++ b/coding-agent/docs/adr/0060-native-platform-sandbox-backends.md @@ -82,6 +82,16 @@ The default config value stays `sandbox_mode = "none"` in this change. Switching the default to `native` is deferred to a separate, final PR after the native backends are implemented and verified, to keep blast radius minimal. +## Implementation Status + +Implemented. The user-visible `sandbox_mode` values are now converged to +`none`, `native`, `podman`, and `docker`; `nsjail` is rejected by validation. +Native sandbox resolution fails closed when the required platform backend is +unavailable, and the default remains `none`. + +Verification is captured by `tests/tools/test_shell.py` and the acceptance +criteria below. + ## Alternatives Rejected - Expose `landlock`, `bwrap`, and `seatbelt` as separate user-visible modes — diff --git a/coding-agent/docs/adr/0068-local-sqlite-transactional-durable-fencing.md b/coding-agent/docs/adr/0068-local-sqlite-transactional-durable-fencing.md index 743cf3f27..ec8df1f4a 100644 --- a/coding-agent/docs/adr/0068-local-sqlite-transactional-durable-fencing.md +++ b/coding-agent/docs/adr/0068-local-sqlite-transactional-durable-fencing.md @@ -1,6 +1,6 @@ # ADR-0068: Local SQLite transactional durable fencing -**Status**: Proposed +**Status**: Accepted **Date**: 2026-06-05 Supersedes ADR-0067. ADR-0067 remains the record of the first ACP SQLite @@ -208,6 +208,22 @@ PR 4 adds PG/cloud parity. It is not required to block the local SQLite implementation, but the PR 3 interfaces must already support a PG transaction implementation. +## Implementation Status + +Accepted and implemented for the local SQLite and PostgreSQL durable fencing +paths. + +Implementation references: + +- `7911839` — `feat(storage): default local runtime to sqlite bundle (#559)` +- `3c5a2f6` — `feat(storage): add sqlite durable fencing` +- `40a3110` — `feat(storage): add pg durable fencing` +- `4d8cafe` — `fix(storage): restore sqlite checkpoint snapshots` + +Operational follow-up: startup diagnostics now reject mixed local SQLite +backend/path configurations instead of silently disabling durable fencing, and +warn when custom in-process stores intentionally disable local or PG fencing. + ## Alternatives Rejected - Keep ADR-0067 as the target architecture — rejected because filesystem @@ -234,28 +250,28 @@ implementation. ## Acceptance Criteria -- [ ] `test_adr_0068_supersedes_adr_0067_without_conflicting_target_architecture` -- [ ] `test_adr_0068_defines_single_db_epoch_and_transactional_fencing_contract` -- [ ] `test_adr_0068_rejects_attach_as_the_durable_fencing_foundation` -- [ ] `test_adr_0068_requires_same_session_target_ownership_for_protected_writes` -- [ ] `test_adr_0068_covers_worker_maintenance_and_cleanup_authority` -- [ ] `test_adr_0068_records_full_write_path_inventory_and_future_pr_sequence` -- [ ] Future PR 3: owner of session A cannot mutate run, checkpoint, tape, +- [x] `test_adr_0068_supersedes_adr_0067_without_conflicting_target_architecture` +- [x] `test_adr_0068_defines_single_db_epoch_and_transactional_fencing_contract` +- [x] `test_adr_0068_rejects_attach_as_the_durable_fencing_foundation` +- [x] `test_adr_0068_requires_same_session_target_ownership_for_protected_writes` +- [x] `test_adr_0068_covers_worker_maintenance_and_cleanup_authority` +- [x] `test_adr_0068_records_full_write_path_inventory_and_future_pr_sequence` +- [x] Future PR 3: owner of session A cannot mutate run, checkpoint, tape, interaction, snapshot, or event target of session B. -- [ ] Future PR 3: A acquires epoch N, A writes own session target, and the +- [x] Future PR 3: A acquires epoch N, A writes own session target, and the write succeeds. -- [ ] Future PR 3: A lease expires, B takes over with epoch N+1, A write fails, +- [x] Future PR 3: A lease expires, B takes over with epoch N+1, A write fails, and B write succeeds. -- [ ] Future PR 3: A renews and keeps epoch N, and A write succeeds. -- [ ] Future PR 3: expired same owner without live lease cannot write until the +- [x] Future PR 3: A renews and keeps epoch N, and A write succeeds. +- [x] Future PR 3: expired same owner without live lease cannot write until the reacquire/takeover policy grants a live epoch. -- [ ] Future PR 3: stale owner cancellation cleanup cannot persist final +- [x] Future PR 3: stale owner cancellation cleanup cannot persist final status, result, or session metadata through the normal write path. -- [ ] Future PR 3: worker claim, heartbeat, and finalize cannot bypass session +- [x] Future PR 3: worker claim, heartbeat, and finalize cannot bypass session takeover or target ownership rules. -- [ ] Future PR 3: checkpoint restore is atomic or has an explicit recovery +- [x] Future PR 3: checkpoint restore is atomic or has an explicit recovery contract. -- [ ] `uv run pytest tests/coding_agent/test_durable_fencing_adr_contract.py -q` +- [x] `uv run pytest tests/coding_agent/test_durable_fencing_adr_contract.py -q` ## References diff --git a/coding-agent/src/coding_agent/local_storage.py b/coding-agent/src/coding_agent/local_storage.py index b41bd6cbc..b37f38269 100644 --- a/coding-agent/src/coding_agent/local_storage.py +++ b/coding-agent/src/coding_agent/local_storage.py @@ -6,6 +6,18 @@ LOCAL_SQLITE_FILENAME = "local.sqlite3" +DURABLE_STORAGE_BACKEND_PATH_KEYS = ( + ("http_session_backend", "http_session_path"), + ("tape_backend", "tape_path"), + ("checkpoint_backend", "checkpoint_path"), + ("runtime_backend", "runtime_path"), +) +DURABLE_STORAGE_BACKEND_KEYS = tuple( + backend_key for backend_key, _ in DURABLE_STORAGE_BACKEND_PATH_KEYS +) +DURABLE_STORAGE_PATH_KEYS = tuple( + path_key for _, path_key in DURABLE_STORAGE_BACKEND_PATH_KEYS +) def normalize_storage_path(path: str) -> Path: @@ -47,13 +59,7 @@ def local_sqlite_path_from_storage_config( if isinstance(local_path, str) and local_path.strip(): return Path(local_path) explicit_paths = [ - storage_config.get(path_key) - for path_key in ( - "http_session_path", - "tape_path", - "checkpoint_path", - "runtime_path", - ) + storage_config.get(path_key) for path_key in DURABLE_STORAGE_PATH_KEYS ] if all(isinstance(path, str) and path.strip() for path in explicit_paths): resolved_paths = [normalize_storage_path(str(path)) for path in explicit_paths] @@ -63,19 +69,45 @@ def local_sqlite_path_from_storage_config( return local_sqlite_path(data_dir) +def durable_storage_backend_values( + storage_config: dict[str, Any], +) -> dict[str, str]: + return { + key: str(storage_config.get(key, "")).strip().lower() + for key in DURABLE_STORAGE_BACKEND_KEYS + } + + +def storage_has_any_sqlite_backend(storage_config: dict[str, Any]) -> bool: + return any( + value == "sqlite" + for value in durable_storage_backend_values(storage_config).values() + ) + + +def storage_uses_local_sqlite_bundle( + storage_config: dict[str, Any], + data_dir: Path | str | None = None, +) -> bool: + backend_values = durable_storage_backend_values(storage_config) + if any(value != "sqlite" for value in backend_values.values()): + return False + local_path = normalize_storage_path( + str(local_sqlite_path_from_storage_config(storage_config, data_dir)) + ) + return all( + normalize_storage_path(str(storage_config.get(key, ""))) == local_path + for key in DURABLE_STORAGE_PATH_KEYS + ) + + def with_local_sqlite_bundle_paths( storage_config: dict[str, Any], data_dir: Path | str | None = None, ) -> dict[str, Any]: config = dict(storage_config) local_path = str(local_sqlite_path_from_storage_config(config, data_dir)) - backend_path_pairs = ( - ("http_session_backend", "http_session_path"), - ("tape_backend", "tape_path"), - ("checkpoint_backend", "checkpoint_path"), - ("runtime_backend", "runtime_path"), - ) - for backend_key, path_key in backend_path_pairs: + for backend_key, path_key in DURABLE_STORAGE_BACKEND_PATH_KEYS: backend = config.get(backend_key) if isinstance(backend, str) and backend.strip().lower() == "sqlite": configured_path = config.get(path_key) diff --git a/coding-agent/src/coding_agent/server/http_server.py b/coding-agent/src/coding_agent/server/http_server.py index e7c48561f..f0bf5b06f 100644 --- a/coding-agent/src/coding_agent/server/http_server.py +++ b/coding-agent/src/coding_agent/server/http_server.py @@ -88,7 +88,8 @@ RuntimeEventRecord, ) from coding_agent.local_storage import local_sqlite_path_from_storage_config -from coding_agent.local_storage import normalize_storage_path +from coding_agent.local_storage import local_sqlite_storage_config +from coding_agent.local_storage import storage_has_any_sqlite_backend from coding_agent.events import DisplayEvent, DisplayEventStreamProjector from coding_agent.scheduled_runs import ( PGScheduledRunStore, @@ -209,7 +210,6 @@ ResumeSessionRequest, RuntimeConfigUpdateRequest, RuntimeConfigUpdateResponse, - ThinkingConfigSchema, RuntimeEventResponse, RuntimeEventsResponse, RuntimeInteractionListResponse, @@ -729,33 +729,6 @@ def _storage_uses_pg_http_sessions(storage_config: dict[str, Any]) -> bool: return tape_backend == "pg" -def _storage_uses_local_sqlite_bundle(storage_config: dict[str, Any]) -> bool: - backend_keys = ( - "http_session_backend", - "tape_backend", - "checkpoint_backend", - "runtime_backend", - ) - if any( - str(storage_config.get(key, "")).strip().lower() != "sqlite" - for key in backend_keys - ): - return False - local_path = normalize_storage_path( - str(local_sqlite_path_from_storage_config(storage_config)) - ) - path_keys = ( - "http_session_path", - "tape_path", - "checkpoint_path", - "runtime_path", - ) - return all( - normalize_storage_path(str(storage_config.get(key, ""))) == local_path - for key in path_keys - ) - - def _configured_owner_id(storage_config: dict[str, Any]) -> str: owner_id = storage_config.get("owner_id") if isinstance(owner_id, str) and owner_id.strip(): @@ -783,6 +756,7 @@ def _configured_owner_lease_seconds(storage_config: dict[str, Any]) -> float: def _build_session_manager() -> SessionManager: storage_config = _load_storage_config() + effective_storage_config = storage_config or local_sqlite_storage_config() remote_retention_config = _load_remote_retention_config() try: _validate_production_config( @@ -794,23 +768,25 @@ def _build_session_manager() -> SessionManager: except Exception: logger.exception("Production config validation failed") raise - uses_local_sqlite_bundle = _storage_uses_local_sqlite_bundle(storage_config) + has_local_sqlite_intent = storage_has_any_sqlite_backend(effective_storage_config) manager = SessionManager( - storage_config=storage_config, + storage_config=effective_storage_config, owner_store=( SQLiteSessionOwnerStore( - local_sqlite_path_from_storage_config(storage_config) + local_sqlite_path_from_storage_config(effective_storage_config) ) - if uses_local_sqlite_bundle + if has_local_sqlite_intent else None ), owner_id=( - _configured_owner_id(storage_config) if uses_local_sqlite_bundle else None + _configured_owner_id(effective_storage_config) + if has_local_sqlite_intent + else None ), - fencing_token=1 if uses_local_sqlite_bundle else None, + fencing_token=1 if has_local_sqlite_intent else None, owner_lease_seconds=( - _configured_owner_lease_seconds(storage_config) - if uses_local_sqlite_bundle + _configured_owner_lease_seconds(effective_storage_config) + if has_local_sqlite_intent else 30.0 ), cloud_workspace_client_factory=( @@ -824,14 +800,14 @@ def _build_session_manager() -> SessionManager: manager.configure_workspace_metadata_store( PGWorkspaceMetadataStore(pool=manager.pg_pool) ) - if not _storage_uses_pg_http_sessions(storage_config): + if not _storage_uses_pg_http_sessions(effective_storage_config): return manager owner_store = SessionOwnerStore(pg_pool=manager.pg_pool) manager.configure_owner_leases( owner_store=owner_store, - owner_id=_configured_owner_id(storage_config), - fencing_token=_configured_fencing_token(storage_config), - owner_lease_seconds=_configured_owner_lease_seconds(storage_config), + owner_id=_configured_owner_id(effective_storage_config), + fencing_token=_configured_fencing_token(effective_storage_config), + owner_lease_seconds=_configured_owner_lease_seconds(effective_storage_config), ) return manager @@ -2478,7 +2454,9 @@ async def create_session( return SessionResponse(session_id=session_id) -@app.patch("/sessions/{session_id}/runtime-config", response_model=RuntimeConfigUpdateResponse) +@app.patch( + "/sessions/{session_id}/runtime-config", response_model=RuntimeConfigUpdateResponse +) @limiter.limit(RateLimits.CREATE_SESSION) async def update_runtime_config( request: Request, diff --git a/coding-agent/src/coding_agent/server/session_manager.py b/coding-agent/src/coding_agent/server/session_manager.py index bedd306b7..bee8fac90 100644 --- a/coding-agent/src/coding_agent/server/session_manager.py +++ b/coding-agent/src/coding_agent/server/session_manager.py @@ -16,6 +16,7 @@ from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol, TypeVar, cast from agentkit.environment import Environment +from agentkit.errors import ConfigError from agentkit.storage.checkpoint_fs import FSCheckpointStore from agentkit.storage.pg import PGPool from agentkit.storage.sqlite import SQLiteCheckpointStore, SQLiteTapeStore @@ -49,6 +50,9 @@ from coding_agent.approval.store import ApprovalStore from coding_agent.core import config as core_config from coding_agent.local_storage import ( + DURABLE_STORAGE_BACKEND_KEYS, + DURABLE_STORAGE_PATH_KEYS, + durable_storage_backend_values, local_sqlite_storage_config, local_sqlite_path_from_storage_config, normalize_storage_path, @@ -162,6 +166,29 @@ logger = logging.getLogger(__name__) + +def _custom_store_names( + *, + store: object | None, + tape_store: object | None, + checkpoint_store: object | None, + checkpoint_service: object | None, + runtime_store: object | None, +) -> list[str]: + names: list[str] = [] + if store is not None: + names.append("store") + if tape_store is not None: + names.append("tape_store") + if checkpoint_store is not None: + names.append("checkpoint_store") + if checkpoint_service is not None: + names.append("checkpoint_service") + if runtime_store is not None: + names.append("runtime_store") + return names + + T = TypeVar("T") @@ -824,19 +851,15 @@ def __init__( ) self._pg_pool = pg_pool self._owns_pg_pool = False - self._custom_store_supplied = store is not None - self._custom_tape_store_supplied = tape_store is not None - self._custom_checkpoint_store_supplied = ( - checkpoint_store is not None or checkpoint_service is not None - ) - self._custom_runtime_store_supplied = runtime_store is not None - self._local_durable_store = self._create_local_durable_store( - data_dir=data_dir, + self._custom_store_names = _custom_store_names( store=store, tape_store=tape_store, checkpoint_store=checkpoint_store, checkpoint_service=checkpoint_service, runtime_store=runtime_store, + ) + self._local_durable_store = self._create_local_durable_store( + data_dir=data_dir, owner_store=owner_store, ) self._pg_durable_store: PGDurableStore | None = None @@ -1194,53 +1217,59 @@ def _create_local_durable_store( self, *, data_dir: Path, - store: SessionStore | None, - tape_store: TapeStore | None, - checkpoint_store: CheckpointStore | None, - checkpoint_service: CheckpointService | None, - runtime_store: RuntimeStore | None, owner_store: SessionOwnerStoreProtocol | None, ) -> SQLiteLocalDurableStore | None: if owner_store is None or not callable( getattr(owner_store, "acquire_authority", None) ): return None - if any( - value is not None - for value in ( - store, - tape_store, - checkpoint_store, - checkpoint_service, - runtime_store, - ) - ): - return None config = self._storage_config - sqlite_backend_keys = ( - "http_session_backend", - "tape_backend", - "checkpoint_backend", - "runtime_backend", - ) - if any( - str(config.get(key, "")).lower() != "sqlite" for key in sqlite_backend_keys + backend_values = durable_storage_backend_values(config) + if all(value == "pg" for value in backend_values.values()): + return None + sqlite_backend_keys = [ + key for key, value in backend_values.items() if value == "sqlite" + ] + if sqlite_backend_keys and len(sqlite_backend_keys) != len( + DURABLE_STORAGE_BACKEND_KEYS ): + mismatches = ", ".join( + f"{key}={config.get(key)!r}" + for key, value in backend_values.items() + if value != "sqlite" + ) + raise ConfigError( + "durable fencing requires all local sqlite backends when any " + f"local sqlite backend is configured; mismatched backends: {mismatches}" + ) + if not sqlite_backend_keys: + return None + # Custom stores own their path semantics; validate sqlite paths only when + # SessionManager will create the local durable bundle itself. + if self._custom_store_names: + logger.warning( + "durable fencing disabled: custom %s supplied", + ", ".join(self._custom_store_names), + ) return None local_path = normalize_storage_path( str(local_sqlite_path_from_storage_config(config, data_dir)) ) - path_keys = ( - "http_session_path", - "tape_path", - "checkpoint_path", - "runtime_path", - ) - configured_paths = [ - normalize_storage_path(str(config.get(key, ""))) for key in path_keys + configured_paths = { + key: normalize_storage_path(str(config.get(key, ""))) + for key in DURABLE_STORAGE_PATH_KEYS + } + path_mismatches = [ + f"{key}={config.get(key)!r}" + for key, path in configured_paths.items() + if path != local_path ] - if any(path != local_path for path in configured_paths): - return None + if path_mismatches: + mismatch_text = ", ".join(path_mismatches) + raise ConfigError( + "durable fencing requires sqlite storage paths to share " + f"{local_path}; mismatched paths: {mismatch_text}" + ) return SQLiteLocalDurableStore(local_path) def _configure_pg_durable_store_if_available(self) -> None: @@ -1252,24 +1281,17 @@ def _configure_pg_durable_store_if_available(self) -> None: getattr(self._owner_store, "acquire_authority", None) ): return - if ( - self._custom_store_supplied - or self._custom_tape_store_supplied - or self._custom_checkpoint_store_supplied - or self._custom_runtime_store_supplied - ): - return - pg_backend_keys = ( - "http_session_backend", - "tape_backend", - "checkpoint_backend", - "runtime_backend", - ) if any( str(self._storage_config.get(key, "")).strip().lower() != "pg" - for key in pg_backend_keys + for key in DURABLE_STORAGE_BACKEND_KEYS ): return + if self._custom_store_names: + logger.warning( + "durable fencing disabled: custom %s supplied", + ", ".join(self._custom_store_names), + ) + return pg_pool = self._get_pg_pool() durable_store = PGDurableStore(pool=pg_pool) self._pg_durable_store = durable_store @@ -3258,7 +3280,9 @@ async def replace_admitted_runtime(session: object) -> Session: model_name if model_name is not None else admitted_session.model_name ) if not resolved_model: - raise RuntimeError("session is missing model_name, cannot replace runtime") + raise RuntimeError( + "session is missing model_name, cannot replace runtime" + ) return await self._runtime_replacement_service.replace_runtime_config( admitted_session, model_name=resolved_model, diff --git a/coding-agent/tests/coding_agent/test_durable_fencing_adr_contract.py b/coding-agent/tests/coding_agent/test_durable_fencing_adr_contract.py index d51db2801..e2f8e31e3 100644 --- a/coding-agent/tests/coding_agent/test_durable_fencing_adr_contract.py +++ b/coding-agent/tests/coding_agent/test_durable_fencing_adr_contract.py @@ -38,7 +38,7 @@ def test_adr_0068_supersedes_adr_0067_without_conflicting_target_architecture() assert "**Status**: Superseded" in old assert "Superseded by ADR-0068" in old - assert "**Status**: Proposed" in new + assert "**Status**: Accepted" in new assert "Supersedes ADR-0067" in new stale_target_phrases = [ diff --git a/coding-agent/tests/coding_agent/test_pg_durable_fencing.py b/coding-agent/tests/coding_agent/test_pg_durable_fencing.py index 948239e66..8831e4520 100644 --- a/coding-agent/tests/coding_agent/test_pg_durable_fencing.py +++ b/coding-agent/tests/coding_agent/test_pg_durable_fencing.py @@ -26,6 +26,7 @@ SessionOwnerStore, SessionOwnershipConflictError, ) +from coding_agent.server.stores.session_store import InMemorySessionStore from coding_agent.server.stores.session_store import PGSessionMetadataStore @@ -152,6 +153,32 @@ async def close(self) -> None: return None +class FakePGTapeStore: + def __init__(self, *, pool: FakePGPool) -> None: + self.pool = pool + + async def load(self, tape_id: str) -> list[dict[str, object]]: + del tape_id + return [] + + async def save( + self, + tape_id: str, + entries: list[dict[str, object]], + ) -> None: + del tape_id, entries + + +class FakePGCheckpointStore: + def __init__(self, *, pool: FakePGPool) -> None: + self.pool = pool + + +class FakePGRuntimeStore: + def __init__(self, *, pool: FakePGPool) -> None: + self.pool = pool + + @pytest.mark.asyncio async def test_pg_owner_store_acquire_authority_allocates_db_epoch() -> None: pool = FakePGPool() @@ -469,30 +496,9 @@ async def test_pg_durable_create_run_requires_stable_tape_binding() -> None: assert not pool.connection.calls -def test_session_manager_enables_fenced_pg_wrappers_after_owner_store_config() -> None: - class FakePGTapeStore: - def __init__(self, *, pool: FakePGPool) -> None: - self.pool = pool - - async def load(self, tape_id: str) -> list[dict[str, object]]: - del tape_id - return [] - - async def save( - self, - tape_id: str, - entries: list[dict[str, object]], - ) -> None: - del tape_id, entries - - class FakePGCheckpointStore: - def __init__(self, *, pool: FakePGPool) -> None: - self.pool = pool - - class FakePGRuntimeStore: - def __init__(self, *, pool: FakePGPool) -> None: - self.pool = pool - +def test_session_manager_enables_fenced_pg_wrappers_after_owner_store_config( + caplog: pytest.LogCaptureFixture, +) -> None: pg_pool = FakePGPool() owner_store = SessionOwnerStore(pg_pool=cast(Any, pg_pool)) with ( @@ -507,6 +513,7 @@ def __init__(self, *, pool: FakePGPool) -> None: "coding_agent.server.session_manager.PGRuntimeStore", FakePGRuntimeStore, ), + caplog.at_level("WARNING", logger="coding_agent.server.session_manager"), ): create_store.return_value = PGSessionMetadataStore(pool=cast(Any, pg_pool)) manager = SessionManager( @@ -527,6 +534,48 @@ def __init__(self, *, pool: FakePGPool) -> None: assert isinstance(manager._tape_store, FencedPGTapeStore) assert isinstance(manager._checkpoint_service._store, FencedPGCheckpointStore) assert isinstance(manager._runtime_store, FencedPGRuntimeStore) + assert "durable fencing disabled" not in caplog.text + + +def test_session_manager_warns_when_custom_store_disables_pg_fencing( + caplog: pytest.LogCaptureFixture, +) -> None: + pg_pool = FakePGPool() + owner_store = SessionOwnerStore(pg_pool=cast(Any, pg_pool)) + + with ( + patch( + "coding_agent.server.session_manager._load_pg_storage_types", + return_value=( + lambda **_: pg_pool, + FakePGTapeStore, + FakePGCheckpointStore, + ), + ), + patch( + "coding_agent.server.session_manager.PGRuntimeStore", + FakePGRuntimeStore, + ), + caplog.at_level("WARNING", logger="coding_agent.server.session_manager"), + ): + manager = SessionManager( + storage_config={ + "http_session_backend": "pg", + "tape_backend": "pg", + "checkpoint_backend": "pg", + "runtime_backend": "pg", + "dsn": "postgresql://example", + }, + pg_pool=cast(Any, pg_pool), + store=InMemorySessionStore(), + owner_store=owner_store, + owner_id="server-a", + fencing_token=1, + ) + + assert manager._pg_durable_store is None + assert "durable fencing disabled: custom store supplied" in caplog.text + assert caplog.text.count("durable fencing disabled") == 1 def _sql_label(query: str) -> str: diff --git a/coding-agent/tests/coding_agent/test_sqlite_local_durable_fencing.py b/coding-agent/tests/coding_agent/test_sqlite_local_durable_fencing.py index ffdbf4bec..26b422796 100644 --- a/coding-agent/tests/coding_agent/test_sqlite_local_durable_fencing.py +++ b/coding-agent/tests/coding_agent/test_sqlite_local_durable_fencing.py @@ -8,8 +8,9 @@ from agentkit.checkpoint.models import CheckpointMeta, CheckpointSnapshot from agentkit.checkpoint import CheckpointService +from agentkit.errors import ConfigError from coding_agent.local_durable_store import SQLiteLocalDurableStore -from coding_agent.local_storage import local_sqlite_path +from coding_agent.local_storage import local_sqlite_path, local_sqlite_storage_config from coding_agent.runtime_store import AgentRunRecord, RuntimeEventRecord from coding_agent.server.session_manager import SessionManager from coding_agent.server.stores.session_owner_store import ( @@ -36,6 +37,15 @@ async def delete(self, checkpoint_id: str) -> None: del checkpoint_id +class FakeTapeStore: + async def save(self, tape_id: str, entries: list[dict[str, object]]) -> None: + del tape_id, entries + + async def load(self, tape_id: str) -> list[dict[str, object]]: + del tape_id + return [] + + @pytest.mark.asyncio async def test_sqlite_owner_authority_renew_does_not_advance_epoch( tmp_path: Path, @@ -102,6 +112,112 @@ async def test_sqlite_owner_authority_same_owner_expired_reacquire_advances_epoc assert reacquired.epoch == 2 +def test_session_manager_partial_sqlite_backend_fails_loudly( + tmp_path: Path, +) -> None: + config = local_sqlite_storage_config(tmp_path) + config["runtime_backend"] = "jsonl" + owner_store = SQLiteSessionOwnerStore(local_sqlite_path(tmp_path)) + + with pytest.raises(ConfigError) as exc_info: + SessionManager( + storage_config=config, + owner_store=owner_store, + owner_id="owner-a", + fencing_token=999, + ) + + message = str(exc_info.value) + assert "durable fencing requires all local sqlite backends" in message + assert "runtime_backend='jsonl'" in message + + +def test_session_manager_sqlite_path_mismatch_fails_loudly( + tmp_path: Path, +) -> None: + config = local_sqlite_storage_config(tmp_path) + config["runtime_path"] = str(tmp_path / "runtime.sqlite3") + owner_store = SQLiteSessionOwnerStore(local_sqlite_path(tmp_path)) + + with pytest.raises(ConfigError) as exc_info: + SessionManager( + storage_config=config, + owner_store=owner_store, + owner_id="owner-a", + fencing_token=999, + ) + + message = str(exc_info.value) + assert "durable fencing requires sqlite storage paths to share" in message + assert "runtime_path" in message + assert "runtime.sqlite3" in message + + +def test_session_manager_custom_tape_store_warns_and_disables_local_fencing( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + owner_store = SQLiteSessionOwnerStore(local_sqlite_path(tmp_path)) + + with caplog.at_level("WARNING", logger="coding_agent.server.session_manager"): + manager = SessionManager( + storage_config=local_sqlite_storage_config(tmp_path), + tape_store=FakeTapeStore(), + owner_store=owner_store, + owner_id="owner-a", + fencing_token=999, + ) + + assert manager._local_durable_store is None + assert "durable fencing disabled: custom tape_store supplied" in caplog.text + + +def test_session_manager_custom_tape_store_does_not_warn_for_non_sqlite_config( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + owner_store = SQLiteSessionOwnerStore(tmp_path / "owners.sqlite3") + + with caplog.at_level("WARNING", logger="coding_agent.server.session_manager"): + manager = SessionManager( + storage_config={ + "http_session_backend": "file", + "tape_backend": "jsonl", + "checkpoint_backend": "fs", + "runtime_backend": "jsonl", + }, + tape_store=FakeTapeStore(), + owner_store=owner_store, + owner_id="owner-a", + fencing_token=999, + ) + + assert manager._local_durable_store is None + assert "durable fencing disabled" not in caplog.text + + +def test_session_manager_partial_sqlite_with_custom_store_fails_loudly( + tmp_path: Path, + caplog: pytest.LogCaptureFixture, +) -> None: + config = local_sqlite_storage_config(tmp_path) + config["runtime_backend"] = "jsonl" + owner_store = SQLiteSessionOwnerStore(local_sqlite_path(tmp_path)) + + with caplog.at_level("WARNING", logger="coding_agent.server.session_manager"): + with pytest.raises(ConfigError) as exc_info: + SessionManager( + storage_config=config, + tape_store=FakeTapeStore(), + owner_store=owner_store, + owner_id="owner-a", + fencing_token=999, + ) + + assert "runtime_backend='jsonl'" in str(exc_info.value) + assert "durable fencing disabled" not in caplog.text + + @pytest.mark.asyncio async def test_session_manager_uses_sqlite_owner_authority_epochs( tmp_path: Path, @@ -147,6 +263,8 @@ async def test_session_manager_default_sqlite_bundle_uses_protected_session_save fencing_token=999, ) + assert manager._local_durable_store is not None + session_id = await manager.create_session() owner = await owner_store.get_owner(session_id) diff --git a/coding-agent/tests/ui/test_http_server.py b/coding-agent/tests/ui/test_http_server.py index 26fe298e4..de5027cee 100644 --- a/coding-agent/tests/ui/test_http_server.py +++ b/coding-agent/tests/ui/test_http_server.py @@ -60,6 +60,7 @@ LocalDaemonExecutorRef, LocalPathWorkspaceRef, RunTarget, + RuntimeContextBindingService, ) from coding_agent.wire.local import LocalWire from coding_agent.server.session_manager import Session, SessionManager @@ -177,8 +178,18 @@ def _test_runtime_profile_config(image: str = "python:3.11-slim") -> dict[str, o @pytest.fixture(autouse=True) -async def clear_sessions(): - """Clear sessions before each test.""" +async def clear_sessions(tmp_path: Path): + """Provide each test with an isolated unfenced session manager.""" + global session_manager + original_session_manager = session_manager + original_http_session_manager = http_server.session_manager + test_session_manager = SessionManager( + storage_config=local_sqlite_storage_config(tmp_path), + provisioned_cloud_binding_cleanup=http_server._cleanup_provisioned_cloud_binding, + ) + session_manager = test_session_manager + http_server.session_manager = test_session_manager + reset_prometheus_metrics() session_manager.configure_owner_leases( owner_store=None, @@ -190,28 +201,22 @@ async def clear_sessions(): session_manager.clear_sessions() # Clear rate limit storage to prevent 429 errors limiter.reset() - # Also close any sessions in session_manager - for session_id in list(session_manager.list_sessions()): - try: - await session_manager.close_session(session_id) - except Exception: - pass - yield - session_manager.configure_owner_leases( - owner_store=None, - owner_id=None, - fencing_token=None, - ) - session_manager.configure_workspace_metadata_store(None) - session_manager.configure_runtime_store(None) - session_manager.clear_sessions() - reset_prometheus_metrics() - # Cleanup session_manager - for session_id in list(session_manager.list_sessions()): - try: - await session_manager.close_session(session_id) - except Exception: - pass + try: + yield + finally: + session_manager.configure_owner_leases( + owner_store=None, + owner_id=None, + fencing_token=None, + ) + session_manager.configure_workspace_metadata_store(None) + session_manager.configure_runtime_store(None) + SessionManager.clear_sessions(test_session_manager) + reset_prometheus_metrics() + limiter.reset() + await SessionManager.close(test_session_manager) + session_manager = original_session_manager + http_server.session_manager = original_http_session_manager def register_session( @@ -2061,6 +2066,24 @@ def test_build_session_manager_enables_owner_store_for_local_sqlite_bundle( finally: asyncio.run(manager.close()) + def test_build_session_manager_enables_owner_store_for_default_sqlite_bundle( + self, monkeypatch, tmp_path + ): + monkeypatch.setenv("AGENT_DATA_DIR", str(tmp_path)) + monkeypatch.setattr( + "coding_agent.server.http_server._load_storage_config", + lambda: {}, + ) + + manager = _build_session_manager() + try: + assert manager._owner_store is not None + assert type(manager._owner_store).__name__ == "SQLiteSessionOwnerStore" + assert manager._owner_id is not None + assert manager._local_durable_store is not None + finally: + asyncio.run(manager.close()) + def test_build_session_manager_enables_owner_store_for_equivalent_local_sqlite_bundle_paths( self, monkeypatch, tmp_path ): @@ -3740,6 +3763,28 @@ async def fatal_publish_subagent_message( "publish_subagent_message", fatal_publish_subagent_message, ) + session_manager._runtime_context_binding_service = RuntimeContextBindingService( + publish_subagent_message=fatal_publish_subagent_message + ) + session_manager._local_daemon_runtime_preparation = replace( + session_manager._local_daemon_runtime_preparation, + bind_subagent_message_publisher=( + session_manager._runtime_context_binding_service.bind_subagent_message_publisher + ), + ) + session_manager._runtime_turn_service_factory = replace( + session_manager._runtime_turn_service_factory, + prepare_runtime=session_manager._local_daemon_runtime_preparation.prepare_runtime, + bind_root_run_identity=( + session_manager._runtime_context_binding_service.bind_root_run_identity + ), + bind_subagent_message_publisher=( + session_manager._runtime_context_binding_service.bind_subagent_message_publisher + ), + ) + session_manager._runtime_turn_service = ( + session_manager._build_runtime_turn_service() + ) session.runtime_pipeline = None session.runtime_ctx = None session.runtime_adapter = None @@ -8003,6 +8048,40 @@ def test_agent_toml_storage_paths_local_drives_sqlite_bundle( asyncio.run(manager.close()) +def test_build_session_manager_partial_sqlite_storage_fails_loudly( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + local_path = tmp_path / "local.sqlite3" + runtime_path = tmp_path / "runtime" + config_path = tmp_path / "agent.toml" + config_path.write_text( + _minimal_agent_toml( + f""" + +[storage] +http_session_backend = "sqlite" +http_session_path = "{local_path}" +tape_backend = "sqlite" +tape_path = "{local_path}" +checkpoint_backend = "sqlite" +checkpoint_path = "{local_path}" +runtime_backend = "jsonl" +runtime_path = "{runtime_path}" +""" + ), + encoding="utf-8", + ) + monkeypatch.setenv("CODING_AGENT_SERVER_CONFIG", str(config_path)) + + with pytest.raises(ConfigError) as exc_info: + http_server._build_session_manager() + + message = str(exc_info.value) + assert "durable fencing requires all local sqlite backends" in message + assert "runtime_backend='jsonl'" in message + + def test_http_server_import_raises_on_invalid_agent_toml(monkeypatch) -> None: original_module = sys.modules.get("coding_agent.server.http_server") server_module = sys.modules.get("coding_agent.server") diff --git a/coding-agent/tests/ui/test_http_server_failover.py b/coding-agent/tests/ui/test_http_server_failover.py index cecc05018..a1a323b4d 100644 --- a/coding-agent/tests/ui/test_http_server_failover.py +++ b/coding-agent/tests/ui/test_http_server_failover.py @@ -20,6 +20,9 @@ limiter, session_manager, ) +import coding_agent.server.http_server as http_server +from coding_agent.local_storage import local_sqlite_storage_config +from coding_agent.server.session_manager import SessionManager from coding_agent.server.stores.session_owner_store import SessionOwnerRecord from coding_agent.server.stores.session_owner_store import SessionOwnershipConflictError from coding_agent.server.stores.session_owner_store import ( @@ -85,12 +88,26 @@ async def release( @pytest.fixture(autouse=True) -async def clear_sessions(): - session_manager.clear_sessions() - limiter.reset() - yield +async def clear_sessions(tmp_path): + global session_manager + original_session_manager = session_manager + original_http_session_manager = http_server.session_manager + test_session_manager = SessionManager( + storage_config=local_sqlite_storage_config(tmp_path) + ) + session_manager = test_session_manager + http_server.session_manager = test_session_manager + session_manager.clear_sessions() limiter.reset() + try: + yield + finally: + SessionManager.clear_sessions(test_session_manager) + limiter.reset() + await SessionManager.close(test_session_manager) + session_manager = original_session_manager + http_server.session_manager = original_http_session_manager @pytest.fixture @@ -139,11 +156,13 @@ def __init__(self, body_iterator: AsyncIterator[dict[str, str]]): @pytest.mark.asyncio async def test_clear_sessions_fixture_resets_limiter_in_teardown( monkeypatch: pytest.MonkeyPatch, + tmp_path, ): session_clear_calls = 0 limiter_reset_calls = 0 - def fake_clear_sessions() -> None: + def fake_clear_sessions(self: SessionManager) -> None: + del self nonlocal session_clear_calls session_clear_calls += 1 @@ -151,14 +170,14 @@ def fake_limiter_reset() -> None: nonlocal limiter_reset_calls limiter_reset_calls += 1 - monkeypatch.setattr(session_manager, "clear_sessions", fake_clear_sessions) + monkeypatch.setattr(SessionManager, "clear_sessions", fake_clear_sessions) monkeypatch.setattr(limiter, "reset", fake_limiter_reset) fixture_factory = cast( - "Callable[[], AsyncGenerator[None, None]]", + "Callable[[object], AsyncGenerator[None, None]]", getattr(clear_sessions, "__wrapped__"), ) - fixture_gen = fixture_factory() + fixture_gen = fixture_factory(tmp_path) await anext(fixture_gen) assert session_clear_calls == 1 diff --git a/coding-agent/tests/ui/test_session_manager_runtime.py b/coding-agent/tests/ui/test_session_manager_runtime.py index 01cfe0977..5e1f5694e 100644 --- a/coding-agent/tests/ui/test_session_manager_runtime.py +++ b/coding-agent/tests/ui/test_session_manager_runtime.py @@ -618,7 +618,11 @@ async def test_replay_display_events_scans_past_internal_runtime_events() -> Non @pytest.mark.asyncio async def test_replay_display_events_skips_storeless_runtime() -> None: - manager = SessionManager(store=InMemorySessionStore(), runtime_store=None) + manager = SessionManager( + store=InMemorySessionStore(), + storage_config={"runtime_backend": "none"}, + runtime_store=None, + ) with pytest.raises(RuntimeError, match="runtime store is not configured"): await manager.replay_display_events("run-display") @@ -3084,7 +3088,9 @@ async def fake_restore_tape(tape_id: str): mp.setattr("coding_agent.__main__.create_agent", fake_create_agent) mp.setattr("coding_agent.server.session_manager.PipelineAdapter", FakeAdapter) mp.setattr(rehydrated, "_restore_tape", fake_restore_tape) - original_restore_tape = rehydrated._local_daemon_runtime_preparation.restore_tape + original_restore_tape = ( + rehydrated._local_daemon_runtime_preparation.restore_tape + ) object.__setattr__( rehydrated._local_daemon_runtime_preparation, "restore_tape", @@ -4385,9 +4391,7 @@ def run_command( @pytest.mark.asyncio -async def test_run_agent_does_not_bootstrap_managed_pool_runtime() -> ( - None -): +async def test_run_agent_does_not_bootstrap_managed_pool_runtime() -> None: runtime_store = FakeRuntimeStore() def fake_create_agent(**kwargs): @@ -4738,10 +4742,12 @@ def fail_cleanup(binding: CloudWorkspaceRef) -> None: provisioned_cloud_binding_cleanup=fail_cleanup, ) session_id = await manager.create_session( - default_run_target=_cloud_run_target(CloudWorkspaceRef( - workspace_url="docker://agent-ws-fails/workspace", - workspace_id="ws-fails", - )), + default_run_target=_cloud_run_target( + CloudWorkspaceRef( + workspace_url="docker://agent-ws-fails/workspace", + workspace_id="ws-fails", + ) + ), origin={ "channel": "http", "placement_kind": "cloud_workspace", @@ -4764,10 +4770,12 @@ async def test_close_session_keeps_explicit_cloud_binding_untouched() -> None: provisioned_cloud_binding_cleanup=cleaned.append, ) session_id = await manager.create_session( - default_run_target=_cloud_run_target(CloudWorkspaceRef( - workspace_url="https://workspace.example.com", - workspace_id="ws-explicit", - )), + default_run_target=_cloud_run_target( + CloudWorkspaceRef( + workspace_url="https://workspace.example.com", + workspace_id="ws-explicit", + ) + ), origin={"channel": "http", "placement_kind": "cloud_workspace"}, )