import asyncio from pathlib import Path from unittest.mock import AsyncMock, MagicMock, patch import pytest import free_claude_code.messaging.session.persistence as persistence_module from free_claude_code.config.admin.persistence import PreparedAdminUpdate from free_claude_code.config.settings import Settings from free_claude_code.messaging.command_context import StopOutcome from free_claude_code.messaging.platforms.ports import ( InboundMessageHandler, MessagingPlatformComponents, MessagingStartupNotice, ) from free_claude_code.messaging.session import SessionStore from free_claude_code.messaging.workflow import MessagingWorkflow from free_claude_code.providers.runtime import ProviderRuntime from free_claude_code.runtime.application import ApplicationRuntime from free_claude_code.runtime.provider_manager import ProviderRuntimeManager class TrackingRuntime(ProviderRuntime): def __init__(self, settings: Settings) -> None: super().__init__(settings) self.cleanup_calls = 0 async def cleanup(self) -> None: self.cleanup_calls += 1 await super().cleanup() class TrackingFactory: def __init__(self) -> None: self.runtimes: list[TrackingRuntime] = [] self.fail = False self.events: list[str] = [] def __call__(self, settings: Settings) -> ProviderRuntime: self.events.append(f"construct:{settings.model}") if self.fail: raise RuntimeError("candidate failed") runtime = TrackingRuntime(settings) self.runtimes.append(runtime) return runtime class TrackingTranscriber: def __init__(self, events: list[str]) -> None: self.events = events self.close_calls = 0 async def transcribe(self, file_path: Path) -> str: assert isinstance(file_path, Path) return "transcribed" async def close(self) -> None: self.close_calls += 1 self.events.append("transcriber.close") class FailingTranscriber(TrackingTranscriber): async def close(self) -> None: await super().close() raise RuntimeError("transcriber close failed") class CancelledTranscriber(TrackingTranscriber): async def close(self) -> None: await super().close() raise asyncio.CancelledError class CancellingOnceTranscriber(TrackingTranscriber): async def close(self) -> None: await super().close() if self.close_calls == 1: raise asyncio.CancelledError class TrackingMessagingRuntime: name = "tracking" def __init__( self, events: list[str], *, fail_quiesce_once: bool = False, fail_close_once: bool = False, ) -> None: self.events = events self.fail_quiesce_once = fail_quiesce_once self.fail_close_once = fail_close_once async def start(self) -> None: self.events.append("messaging.start") async def quiesce(self) -> None: self.events.append("messaging.quiesce") if self.fail_quiesce_once: self.fail_quiesce_once = False raise RuntimeError("quiesce failed") async def close(self) -> None: self.events.append("messaging.close") if self.fail_close_once: self.fail_close_once = False raise RuntimeError("close failed") def on_message(self, handler: InboundMessageHandler) -> None: assert callable(handler) @property def is_connected(self) -> bool: return True class PersistentlyFailingMessagingRuntime(TrackingMessagingRuntime): def __init__(self, events: list[str]) -> None: super().__init__(events) self.fail_quiesce = True async def quiesce(self) -> None: self.events.append("messaging.quiesce") if self.fail_quiesce: raise RuntimeError("quiesce failed") def _settings(model: str, *, port: int = 8082) -> Settings: return Settings().model_copy(update={"model": model, "port": port}) def _prepared( settings: Settings, tmp_path, *, pending_fields: tuple[str, ...] = (), ) -> PreparedAdminUpdate: return PreparedAdminUpdate( target_values={"MODEL": settings.model}, settings=settings, errors=(), pending_fields=pending_fields, path=tmp_path / ".env", ) def _applied_response(pending_fields: tuple[str, ...] = ()) -> dict[str, object]: return { "applied": True, "valid": True, "errors": [], "env_preview": "MODEL=updated\n", "path": ".env", "pending_fields": list(pending_fields), } @pytest.mark.asyncio async def test_stop_all_maps_messaging_outcome_to_application_count() -> None: manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) workflow = MagicMock() workflow.stop_all_tasks = AsyncMock( return_value=StopOutcome( cancelled_count=3, status_feedback_scopes=frozenset(), fallback_required=False, ) ) runtime._messaging_workflow = workflow result = await runtime.stop_all() assert result is not None assert result.cancelled_count == 3 workflow.stop_all_tasks.assert_awaited_once() await manager.close() @pytest.mark.asyncio async def test_provider_apply_constructs_before_commit_then_publishes(tmp_path) -> None: factory = TrackingFactory() manager = ProviderRuntimeManager( _settings("nvidia_nim/old"), runtime_factory=factory, ) runtime = ApplicationRuntime(manager, transcriber=None) prepared = _prepared(_settings("nvidia_nim/new"), tmp_path) factory.events.clear() def commit(_prepared_update: PreparedAdminUpdate) -> dict[str, object]: factory.events.append("commit") assert manager.current_generation_id == 1 return _applied_response() with ( patch( "free_claude_code.runtime.application.prepare_admin_update", return_value=prepared, ), patch( "free_claude_code.runtime.application.commit_prepared_admin_update", side_effect=commit, ), ): result = await runtime.apply_admin_config({"MODEL": "nvidia_nim/new"}) assert factory.events == ["construct:nvidia_nim/new", "commit"] assert manager.current_generation_id == 2 assert manager.current_settings().model == "nvidia_nim/new" assert result["restart"] == { "required": False, "automatic": False, "admin_url": None, "fields": [], } await manager.close() @pytest.mark.asyncio async def test_candidate_failure_never_commits_and_preserves_current(tmp_path) -> None: factory = TrackingFactory() manager = ProviderRuntimeManager( _settings("nvidia_nim/old"), runtime_factory=factory, ) runtime = ApplicationRuntime(manager, transcriber=None) prepared = _prepared(_settings("nvidia_nim/new"), tmp_path) factory.fail = True with ( patch( "free_claude_code.runtime.application.prepare_admin_update", return_value=prepared, ), patch( "free_claude_code.runtime.application.commit_prepared_admin_update" ) as commit, pytest.raises(RuntimeError, match="candidate failed"), ): await runtime.apply_admin_config({"MODEL": "nvidia_nim/new"}) commit.assert_not_called() assert manager.current_generation_id == 1 assert manager.current_settings().model == "nvidia_nim/old" await manager.close() @pytest.mark.asyncio async def test_persistence_failure_closes_candidate_and_preserves_current( tmp_path, ) -> None: factory = TrackingFactory() manager = ProviderRuntimeManager( _settings("nvidia_nim/old"), runtime_factory=factory, ) runtime = ApplicationRuntime(manager, transcriber=None) prepared = _prepared(_settings("nvidia_nim/new"), tmp_path) with ( patch( "free_claude_code.runtime.application.prepare_admin_update", return_value=prepared, ), patch( "free_claude_code.runtime.application.commit_prepared_admin_update", side_effect=OSError("disk full"), ), pytest.raises(OSError, match="disk full"), ): await runtime.apply_admin_config({"MODEL": "nvidia_nim/new"}) assert manager.current_generation_id == 1 assert factory.runtimes[0].cleanup_calls == 0 assert factory.runtimes[1].cleanup_calls == 1 await manager.close() @pytest.mark.asyncio async def test_restart_required_apply_commits_without_hot_publication(tmp_path) -> None: factory = TrackingFactory() manager = ProviderRuntimeManager( _settings("nvidia_nim/old"), runtime_factory=factory, ) restart = AsyncMock() runtime = ApplicationRuntime( manager, transcriber=None, restart_callback=restart, ) prepared = _prepared( _settings("nvidia_nim/old", port=9090), tmp_path, pending_fields=("PORT",), ) with ( patch( "free_claude_code.runtime.application.prepare_admin_update", return_value=prepared, ), patch( "free_claude_code.runtime.application.commit_prepared_admin_update", return_value=_applied_response(("PORT",)), ) as commit, ): result = await runtime.apply_admin_config({"PORT": "9090"}) commit.assert_called_once_with(prepared) assert manager.current_generation_id == 1 assert len(factory.runtimes) == 1 assert result["restart"] == { "required": True, "automatic": True, "admin_url": "http://127.0.0.1:9090/admin", "fields": ["PORT"], } restart.assert_not_awaited() await runtime.request_restart() restart.assert_awaited_once() await manager.close() @pytest.mark.asyncio async def test_close_drains_messaging_before_transcriber_and_is_idempotent() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) transcriber = TrackingTranscriber(events) runtime = ApplicationRuntime(manager, transcriber=transcriber) runtime._messaging_runtime = TrackingMessagingRuntime(events) workflow = MagicMock() workflow.close = AsyncMock(side_effect=lambda: events.append("workflow.close")) runtime._messaging_workflow = workflow runtime._cli_manager = MagicMock() assert runtime.is_closed is False assert await runtime.close() is True assert await runtime.close() is True assert events == [ "messaging.quiesce", "workflow.close", "messaging.close", "transcriber.close", ] assert transcriber.close_calls == 1 assert runtime._transcriber is None assert runtime._messaging_runtime is None assert runtime._messaging_workflow is None assert runtime.is_closed is True @pytest.mark.asyncio async def test_close_retains_transcriber_ownership_when_close_fails() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) transcriber = FailingTranscriber(events) runtime = ApplicationRuntime(manager, transcriber=transcriber) runtime._messaging_runtime = TrackingMessagingRuntime(events) assert await runtime.close() is False assert events == [ "messaging.quiesce", "messaging.close", "transcriber.close", ] assert transcriber.close_calls == 1 assert runtime._transcriber is transcriber assert runtime._closed is False await manager.close() @pytest.mark.asyncio async def test_close_retries_runtime_before_closing_later_resources() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) transcriber = TrackingTranscriber(events) runtime = ApplicationRuntime(manager, transcriber=transcriber) messaging = TrackingMessagingRuntime(events, fail_close_once=True) runtime._messaging_runtime = messaging assert await runtime.close() is False assert events == ["messaging.quiesce", "messaging.close"] assert runtime._messaging_runtime is messaging assert runtime._transcriber is transcriber assert runtime._closed is False assert await runtime.close() is True assert events == [ "messaging.quiesce", "messaging.close", "messaging.quiesce", "messaging.close", "transcriber.close", ] assert runtime._messaging_runtime is None assert runtime._transcriber is None assert runtime._closed is True @pytest.mark.asyncio async def test_close_retries_workflow_close_before_closing_delivery() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) messaging = TrackingMessagingRuntime(events) workflow = MagicMock() close_calls = 0 async def close_workflow() -> None: nonlocal close_calls close_calls += 1 if close_calls == 1: raise RuntimeError("drain failed") events.append("workflow.close") workflow.close = AsyncMock(side_effect=close_workflow) runtime._messaging_runtime = messaging runtime._messaging_workflow = workflow assert await runtime.close() is False assert events == ["messaging.quiesce"] assert runtime._messaging_runtime is messaging assert runtime._messaging_workflow is workflow assert runtime._closed is False assert await runtime.close() is True assert events == [ "messaging.quiesce", "messaging.quiesce", "workflow.close", "messaging.close", ] assert runtime._closed is True @pytest.mark.asyncio async def test_close_does_not_drain_workflow_until_ingress_is_quiescent() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) messaging = TrackingMessagingRuntime(events, fail_quiesce_once=True) workflow = MagicMock() workflow.close = AsyncMock(side_effect=lambda: events.append("workflow.close")) runtime._messaging_runtime = messaging runtime._messaging_workflow = workflow assert await runtime.close() is False workflow.close.assert_not_awaited() assert runtime._messaging_runtime is messaging assert runtime._messaging_workflow is workflow assert await runtime.close() is True assert events == [ "messaging.quiesce", "messaging.quiesce", "workflow.close", "messaging.close", ] assert runtime._closed is True @pytest.mark.asyncio async def test_close_retries_failed_persistence_before_closing_delivery() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) messaging = TrackingMessagingRuntime(events) workflow = MagicMock() close_calls = 0 async def close_workflow() -> None: nonlocal close_calls close_calls += 1 if close_calls == 1: raise RuntimeError("flush failed") events.append("workflow.close") workflow.close = AsyncMock(side_effect=close_workflow) runtime._messaging_runtime = messaging runtime._messaging_workflow = workflow await runtime.close() assert runtime._messaging_workflow is workflow assert runtime._messaging_runtime is messaging assert "messaging.close" not in events await runtime.close() assert events == [ "messaging.quiesce", "messaging.quiesce", "workflow.close", "messaging.close", ] assert runtime._closed is True @pytest.mark.asyncio async def test_close_retries_real_workflow_persistence_without_losing_latest_state( tmp_path: Path, ) -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) messaging = TrackingMessagingRuntime(events) cli_manager = MagicMock() cli_manager.stop_all = AsyncMock() outbound = MagicMock() store_path = tmp_path / "sessions.json" real_replace = persistence_module.os.replace replace_calls = 0 def fail_first_replace(source: str, target: str) -> None: nonlocal replace_calls replace_calls += 1 if replace_calls == 1: raise OSError("replace failed once") real_replace(source, target) with patch.object(persistence_module.threading, "Timer"): store = SessionStore(storage_path=str(store_path)) store.record_message_id( "telegram", "chat_1", "old", "out", "status", ) store.flush_pending_save() store.record_message_id( "telegram", "chat_1", "latest", "out", "status", ) workflow = MessagingWorkflow(outbound, cli_manager, store) runtime._messaging_runtime = messaging runtime._messaging_workflow = workflow runtime._cli_manager = cli_manager with patch.object( persistence_module.os, "replace", side_effect=fail_first_replace, ): assert await runtime.close() is False assert runtime._messaging_runtime is messaging assert runtime._messaging_workflow is workflow assert runtime._cli_manager is cli_manager assert runtime.is_closed is False assert store.dirty is True assert store.get_tracked_message_ids_for_chat("telegram", "chat_1") == [ "old", "latest", ] assert SessionStore( storage_path=str(store_path) ).get_tracked_message_ids_for_chat("telegram", "chat_1") == ["old"] assert await runtime.close() is True assert runtime._messaging_runtime is None assert runtime._messaging_workflow is None assert runtime._cli_manager is None assert runtime.is_closed is True assert store.dirty is False assert SessionStore( storage_path=str(store_path) ).get_tracked_message_ids_for_chat("telegram", "chat_1") == [ "old", "latest", ] assert events == [ "messaging.quiesce", "messaging.quiesce", "messaging.close", ] @pytest.mark.asyncio async def test_cancelled_transcriber_close_retains_ownership() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) transcriber = CancelledTranscriber(events) runtime = ApplicationRuntime(manager, transcriber=transcriber) with pytest.raises(asyncio.CancelledError): await runtime._cleanup_transcriber() assert transcriber.close_calls == 1 assert runtime._transcriber is transcriber await manager.close() @pytest.mark.asyncio async def test_cancelled_application_close_remains_retryable() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) transcriber = CancellingOnceTranscriber(events) runtime = ApplicationRuntime(manager, transcriber=transcriber) with pytest.raises(asyncio.CancelledError): await runtime.close() assert runtime._closed is False assert runtime._transcriber is transcriber await runtime.close() assert transcriber.close_calls == 2 assert runtime._transcriber is None assert runtime._closed is True @pytest.mark.asyncio async def test_startup_failure_closes_owned_transcriber() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) transcriber = TrackingTranscriber(events) runtime = ApplicationRuntime(manager, transcriber=transcriber) with ( patch.object( manager, "validate_configured_models", AsyncMock(side_effect=RuntimeError("startup failed")), ), pytest.raises(RuntimeError, match="startup failed"), ): await runtime.start() assert transcriber.close_calls == 1 @pytest.mark.asyncio async def test_startup_cancellation_cleans_partial_messaging_and_reraises() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) transcriber = TrackingTranscriber(events) runtime = ApplicationRuntime(manager, transcriber=transcriber) messaging = TrackingMessagingRuntime(events) entered = asyncio.Event() async def start_messaging() -> None: runtime._messaging_runtime = messaging entered.set() await asyncio.Event().wait() with patch.object( runtime, "_start_messaging_if_configured", side_effect=start_messaging, ): start_task = asyncio.create_task(runtime.start()) await entered.wait() start_task.cancel() with pytest.raises(asyncio.CancelledError): await start_task assert events == [ "messaging.quiesce", "messaging.close", "transcriber.close", ] assert runtime._closed is True assert runtime._messaging_runtime is None assert runtime._transcriber is None @pytest.mark.asyncio async def test_public_start_retries_transient_partial_messaging_cleanup() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) messaging = TrackingMessagingRuntime(events, fail_quiesce_once=True) workflow = MagicMock() workflow.close = AsyncMock(side_effect=lambda: events.append("workflow.close")) cli_manager = MagicMock() components = MessagingPlatformComponents( name="tracking", runtime=messaging, outbound=MagicMock(), ) startup_failure = RuntimeError("partial messaging startup failed") async def fail_after_publication( published: MessagingPlatformComponents, ) -> None: assert published is components runtime._messaging_runtime = messaging runtime._messaging_workflow = workflow runtime._cli_manager = cli_manager raise startup_failure with ( patch.object( runtime, "_validate_configured_models_best_effort", AsyncMock(), ), patch.object(manager, "start_model_list_refresh"), patch( "free_claude_code.runtime.application.messaging_platform_factory.create_messaging_components", return_value=components, ), patch.object( runtime, "_start_messaging_workflow", side_effect=fail_after_publication, ), pytest.raises(RuntimeError, match="cleanup incomplete") as raised, ): await runtime.start() assert raised.value.__cause__ is startup_failure assert events == [ "messaging.quiesce", "messaging.quiesce", "workflow.close", "messaging.close", ] workflow.close.assert_awaited_once() assert runtime._messaging_runtime is None assert runtime._messaging_workflow is None assert runtime._cli_manager is None assert runtime.is_closed is True @pytest.mark.asyncio async def test_public_start_retains_persistently_unclean_partial_messaging_graph() -> ( None ): events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) transcriber = TrackingTranscriber(events) runtime = ApplicationRuntime(manager, transcriber=transcriber) messaging = PersistentlyFailingMessagingRuntime(events) workflow = MagicMock() workflow.close = AsyncMock(side_effect=lambda: events.append("workflow.close")) cli_manager = MagicMock() components = MessagingPlatformComponents( name="tracking", runtime=messaging, outbound=MagicMock(), ) startup_failure = RuntimeError("partial messaging startup failed") async def fail_after_publication( published: MessagingPlatformComponents, ) -> None: assert published is components runtime._messaging_runtime = messaging runtime._messaging_workflow = workflow runtime._cli_manager = cli_manager raise startup_failure with ( patch.object( runtime, "_validate_configured_models_best_effort", AsyncMock(), ), patch.object(manager, "start_model_list_refresh"), patch( "free_claude_code.runtime.application.messaging_platform_factory.create_messaging_components", return_value=components, ), patch.object( runtime, "_start_messaging_workflow", side_effect=fail_after_publication, ), pytest.raises(RuntimeError, match="cleanup incomplete") as raised, ): await runtime.start() assert raised.value.__cause__ is startup_failure assert events == ["messaging.quiesce", "messaging.quiesce"] workflow.close.assert_not_awaited() assert transcriber.close_calls == 0 assert runtime._messaging_runtime is messaging assert runtime._messaging_workflow is workflow assert runtime._cli_manager is cli_manager assert runtime._transcriber is transcriber assert runtime._provider_manager_closed is False assert runtime.is_closed is False messaging.fail_quiesce = False assert await runtime.close() is True assert events == [ "messaging.quiesce", "messaging.quiesce", "messaging.quiesce", "workflow.close", "messaging.close", "transcriber.close", ] assert runtime.is_closed is True @pytest.mark.asyncio async def test_messaging_start_failure_is_nonfatal_after_complete_cleanup() -> None: manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) with ( patch( "free_claude_code.runtime.application.messaging_platform_factory.create_messaging_components", side_effect=RuntimeError("messaging unavailable"), ), patch.object(runtime, "_cleanup_messaging", AsyncMock(return_value=True)), ): await runtime._start_messaging_if_configured() await manager.close() @pytest.mark.asyncio async def test_messaging_start_failure_fails_closed_when_cleanup_is_incomplete() -> ( None ): manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) with ( patch( "free_claude_code.runtime.application.messaging_platform_factory.create_messaging_components", side_effect=RuntimeError("messaging unavailable"), ), patch.object(runtime, "_cleanup_messaging", AsyncMock(return_value=False)), pytest.raises(RuntimeError, match="cleanup incomplete") as exc_info, ): await runtime._start_messaging_if_configured() assert isinstance(exc_info.value.__cause__, RuntimeError) await manager.close() @pytest.mark.asyncio async def test_composition_records_runtime_before_workspace_setup() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) messaging = TrackingMessagingRuntime(events) components = MessagingPlatformComponents( name="tracking", runtime=messaging, outbound=MagicMock(), ) with ( patch( "free_claude_code.runtime.application.os.makedirs", side_effect=OSError("workspace failed"), ), pytest.raises(OSError, match="workspace failed"), ): await runtime._start_messaging_workflow(components) assert runtime._messaging_runtime is messaging await runtime.close() assert events == ["messaging.quiesce", "messaging.close"] assert runtime._closed is True @pytest.mark.asyncio async def test_composition_publishes_startup_notice_after_runtime_and_repair() -> None: events: list[str] = [] manager = ProviderRuntimeManager(_settings("nvidia_nim/model")) runtime = ApplicationRuntime(manager, transcriber=None) messaging = TrackingMessagingRuntime(events) notice = MessagingStartupNotice( chat_id="chat", transport_label="test transport", ) components = MessagingPlatformComponents( name="tracking", runtime=messaging, outbound=MagicMock(), startup_notice=notice, ) workflow = MagicMock() workflow.handle_message = AsyncMock() workflow.restore.side_effect = lambda: events.append("workflow.restore") workflow.repair_restored_statuses = AsyncMock( side_effect=lambda: events.append("workflow.repair") ) workflow.publish_startup_notice = AsyncMock( side_effect=lambda published: events.append("workflow.notice") ) workflow.close = AsyncMock() cli_manager = MagicMock() with ( patch( "free_claude_code.runtime.application.cli_managed.ManagedClaudeSessionManager", return_value=cli_manager, ) as manager_constructor, patch("free_claude_code.runtime.application.messaging_session.SessionStore"), patch( "free_claude_code.runtime.application.messaging_workflow_module.MessagingWorkflow", return_value=workflow, ), ): await runtime._start_messaging_workflow(components) assert events == [ "workflow.restore", "messaging.start", "workflow.repair", "workflow.notice", ] workflow.publish_startup_notice.assert_awaited_once_with(notice) assert manager_constructor.call_args.kwargs["proxy_root_url"] == ( "http://127.0.0.1:8082" ) assert "api_url" not in manager_constructor.call_args.kwargs assert "plans_directory" not in manager_constructor.call_args.kwargs assert await runtime.close() is True