diff --git a/src/free_claude_code/api/handlers/messages.py b/src/free_claude_code/api/handlers/messages.py index 505a25652a..e5eec9f829 100644 --- a/src/free_claude_code/api/handlers/messages.py +++ b/src/free_claude_code/api/handlers/messages.py @@ -121,7 +121,7 @@ async def create( result = _MessagesStreamResult( self._provider_executor.stream_messages( routed, - raw_log_payload=routed.request.model_dump(), + raw_log_payload=routed.request.model_dump, request_id=request_id, ) ) diff --git a/src/free_claude_code/api/handlers/responses.py b/src/free_claude_code/api/handlers/responses.py index 8c10f100b7..f0d231bb74 100644 --- a/src/free_claude_code/api/handlers/responses.py +++ b/src/free_claude_code/api/handlers/responses.py @@ -62,7 +62,6 @@ async def create( ) -> object: """Create a streaming OpenAI Responses-compatible response.""" request_id = request_id or new_request_id() - request_payload = request_data.model_dump(mode="json", exclude_none=True) if request_data.stream is False: raise InvalidRequestError( "FCC /v1/responses supports streaming only; omit stream or set stream=true." @@ -84,7 +83,9 @@ async def create( ) streamed = self._provider_executor.stream_responses( routed, - raw_log_payload=request_payload, + raw_log_payload=lambda: request_data.model_dump( + mode="json", exclude_none=True + ), request_id=request_id, ) return await openai_responses_sse_streaming_response( diff --git a/src/free_claude_code/api/handlers/token_count.py b/src/free_claude_code/api/handlers/token_count.py index f2bea621a4..d9bc638dfa 100644 --- a/src/free_claude_code/api/handlers/token_count.py +++ b/src/free_claude_code/api/handlers/token_count.py @@ -60,16 +60,19 @@ def count( provider_model_ref=routed.resolved.primary.provider_model_ref, gateway_model=routed.resolved.original_model, ) - request_snapshot = anthropic_request_snapshot(routed.request) - request_snapshot["model"] = routed.resolved.original_model trace_event( + lambda: { + "snapshot": { + **anthropic_request_snapshot(routed.request), + "model": routed.resolved.original_model, + } + }, stage="ingress", event="free_claude_code.api.count_tokens.completed", source="api", request_id=request_id, message_count=len(routed.request.messages), input_tokens=tokens, - snapshot=request_snapshot, ) return TokenCountResponse(input_tokens=tokens) except ApplicationError: diff --git a/src/free_claude_code/api/response_streams.py b/src/free_claude_code/api/response_streams.py index 2a1777af1e..9c3ea29d85 100644 --- a/src/free_claude_code/api/response_streams.py +++ b/src/free_claude_code/api/response_streams.py @@ -10,6 +10,7 @@ from typing import Literal from fastapi.responses import JSONResponse, Response, StreamingResponse +from loguru import logger from starlette.background import BackgroundTask from starlette.responses import ContentStream from starlette.types import Receive, Scope, Send @@ -114,7 +115,7 @@ async def _cleanup(self, *, preserved_error: BaseException | None) -> None: preserved_error=preserved_error, ) except Exception as exc: - _trace_response_cleanup_failure("close_body", exc) + _log_response_cleanup_failure("close_body", exc) release = self._release if release is None: @@ -122,7 +123,7 @@ async def _cleanup(self, *, preserved_error: BaseException | None) -> None: try: await release() except Exception as exc: - _trace_response_cleanup_failure("release_resource", exc) + _log_response_cleanup_failure("release_resource", exc) async def _wait_for_cleanup(task: asyncio.Task[None]) -> None: @@ -134,7 +135,7 @@ async def _wait_for_cleanup(task: asyncio.Task[None]) -> None: except asyncio.CancelledError as exc: cancellation = exc - # Ordinary defensive failures are trace-only; cancellation remains control flow. + # Ordinary cleanup failures preserve the response; cancellation remains control flow. try: task.result() except asyncio.CancelledError: @@ -142,20 +143,20 @@ async def _wait_for_cleanup(task: asyncio.Task[None]) -> None: raise cancellation from None raise except Exception as exc: - _trace_response_cleanup_failure("cleanup_task", exc) + _log_response_cleanup_failure("cleanup_task", exc) if cancellation is not None: raise cancellation -def _trace_response_cleanup_failure(operation: str, exc: BaseException) -> None: - trace_event( +def _log_response_cleanup_failure(operation: str, exc: BaseException) -> None: + logger.bind( stage="egress", event="free_claude_code.api.response.cleanup_failed", source="api", operation=operation, exc_type=type(exc).__name__, - ) + ).opt(exception=exc).warning("Response cleanup failed") async def bind_response_lifetime( diff --git a/src/free_claude_code/application/code_sessions/service.py b/src/free_claude_code/application/code_sessions/service.py index b8ec855df6..3651e3e44d 100644 --- a/src/free_claude_code/application/code_sessions/service.py +++ b/src/free_claude_code/application/code_sessions/service.py @@ -183,9 +183,10 @@ async def _close(self) -> None: await self._stop( owner.state.session.id, owner.state.run.id, shutting_down=True ) - except Exception: + except Exception as exc: + self._mark_storage_failed(owner, "stop_run", exc) await self._close_connection(owner) - self._storage_failure(owner) + self._notify_storage_failure(owner) jobs = [owner.job for owner in owners if owner.job is not None] if jobs: await asyncio.wait(jobs, timeout=5) @@ -199,8 +200,9 @@ async def _close(self) -> None: "interrupted", "FCC stopped before this turn finished.", ) - except Exception: - self._storage_failure(owner) + except Exception as exc: + self._mark_storage_failed(owner, "finish_run", exc) + self._notify_storage_failure(owner) if self._jobs: await asyncio.gather(*tuple(self._jobs), return_exceptions=True) self._events.close() @@ -978,8 +980,8 @@ async def _commit_progress( ) except CodeConflictError, CodeNotFoundError: raise - except Exception: - owner.storage_failed = True + except Exception as exc: + self._mark_storage_failed(owner, "save_progress", exc) if owner.failure_task is None: owner.failure_task = self._job(self._halt_storage(owner)) raise @@ -987,7 +989,7 @@ async def _commit_progress( async def _halt_storage(self, owner: _SessionRuntime) -> None: await self._close_connection(owner) - self._storage_failure(owner) + self._notify_storage_failure(owner) def _detach_connection_locked( self, owner: _SessionRuntime @@ -1017,8 +1019,9 @@ async def _close_connection(self, owner: _SessionRuntime) -> None: owner, owner.state.attach_thread(connection.thread_id) ) await self._expire_prompts_locked(owner) - except Exception: - self._storage_failure(owner) + except Exception as exc: + self._mark_storage_failed(owner, "close_connection", exc) + self._notify_storage_failure(owner) async def _fail( self, @@ -1049,11 +1052,24 @@ async def _fail( else "failed" ) await self._finish_locked(owner, status, message) - except Exception: - self._storage_failure(owner) + except Exception as exc: + self._mark_storage_failed(owner, "fail_run", exc) + self._notify_storage_failure(owner) - def _storage_failure(self, owner: _SessionRuntime) -> None: + def _mark_storage_failed( + self, owner: _SessionRuntime, operation: str, error: Exception + ) -> None: + if owner.storage_failed: + return owner.storage_failed = True + logger.bind( + event="code.storage_failed", + operation=operation, + session_id=owner.state.session.id, + run_id=owner.state.run.id if owner.state.run is not None else None, + ).opt(exception=error).error("Code session stopped after a storage failure") + + def _notify_storage_failure(self, owner: _SessionRuntime) -> None: owner.finished.set() self._publish( owner, @@ -1104,8 +1120,11 @@ async def _delete_native( progress = owner.state.deletion_failed(status, _error_message(exc)) try: await self._commit_progress(owner, progress) - except Exception: - self._storage_failure(owner) + except Exception as save_error: + self._mark_storage_failed( + owner, "save_deletion_failure", save_error + ) + self._notify_storage_failure(owner) return self._publish(owner, "session.updated") if status == "delete_uncertain" and not reconcile and not native_complete: diff --git a/src/free_claude_code/application/execution.py b/src/free_claude_code/application/execution.py index 03c9582bc9..1796f201e4 100644 --- a/src/free_claude_code/application/execution.py +++ b/src/free_claude_code/application/execution.py @@ -167,7 +167,7 @@ def stream_messages( self, routed: RoutedMessagesRequest, *, - raw_log_payload: object, + raw_log_payload: Callable[[], object], request_id: str, ) -> AsyncIterator[str]: """Execute one Anthropic Messages request.""" @@ -210,7 +210,7 @@ async def open_candidate( wire_api="messages", raw_log_label="FULL_PAYLOAD", raw_log_payload=raw_log_payload, - request_snapshot=anthropic_request_snapshot(routed.request), + request_snapshot=lambda: anthropic_request_snapshot(routed.request), ingress_count_name="message_count", ingress_count=len(routed.request.messages), request_id=request_id, @@ -221,7 +221,7 @@ def stream_responses( self, routed: RoutedResponsesRequest, *, - raw_log_payload: object, + raw_log_payload: Callable[[], object], request_id: str, ) -> AsyncIterator[str]: """Execute one native OpenAI Responses request.""" @@ -266,7 +266,7 @@ async def open_candidate( wire_api="responses", raw_log_label="FULL_RESPONSES_PAYLOAD", raw_log_payload=raw_log_payload, - request_snapshot={ + request_snapshot=lambda: { "model": routed.request.model, "input_item_count": input_item_count, "tool_count": len(routed.request.tools or ()), @@ -284,8 +284,8 @@ def _stream_candidates( reasoning: ReasoningPolicy, wire_api: WireApi, raw_log_label: str, - raw_log_payload: object, - request_snapshot: dict[str, object], + raw_log_payload: Callable[[], object], + request_snapshot: Callable[[], dict[str, object]], ingress_count_name: str, ingress_count: int, request_id: str, @@ -318,7 +318,6 @@ def _stream_candidates( route_trace["generation_id"] = self._generation_id trace_event(**route_trace) - request_snapshot["model"] = gateway_model ingress_trace: dict[str, object] = { "stage": "ingress", "event": ( @@ -327,16 +326,18 @@ def _stream_candidates( else "free_claude_code.api.request.received" ), "source": "api", - "snapshot": request_snapshot, "request_id": request_id, ingress_count_name: ingress_count, } trace_event( + lambda: {"snapshot": {**request_snapshot(), "model": gateway_model}}, **ingress_trace, ) if self._log_raw_payloads: - logger.debug(f"{raw_log_label} [{{}}]: {{}}", request_id, raw_log_payload) + logger.opt(lazy=True).debug( + f"{raw_log_label} [{{}}]: {{}}", lambda: request_id, raw_log_payload + ) async def provider_body() -> AsyncIterator[str]: loop = asyncio.get_running_loop() diff --git a/src/free_claude_code/application/web_tools/service.py b/src/free_claude_code/application/web_tools/service.py index 2897757e08..7b7924f902 100644 --- a/src/free_claude_code/application/web_tools/service.py +++ b/src/free_claude_code/application/web_tools/service.py @@ -138,7 +138,7 @@ async def _stream_automatic_search( ) provider_stream = self._executor.stream_messages( translated, - raw_log_payload=plan.request.model_dump(), + raw_log_payload=plan.request.model_dump, request_id=request_id, ) chunks: list[str] = [] diff --git a/src/free_claude_code/cli/commands.py b/src/free_claude_code/cli/commands.py index 5421a461cd..25b8aa57ad 100644 --- a/src/free_claude_code/cli/commands.py +++ b/src/free_claude_code/cli/commands.py @@ -35,20 +35,23 @@ def _start_admin_browser( - settings: Settings, eligible: Callable[[], bool] + settings: Settings, eligible: Callable[[], bool], *, instance_id: str ) -> threading.Event: """Hand off an optional browser action without keeping FCC alive.""" completed = threading.Event() url = local_admin_url(settings) + browser_logger = logger.bind(instance_id=instance_id) def open_browser() -> None: try: if eligible() and not webbrowser.open(url): - logger.warning( + browser_logger.warning( "Could not open Admin in a browser. Open {} manually.", url ) except Exception as exc: - logger.warning("Could not open Admin: {}. Open {} manually.", exc, url) + browser_logger.warning( + "Could not open Admin: {}. Open {} manually.", exc, url + ) finally: completed.set() @@ -57,7 +60,7 @@ def open_browser() -> None: target=open_browser, name="fcc-open-admin-browser", daemon=True ).start() except Exception as exc: - logger.warning( + browser_logger.warning( "Could not start the Admin browser: {}. Open {} manually.", exc, url ) completed.set() @@ -93,6 +96,7 @@ def __init__(self, *, console_logging: bool = True) -> None: self._running = False self.stop_event = threading.Event() self._ready_settings: Settings | None = None + self._ready_instance_id: str | None = None self._pending_admin = False self._auto_browser_opened = False self._owned_server = False @@ -211,11 +215,14 @@ def request_open_admin(self) -> None: return self._pending_admin = True settings = self._ready_settings + instance_id = self._ready_instance_id generation = self._restart_generation - if settings is not None: - self._open_admin(settings, generation) + if settings is not None and instance_id is not None: + self._open_admin(settings, generation, instance_id) - def _open_admin(self, settings: Settings, generation: int) -> None: + def _open_admin( + self, settings: Settings, generation: int, instance_id: str + ) -> None: def eligible() -> bool: with self._lock: if ( @@ -227,7 +234,7 @@ def eligible() -> bool: self._pending_admin = False return True - _start_admin_browser(settings, eligible) + _start_admin_browser(settings, eligible, instance_id=instance_id) def _run_once( self, @@ -283,12 +290,15 @@ def on_started() -> None: ): return self._ready_settings = settings + self._ready_instance_id = asgi_app.runtime.instance_id should_open = open_admin_browser or self._pending_admin if open_admin_browser: self._auto_browser_opened = True asgi_app.runtime.http_started() if should_open: - self._open_admin(settings, restart_generation) + self._open_admin( + settings, restart_generation, asgi_app.runtime.instance_id + ) server = RuntimeServer( config, @@ -305,12 +315,14 @@ def on_started() -> None: server.should_exit = True try: - server.run(sockets=sockets) + with logger.contextualize(instance_id=asgi_app.runtime.instance_id): + server.run(sockets=sockets) finally: with self._lock: if self._server is server: self._server = None self._ready_settings = None + self._ready_instance_id = None with self._lock: restart_requested = self._restart_generation != restart_generation @@ -361,7 +373,11 @@ def open_admin_when_ready( ): return False if payload["status"] == "running" and not stop.is_set(): - completed = _start_admin_browser(settings, lambda: not stop.is_set()) + completed = _start_admin_browser( + settings, + lambda: not stop.is_set(), + instance_id=payload["instance_id"], + ) # This extra launcher is about to exit: allow a brief URL handoff. handoff_deadline = time.monotonic() + _BROWSER_HANDOFF_SECONDS while not stop.is_set(): diff --git a/src/free_claude_code/core/trace.py b/src/free_claude_code/core/trace.py index 4d3bafeff7..6f0cf0bb25 100644 --- a/src/free_claude_code/core/trace.py +++ b/src/free_claude_code/core/trace.py @@ -1,4 +1,4 @@ -"""Structured DEBUG traces for end-to-end request / CLI / provider logging. +"""Structured DEBUG traces and stream cleanup warnings. Trace payloads are preserved in ``record.extra.trace_payload`` by Loguru. Conversation and Claude Code prompts are logged verbatim unless values live under @@ -8,7 +8,7 @@ import asyncio import sys -from collections.abc import AsyncGenerator, AsyncIterator, Mapping +from collections.abc import AsyncGenerator, AsyncIterator, Callable, Mapping from typing import Any from loguru import logger @@ -49,17 +49,29 @@ def sanitize_trace_value(obj: Any) -> Any: return obj -def trace_event(*, stage: str, event: str, source: str, **fields: Any) -> None: - """Emit a DEBUG record with a structured payload bound to its extra fields.""" - payload = sanitize_trace_value( - { - "stage": stage, - "event": event, - "source": source, - **fields, - }, +def trace_event( + lazy_fields: Callable[[], Mapping[str, Any]] | None = None, + /, + *, + stage: str, + event: str, + source: str, + **fields: Any, +) -> None: + """Defer sanitization and the optional field factory until DEBUG is enabled.""" + logger.opt(lazy=True).debug( + "TRACE {}", + lambda: event, + trace_payload=lambda: sanitize_trace_value( + { + "stage": stage, + "event": event, + "source": source, + **fields, + **(lazy_fields() if lazy_fields is not None else {}), + } + ), ) - logger.bind(trace_payload=payload).debug("TRACE {}", event) async def close_stream_input( @@ -73,7 +85,7 @@ async def close_stream_input( close_error = await try_close_async_iterator(iterator) if close_error is None: return - trace_event( + logger.bind( stage="lifecycle", event="stream.input.close_failed", source=source, @@ -82,7 +94,7 @@ async def close_stream_input( preserved_exc_type=( type(preserved_error).__name__ if preserved_error is not None else None ), - ) + ).opt(exception=close_error).warning("Stream input cleanup failed") def extract_claude_session_id_from_headers(headers: Mapping[str, str]) -> str | None: diff --git a/src/free_claude_code/harnesses/claude_desktop_integration.py b/src/free_claude_code/harnesses/claude_desktop_integration.py index f29f5772c8..129612ee8d 100644 --- a/src/free_claude_code/harnesses/claude_desktop_integration.py +++ b/src/free_claude_code/harnesses/claude_desktop_integration.py @@ -295,6 +295,8 @@ def _update_profile(library: _Library, values: JsonObject) -> bool: } headers[_VIEW_HEADER] = "claude-desktop" profile.update(values) + # Desktop otherwise restricts tool network access to the inference gateway. + profile.setdefault("coworkEgressAllowedHosts", ["*"]) profile["inferenceCustomHeaders"] = headers profile.pop("inferenceModels", None) return _write(library.profile_path, profile) diff --git a/src/free_claude_code/providers/openai_chat/transport.py b/src/free_claude_code/providers/openai_chat/transport.py index db7f27150d..e2154d85fb 100644 --- a/src/free_claude_code/providers/openai_chat/transport.py +++ b/src/free_claude_code/providers/openai_chat/transport.py @@ -934,6 +934,7 @@ def hold_event(event: str) -> Iterator[str]: output_reasoning = self._reasoning.output_enabled corrections = RequestCorrections("chat", self._reasoning_correction) trace_event( + lambda: {"body": provider_chat_body_snapshot(body)}, stage="provider", event="provider.request.sent", source="provider", @@ -944,7 +945,6 @@ def hold_event(event: str) -> Iterator[str]: downstream_model=body.get("model"), message_count=len(body.get("messages", [])), tool_count=len(body.get("tools", [])), - body=provider_chat_body_snapshot(body), ) while True: diff --git a/src/free_claude_code/runtime/application.py b/src/free_claude_code/runtime/application.py index d1491eaba4..7f0d8a12a7 100644 --- a/src/free_claude_code/runtime/application.py +++ b/src/free_claude_code/runtime/application.py @@ -32,6 +32,7 @@ from free_claude_code.config.settings import Settings from free_claude_code.core.async_tasks import run_sync_owned from free_claude_code.core.json_types import JsonObject +from free_claude_code.core.version import package_version from free_claude_code.messaging.voice import Transcriber from free_claude_code.providers.credential_validation import ( CredentialStatus, @@ -149,6 +150,11 @@ def __init__( transcriber_factory=transcriber_factory, ) + @property + def instance_id(self) -> str: + """Identity shared by this runtime's status responses and logs.""" + return self._instance_id + @property def settings(self) -> Settings: return self.provider_manager.current_settings() @@ -167,7 +173,11 @@ async def start(self) -> None: ) if self._started: return - logger.info("Starting Claude Code Proxy...") + logger.bind( + event="server.starting", + instance_id=self.instance_id, + fcc_version=package_version(), + ).info("Starting Claude Code Proxy...") await _await_owned_task( asyncio.create_task(self._configuration.initialize()) ) diff --git a/src/free_claude_code/runtime/asgi.py b/src/free_claude_code/runtime/asgi.py index 0b32da6a7c..69562addf4 100644 --- a/src/free_claude_code/runtime/asgi.py +++ b/src/free_claude_code/runtime/asgi.py @@ -18,10 +18,11 @@ def __getattr__(self, name: str) -> object: return getattr(self.app, name) async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: - if scope["type"] != "lifespan": - await self.app(scope, receive, send) - return - await self._lifespan(receive, send) + with logger.contextualize(instance_id=self.runtime.instance_id): + if scope["type"] != "lifespan": + await self.app(scope, receive, send) + return + await self._lifespan(receive, send) async def _lifespan(self, receive: Receive, send: Send) -> None: started = False diff --git a/tests/api/test_app_lifespan_and_errors.py b/tests/api/test_app_lifespan_and_errors.py index aa70da20c5..2604732f8c 100644 --- a/tests/api/test_app_lifespan_and_errors.py +++ b/tests/api/test_app_lifespan_and_errors.py @@ -1,3 +1,4 @@ +import asyncio import logging from pathlib import Path from typing import cast @@ -6,13 +7,17 @@ import pytest from fastapi import FastAPI from fastapi.testclient import TestClient +from loguru import logger +from free_claude_code.api.request_ids import RequestCorrelationMiddleware from free_claude_code.application.errors import ( ApplicationUnavailableError, InvalidRequestError, ) from free_claude_code.application.model_metadata import ProviderModelInfo from free_claude_code.config.settings import Settings +from free_claude_code.core.async_tasks import run_sync_owned +from free_claude_code.core.version import package_version from free_claude_code.messaging.transcription import TranscriptionService from free_claude_code.providers.nvidia_nim.client import NvidiaNimProvider from free_claude_code.providers.nvidia_nim.voice import NvidiaNimTranscriber @@ -39,7 +44,7 @@ def _redirect_fcc_home(monkeypatch, tmp_path): @pytest.mark.asyncio -async def test_runtime_startup_logs_admin_url_without_printed_server_banner(): +async def test_runtime_startup_logs_admin_url_without_printed_server_banner(caplog): settings = _settings( messaging_platform="none", host="127.0.0.1", @@ -61,6 +66,7 @@ async def test_runtime_startup_logs_admin_url_without_printed_server_banner(): ), patch.object(uvicorn_logger, "info") as log_info, ): + await runtime.start() await runtime.start() log_info.assert_not_called() runtime.http_started() @@ -72,6 +78,14 @@ async def test_runtime_startup_logs_admin_url_without_printed_server_banner(): "Admin UI: %s (local-only)", "http://127.0.0.1:9099/admin", ) + startup = [ + record + for record in caplog.records + if record.extra.get("event") == "server.starting" + ] + assert len(startup) == 1 + assert startup[0].extra["instance_id"] == runtime.instance_id + assert startup[0].extra["fcc_version"] == package_version() def test_create_app_application_error_handler_returns_anthropic_format(): @@ -196,11 +210,16 @@ def test_startup_failure_message_preserves_existing_concise_contract(): @pytest.mark.asyncio -async def test_runtime_asgi_app_starts_and_closes_owner_once(): - runtime = MagicMock(spec=ApplicationRuntime) +async def test_runtime_asgi_app_starts_and_closes_owner_once(caplog): + runtime = MagicMock(spec=ApplicationRuntime, instance_id="lifespan-instance") runtime.settings = _settings() - runtime.start = AsyncMock() - runtime.close = AsyncMock(return_value=True) + runtime.start = AsyncMock(side_effect=lambda: logger.info("lifespan starting")) + + async def close(): + logger.info("lifespan closing") + return True + + runtime.close = AsyncMock(side_effect=close) app = RuntimeASGIApp(AsyncMock(), runtime) received = iter( [ @@ -224,6 +243,66 @@ async def send(message): {"type": "lifespan.startup.complete"}, {"type": "lifespan.shutdown.complete"}, ] + assert len(caplog.records) == 2 + assert all( + record.extra["instance_id"] == "lifespan-instance" for record in caplog.records + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("failure", [None, RuntimeError, asyncio.CancelledError]) +async def test_asgi_log_context_isolated_across_runtimes_tasks_and_workers( + caplog, failure +): + release = asyncio.Event() + jobs = [] + + async def background(path): + await release.wait() + logger.info("background {}", path) + + async def app(scope, receive, send): + path = scope["path"] + logger.info("request {}", path) + jobs.append(asyncio.create_task(background(path))) + await run_sync_owned(lambda: logger.info("worker {}", path)) + if failure is not None: + raise failure() + + async def request(name): + runtime = MagicMock(spec=ApplicationRuntime, instance_id=name) + wrapped = RuntimeASGIApp(RequestCorrelationMiddleware(app), runtime) + try: + await wrapped( + {"type": "http", "method": "GET", "path": "/" + name, "headers": []}, + AsyncMock(), + AsyncMock(), + ) + finally: + logger.info("after {}", name) + + try: + results = await asyncio.gather( + request("first"), request("second"), return_exceptions=True + ) + finally: + release.set() + await asyncio.gather(*jobs) + for name in ("first", "second"): + records = [ + record for record in caplog.records if record.message.endswith("/" + name) + ] + assert len(records) == 3 + assert all(record.extra["instance_id"] == name for record in records) + assert len({record.extra["request_id"] for record in records}) == 1 + after = next( + record for record in caplog.records if record.message == "after " + name + ) + assert "instance_id" not in after.extra + assert "request_id" not in after.extra + assert all( + isinstance(result, failure) if failure else result is None for result in results + ) @pytest.mark.asyncio diff --git a/tests/api/test_response_streams.py b/tests/api/test_response_streams.py index ce71e688aa..a53f93af18 100644 --- a/tests/api/test_response_streams.py +++ b/tests/api/test_response_streams.py @@ -4,10 +4,11 @@ import json from collections.abc import AsyncGenerator from typing import Any, cast -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock import pytest from fastapi.responses import JSONResponse, StreamingResponse +from loguru import logger from starlette.types import Message, Scope from free_claude_code.api.request_ids import RequestCorrelationMiddleware @@ -573,7 +574,10 @@ async def aclose(self) -> None: @pytest.mark.asyncio -async def test_cleanup_failures_are_trace_only_and_do_not_replace_success() -> None: +async def test_cleanup_failures_warn_once_and_do_not_replace_success(caplog) -> None: + close_error = RuntimeError("close detail") + release_error = RuntimeError("release detail") + class CloseFails: def __init__(self) -> None: self._yielded = False @@ -588,33 +592,31 @@ async def __anext__(self) -> str: return "ok" async def aclose(self) -> None: - raise RuntimeError("secret close detail") + raise close_error - release = AsyncMock(side_effect=RuntimeError("secret release detail")) + release = AsyncMock(side_effect=release_error) response = ManagedStreamingResponse(CloseFails()) await bind_response_lifetime(response, release) - with ( - patch("free_claude_code.core.trace.trace_event") as close_trace, - patch("free_claude_code.api.response_streams.trace_event") as release_trace, - ): + with logger.contextualize(request_id="req_cleanup"): messages = await _serve(response) + await response.aclose() assert b"".join(message.get("body", b"") for message in messages) == b"ok" - close_trace.assert_called_once() - assert close_trace.call_args.kwargs["owner"] == "ManagedStreamingResponse" - assert close_trace.call_args.kwargs["close_exc_type"] == "RuntimeError" - assert release_trace.call_args.kwargs["operation"] == "release_resource" - trace_blob = " ".join( - str(call) - for call in [*close_trace.call_args_list, *release_trace.call_args_list] - ) - assert "secret close detail" not in trace_blob - assert "secret release detail" not in trace_blob + warnings = [record for record in caplog.records if record.levelname == "WARNING"] + assert len(warnings) == 2 + assert warnings[0].extra["event"] == "stream.input.close_failed" + assert warnings[0].extra["owner"] == "ManagedStreamingResponse" + assert warnings[0].exc_info[1] is close_error + assert warnings[1].extra["operation"] == "release_resource" + assert warnings[1].exc_info[1] is release_error + assert all(record.extra["request_id"] == "req_cleanup" for record in warnings) + assert "aclose" in caplog.text + release.assert_awaited_once() @pytest.mark.asyncio -async def test_body_close_cancellation_propagates_without_releasing() -> None: +async def test_body_close_cancellation_propagates_without_releasing(caplog) -> None: class CloseIsCancelled: def __aiter__(self): return self @@ -633,6 +635,7 @@ async def aclose(self) -> None: await response.aclose() release.assert_not_awaited() + assert not [record for record in caplog.records if record.levelname == "WARNING"] @pytest.mark.asyncio diff --git a/tests/api/test_safe_logging.py b/tests/api/test_safe_logging.py index 73eb6cfcf8..8b70f2ea84 100644 --- a/tests/api/test_safe_logging.py +++ b/tests/api/test_safe_logging.py @@ -1,5 +1,6 @@ """Tests that API and SSE logging avoid raw sensitive payloads by default.""" +import logging from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -8,7 +9,6 @@ from free_claude_code.api import request_errors from free_claude_code.api.handlers import MessagesHandler, TokenCountHandler -from free_claude_code.application import execution from free_claude_code.config.settings import Settings from free_claude_code.core.anthropic import AnthropicStreamLedger from free_claude_code.core.anthropic.models import Message, MessagesRequest @@ -16,7 +16,7 @@ @pytest.mark.asyncio -async def test_create_message_skips_full_payload_debug_log_by_default(): +async def test_create_message_skips_full_payload_debug_log_by_default(caplog): settings = Settings() assert settings.log_raw_api_payloads is False mock_provider = MagicMock() @@ -37,19 +37,16 @@ async def fake_stream(*_a, **_kw): messages=[Message(role="user", content="secret-user-text")], ) - with patch.object(execution.logger, "debug") as mock_debug: + with caplog.at_level(logging.DEBUG): await service.create(request) - full_payload_calls = [ - c - for c in mock_debug.call_args_list - if c.args and str(c.args[0]) == "FULL_PAYLOAD [{}]: {}" - ] - assert not full_payload_calls + assert not any( + record.message.startswith("FULL_PAYLOAD [") for record in caplog.records + ) @pytest.mark.asyncio -async def test_create_message_logs_full_payload_when_opt_in(): +async def test_create_message_logs_full_payload_when_opt_in(caplog): settings = Settings() settings.log_raw_api_payloads = True mock_provider = MagicMock() @@ -69,11 +66,16 @@ async def fake_stream(*_a, **_kw): messages=[Message(role="user", content="visible")], ) - with patch.object(execution.logger, "debug") as mock_debug: + with caplog.at_level(logging.DEBUG): await service.create(request) - keys = [c.args[0] for c in mock_debug.call_args_list if c.args] - assert any(k == "FULL_PAYLOAD [{}]: {}" for k in keys) + payloads = [ + record.message + for record in caplog.records + if record.message.startswith("FULL_PAYLOAD [") + ] + assert len(payloads) == 1 + assert "visible" in payloads[0] def test_stream_ledger_default_debug_has_no_serialized_json_content(): diff --git a/tests/application/test_code_sessions.py b/tests/application/test_code_sessions.py index caae66aa11..62c037e32b 100644 --- a/tests/application/test_code_sessions.py +++ b/tests/application/test_code_sessions.py @@ -1286,19 +1286,21 @@ async def test_shutdown_settles_run_when_native_interrupt_never_acknowledges(cod @pytest.mark.asyncio async def test_storage_failure_retires_native_work_even_with_pending_prompt( - code, monkeypatch + code, monkeypatch, caplog ): service, harness, _ = code session = await session_for(code) - await service.send( + run = await service.send( session.id, new_id(), session.revision, "prompt", expected_epoch=service.epoch ) await harness.started.wait() connection = harness.connections[0] await connection.prompt(1) + error = CodeUnavailableError("Code disk unavailable") + async def fail_save(*args, **kwargs): - raise CodeUnavailableError("Code disk unavailable") + raise error monkeypatch.setattr(service._store, "save_progress", fail_save) await connection.text("turn-1", "text", "unsaved", complete=True) @@ -1306,6 +1308,74 @@ async def fail_save(*args, **kwargs): assert connection.closed with pytest.raises(CodeUnavailableError): await service.get_detail(session.id) + await service.close() + failures = [record for record in caplog.records if record.levelname == "ERROR"] + assert len(failures) == 1 + assert failures[0].extra == { + "event": "code.storage_failed", + "operation": "save_progress", + "session_id": session.id, + "run_id": run.id, + } + assert failures[0].exc_info[1] is error + assert "fail_save" in caplog.text + + +@pytest.mark.asyncio +async def test_shutdown_storage_read_failure_logs_original_cause( + code, monkeypatch, caplog +): + service, harness, _ = code + session = await session_for(code) + run = await service.send( + session.id, new_id(), session.revision, "work", expected_epoch=service.epoch + ) + await harness.started.wait() + error = CodeUnavailableError("Cannot read run during shutdown") + monkeypatch.setattr(service._store, "get_run", AsyncMock(side_effect=error)) + + await service.close() + + assert harness.connections[0].closed + failures = [record for record in caplog.records if record.levelname == "ERROR"] + assert len(failures) == 1 + assert failures[0].extra["operation"] == "stop_run" + assert failures[0].extra["session_id"] == session.id + assert failures[0].extra["run_id"] == run.id + assert failures[0].exc_info[1] is error + + +@pytest.mark.asyncio +@pytest.mark.parametrize("conflict", [False, True]) +async def test_delete_before_first_run_logs_only_terminal_storage_failure( + code, monkeypatch, caplog, conflict +): + service, _, _ = code + session = await session_for(code) + error = ( + CodeConflictError("Revision changed") + if conflict + else CodeUnavailableError("Cannot save deletion") + ) + save = service._store.save_progress + monkeypatch.setattr(service._store, "save_progress", AsyncMock(side_effect=error)) + + with pytest.raises(type(error)): + await service.delete_session(session.id, session.revision) + if conflict: + monkeypatch.setattr(service._store, "save_progress", save) + assert (await service.get_detail(session.id)).session == session + else: + await asyncio.wait_for(service.wait_idle(session.id), 3) + await service.close() + + failures = [record for record in caplog.records if record.levelname == "ERROR"] + assert len(failures) == (0 if conflict else 1) + if not conflict: + assert failures[0].extra["session_id"] == session.id + assert failures[0].extra["run_id"] is None + assert failures[0].extra["operation"] == "save_progress" + assert failures[0].exc_info[1] is error @pytest.mark.asyncio diff --git a/tests/application/test_execution.py b/tests/application/test_execution.py index 8426b33132..bfadbac205 100644 --- a/tests/application/test_execution.py +++ b/tests/application/test_execution.py @@ -389,7 +389,7 @@ def lookup(provider_id, model_id): event async for event in getattr(executor, f"stream_{ingress}")( route(_target("fallback", "fallback-model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="capability-fallback", ) ] @@ -433,7 +433,7 @@ def _executor_stream( ) return executor.stream_messages( _routed_request(), - raw_log_payload={}, + raw_log_payload=dict, request_id=request_id, ) @@ -451,7 +451,7 @@ async def test_executor_routes_native_responses_without_messages_conversion() -> stream = executor.stream_responses( routed, - raw_log_payload=request.model_dump(mode="json"), + raw_log_payload=lambda: request.model_dump(mode="json"), request_id="req_responses_application", ) @@ -486,7 +486,7 @@ async def test_executor_uses_structural_provider_port_and_defers_stream_startup( stream = executor.stream_messages( routed, - raw_log_payload=request.model_dump(), + raw_log_payload=request.model_dump, request_id="req_application", ) @@ -530,7 +530,7 @@ async def resolve(provider_id: str) -> FakeProvider: executor = ProviderExecutor(resolve, progress_timeout_seconds=60.0) stream = executor.stream_messages( _routed_request(_target("fallback", "fallback-model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_primary_success", ) @@ -566,7 +566,7 @@ async def resolve(provider_id: str) -> FakeProvider: chunk async for chunk in executor.stream_messages( routed, - raw_log_payload={}, + raw_log_payload=dict, request_id="req_fallback", ) ] @@ -629,7 +629,7 @@ async def test_fallback_chain_preserves_exact_last_failure() -> None: ) stream = executor.stream_messages( _routed_request(_target("second", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_exhausted", ) @@ -659,7 +659,7 @@ async def resolve(provider_id: str) -> FakeProvider: _target("second", "second-model"), _target("third", "third-model"), ), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_ordered_fallbacks", ) @@ -681,7 +681,7 @@ async def resolve(provider_id: str) -> FakeProvider: executor = ProviderExecutor(resolve, progress_timeout_seconds=60.0) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_empty_primary", ) @@ -703,7 +703,7 @@ async def resolve(provider_id: str) -> FakeProvider: executor = ProviderExecutor(resolve, progress_timeout_seconds=60.0) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_unexpected_primary", ) @@ -728,7 +728,7 @@ async def resolve(provider_id: str) -> FakeProvider: with patch.object(primary, "stream_messages", return_value=None): stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_invalid_stream", ) with pytest.raises(TypeError): @@ -760,7 +760,7 @@ async def resolve(provider_id: str) -> FakeProvider: executor = ProviderExecutor(resolve, progress_timeout_seconds=60.0) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_rejected", ) @@ -796,7 +796,7 @@ async def test_primary_canonical_startup_failure_selects_fallback() -> None: stream = executor.stream_messages( _routed_request(_target("fallback", "fallback-model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_startup_fallback", ) @@ -824,7 +824,7 @@ async def test_nonretryable_stream_construction_failure_selects_fallback() -> No ) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_construction_failure", ) @@ -846,7 +846,7 @@ async def resolve(provider_id: str) -> FakeProvider: executor = ProviderExecutor(resolve, progress_timeout_seconds=60.0) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_committed", ) @@ -870,7 +870,7 @@ async def test_lazy_fallback_startup_application_error_stops_unchanged() -> None ) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_startup", ) @@ -919,7 +919,7 @@ async def stream_messages( chunk async for chunk in executor.stream_messages( routed, - raw_log_payload={}, + raw_log_payload=dict, request_id="req_isolated", ) ] @@ -940,7 +940,7 @@ async def test_closing_executor_stream_closes_provider_stream_once() -> None: ) stream = executor.stream_messages( routed, - raw_log_payload={}, + raw_log_payload=dict, request_id="req_early_close", ) @@ -969,7 +969,7 @@ async def resolve(provider_id: str) -> FakeProvider: stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_deferred_construction", ) @@ -991,7 +991,7 @@ async def test_executor_validation_is_deferred_until_iteration() -> None: stream = executor.stream_messages( _routed_request(), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_application", ) token_counter.assert_called_once() @@ -1067,7 +1067,7 @@ async def resolve(provider_id: str) -> FakeProvider: executor = ProviderExecutor(resolve, progress_timeout_seconds=0.02) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_terminal_progress_timeout", ) @@ -1098,7 +1098,7 @@ async def resolve(provider_id: str) -> FakeProvider: executor = ProviderExecutor(resolve, progress_timeout_seconds=0.02) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_cleanup_timeout", ) @@ -1120,7 +1120,7 @@ async def resolve(provider_id: str) -> FakeProvider: @pytest.mark.asyncio -async def test_cleanup_failure_trace_names_preserved_provider_failure() -> None: +async def test_cleanup_warning_names_preserved_provider_failure(caplog) -> None: failure = _execution_failure("primary overloaded") provider = CloseControlledProvider( failure, @@ -1132,20 +1132,16 @@ async def test_cleanup_failure_trace_names_preserved_provider_failure() -> None: request_id="req_close_failure_trace", ) - with ( - patch("free_claude_code.core.trace.trace_event") as trace_mock, - pytest.raises(ExecutionFailure) as exc_info, - ): + with pytest.raises(ExecutionFailure) as exc_info: await anext(stream) assert exc_info.value is failure - close_trace = next( - call.kwargs - for call in trace_mock.call_args_list - if call.kwargs.get("event") == "stream.input.close_failed" - ) - assert close_trace["close_exc_type"] == "RuntimeError" - assert close_trace["preserved_exc_type"] == "ExecutionFailure" + warnings = [record for record in caplog.records if record.levelname == "WARNING"] + assert len(warnings) == 1 + assert warnings[0].extra["event"] == "stream.input.close_failed" + assert warnings[0].extra["close_exc_type"] == "RuntimeError" + assert warnings[0].extra["preserved_exc_type"] == "ExecutionFailure" + assert str(warnings[0].exc_info[1]) == "close failed" @pytest.mark.asyncio @@ -1203,7 +1199,7 @@ async def test_fallback_transition_does_not_reset_shared_progress_deadline() -> ) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_shared_deadline", ) deadlines: list[float | None] = [] @@ -1281,7 +1277,7 @@ async def resolve(provider_id: str) -> FakeProvider: ) stream = executor.stream_messages( _routed_request(_target("fallback", "model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_cancelled_progress", ) task = asyncio.ensure_future(anext(stream)) diff --git a/tests/application/test_request_conversion.py b/tests/application/test_request_conversion.py index 49508eb478..2446d56ccf 100644 --- a/tests/application/test_request_conversion.py +++ b/tests/application/test_request_conversion.py @@ -71,7 +71,7 @@ def reply(request): provider._chat, builder_name, wraps=getattr(provider._chat, builder_name) ) as build: stream = getattr(executor, f"stream_{wire}")( - routed, raw_log_payload={}, request_id="conversion-once" + routed, raw_log_payload=dict, request_id="conversion-once" ) output = "".join([event async for event in stream]) assert "hello" in output @@ -106,7 +106,7 @@ async def test_executor_converts_selected_opencode_route_once(wire, egress, warm AsyncMock(side_effect=lambda _: provider), progress_timeout_seconds=60 ) stream = getattr(executor, f"stream_{wire}")( - routed, raw_log_payload={}, request_id="opencode-conversion-once" + routed, raw_log_payload=dict, request_id="opencode-conversion-once" ) assert f"{egress}-ok" in "".join([event async for event in stream]) assert build.call_count == 1 @@ -145,7 +145,7 @@ def reply(request): AsyncMock(side_effect=lambda _: provider), progress_timeout_seconds=60 ) stream = getattr(executor, f"stream_{wire}")( - routed, raw_log_payload={}, request_id="subscription-conversion-once" + routed, raw_log_payload=dict, request_id="subscription-conversion-once" ) assert auth.access_calls == 0 assert "hello" in "".join([event async for event in stream]) diff --git a/tests/cli/test_entrypoints.py b/tests/cli/test_entrypoints.py index 856dbe6d12..a922ff30d2 100644 --- a/tests/cli/test_entrypoints.py +++ b/tests/cli/test_entrypoints.py @@ -116,8 +116,9 @@ def test_explicit_open_admin_waits_for_owned_http_ready(): open_admin.assert_not_called() settings = _launcher_settings() supervisor._ready_settings = settings + supervisor._ready_instance_id = "ready-instance" supervisor.request_open_admin() - open_admin.assert_called_once_with(settings, 0) + open_admin.assert_called_once_with(settings, 0, "ready-instance") @pytest.mark.parametrize("open_admin_browser", (False, True)) @@ -296,6 +297,7 @@ def build_asgi_app(_settings: Settings, restart_callback: Callable[[], None]): restart_callbacks.append(restart_callback) app = SimpleNamespace( runtime=SimpleNamespace( + instance_id=f"instance-{len(apps)}", is_closed=False, begin_shutdown=lambda: None, http_started=lambda: None, @@ -340,7 +342,7 @@ def fake_config(app, **kwargs): commands.serve() assert len(servers) == 2 - open_admin.assert_called_once_with(settings, 0) + open_admin.assert_called_once_with(settings, 0, "instance-0") clear_settings_cache.assert_called_once() kill_all.assert_called_once() @@ -357,6 +359,7 @@ def build_asgi_app(_settings: Settings, restart_callback: Callable[[], None]): restart_callbacks.append(restart_callback) return SimpleNamespace( runtime=SimpleNamespace( + instance_id="incomplete-instance", is_closed=False, begin_shutdown=lambda: None, http_started=lambda: None, diff --git a/tests/cli/test_progressive_launch.py b/tests/cli/test_progressive_launch.py index 35410b65a4..08b23b60ea 100644 --- a/tests/cli/test_progressive_launch.py +++ b/tests/cli/test_progressive_launch.py @@ -113,7 +113,7 @@ def started(): @pytest.mark.parametrize("valid", [False, True]) @pytest.mark.parametrize("browser_result", [True, False, RuntimeError("opener failed")]) def test_existing_server_must_identify_itself_as_fcc( - monkeypatch, browser_workers, valid, browser_result + monkeypatch, browser_workers, valid, browser_result, caplog ): settings = Settings() payload = {"unrelated": "server"} @@ -138,6 +138,70 @@ def test_existing_server_must_identify_itself_as_fcc( monkeypatch.setattr(commands.webbrowser, "open", browser) assert commands.open_admin_when_ready(settings) is valid assert browser.call_count == int(valid) + if valid and browser_result is not True: + assert len(caplog.records) == 1 + assert caplog.records[0].extra["instance_id"] == payload["instance_id"] + + +@pytest.mark.parametrize("automatic", [False, True], ids=["tray", "automatic"]) +@pytest.mark.parametrize("raises", [False, True]) +def test_browser_warning_retains_owning_instance_after_server_exit( + monkeypatch, browser_workers, caplog, automatic, raises +): + supervisor = commands.ServerSupervisor(console_logging=False) + settings = Settings() + entered, release = threading.Event(), threading.Event() + instances = [] + + def build(*args, **kwargs): + instance_id = f"instance-{len(instances)}" + instances.append(instance_id) + return SimpleNamespace( + runtime=SimpleNamespace( + instance_id=instance_id, + is_closed=True, + http_started=lambda: None, + begin_shutdown=lambda: None, + close=AsyncMock(return_value=True), + ) + ) + + def browser(url): + entered.set() + assert release.wait(5) + if raises: + raise RuntimeError("opener failed") + return False + + class Server: + def __init__(self, config, *, on_started, **kwargs): + self.on_started = on_started + + def run(self, **kwargs): + self.on_started() + if not automatic: + tray = threading.Thread(target=supervisor.request_open_admin) + tray.start() + tray.join(2) + assert not tray.is_alive() + assert entered.wait(2) + + monkeypatch.setattr("free_claude_code.runtime.bootstrap.build_asgi_app", build) + monkeypatch.setattr(uvicorn, "Config", lambda *args, **kwargs: None) + monkeypatch.setattr("free_claude_code.cli.uvicorn_server.RuntimeServer", Server) + monkeypatch.setattr(commands.webbrowser, "open", browser) + try: + for _ in range(2): + entered.clear() + supervisor._run_bound( + settings, [], open_admin_browser=automatic, restart_generation=0 + ) + finally: + release.set() + for worker in browser_workers: + worker.join(2) + assert len(caplog.records) == 2 + assert {record.extra["instance_id"] for record in caplog.records} == set(instances) @pytest.mark.parametrize("change", ["none", "stop", "restart", "settings"]) @@ -146,6 +210,7 @@ def test_queued_browser_rechecks_owner_before_handoff(monkeypatch, change): settings = Settings() supervisor.schedule_run() supervisor._ready_settings = settings + supervisor._ready_instance_id = "queued-instance" queued = [] monkeypatch.setattr(threading.Thread, "start", lambda thread: queued.append(thread)) browser = MagicMock(return_value=True) @@ -163,7 +228,7 @@ def test_queued_browser_rechecks_owner_before_handoff(monkeypatch, change): @pytest.mark.parametrize("reuse", [False, True]) -def test_browser_thread_start_failure_does_not_fail_fcc(monkeypatch, reuse): +def test_browser_thread_start_failure_does_not_fail_fcc(monkeypatch, reuse, caplog): settings = Settings() browser = MagicMock() monkeypatch.setattr(commands.webbrowser, "open", browser) @@ -187,9 +252,11 @@ def test_browser_thread_start_failure_does_not_fail_fcc(monkeypatch, reuse): else: supervisor = commands.ServerSupervisor() supervisor._ready_settings = settings + supervisor._ready_instance_id = "a" * 32 supervisor.request_open_admin() supervisor.request_stop() browser.assert_not_called() + assert caplog.records[-1].extra["instance_id"] == "a" * 32 def _run_browser_shutdown_probe(mode, outcome, directory, setup_delay="0"): @@ -224,6 +291,7 @@ def browser(url): if mode in {"server", "desktop"}: supervisor = commands.ServerSupervisor(console_logging=False) runtime = SimpleNamespace( + instance_id="browser-probe-instance", is_closed=True, begin_shutdown=lambda: None, http_started=lambda: None, diff --git a/tests/cli/test_server_logging.py b/tests/cli/test_server_logging.py index 91295f311f..da720ac9bf 100644 --- a/tests/cli/test_server_logging.py +++ b/tests/cli/test_server_logging.py @@ -30,11 +30,14 @@ def _exercise_server_logging( # An earlier server may already have installed Uvicorn's console handlers. logging.config.dictConfig(baseline) settings = Settings(host="127.0.0.1", port=8082, log_level=level) + instances = [] def build_app(settings, restart_callback): + instances.append(f"instance-{len(instances) + 1}") configure_logging(path, level=settings.log_level) runtime = MagicMock( spec=ApplicationRuntime, + instance_id=instances[-1], settings=settings, _draining=False, _http_ready=asyncio.Event(), @@ -75,6 +78,7 @@ def run(server, sockets): supervisor._run_bound( settings, [], open_admin_browser=False, restart_generation=0 ) + logger.warning("outside server run") logger.complete() assert baseline == uvicorn.config.LOGGING_CONFIG @@ -122,6 +126,17 @@ def test_server_logs_reach_file_once_per_start( assert row["record"]["exception"]["type"] == "ValueError" assert "controlled server failure" in row["text"] assert "Traceback" in row["text"] + server_records = [ + row["record"] + for row in rows + if row["record"]["message"] != "outside server run" + ] + assert {record["extra"]["instance_id"] for record in server_records} == { + "instance-1", + "instance-2", + } + assert len({record["process"]["id"] for record in server_records}) == 1 + assert "instance_id" not in rows[-1]["record"]["extra"] assert result.stdout.count("/test-request") == (2 if console else 0) assert result.stderr.count("Admin UI:") == (2 if console else 0) assert result.stderr.count("ASGI request failed") == (2 if console else 0) diff --git a/tests/cli/test_server_shutdown.py b/tests/cli/test_server_shutdown.py index 7f3fe4c8ec..3dac664d58 100644 --- a/tests/cli/test_server_shutdown.py +++ b/tests/cli/test_server_shutdown.py @@ -108,3 +108,14 @@ def create_app(settings: Settings, *, restart_callback: RestartCallback): assert "timeout graceful shutdown exceeded" not in caplog.text assert all(app.runtime.is_closed for app in apps) assert len(apps) == (2 if restart else 1) + starts = [ + extra + for record in caplog.records + if (extra := getattr(record, "extra", {})).get("event") == "server.starting" + ] + assert [extra["instance_id"] for extra in starts] == [ + app.runtime.instance_id for app in apps + ] + assert starts[0]["instance_id"] == old_instance + if restart: + assert starts[1]["instance_id"] == new_status["instance_id"] diff --git a/tests/cli/test_update_wrapper.py b/tests/cli/test_update_wrapper.py index dcb7ac7a3d..43d41e6e2c 100644 --- a/tests/cli/test_update_wrapper.py +++ b/tests/cli/test_update_wrapper.py @@ -45,7 +45,9 @@ def log_message(self, format, *args): pass server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) - thread = threading.Thread(target=server.serve_forever) + thread = threading.Thread( + target=server.serve_forever, kwargs={"poll_interval": 0.01} + ) thread.start() try: yield f"http://127.0.0.1:{server.server_port}/install", requests @@ -116,7 +118,6 @@ def test_installed_update_delegates_and_survives_replacement(tmp_path, outcome): env.update( UV_TOOL_DIR=str(area / "tools"), UV_TOOL_BIN_DIR=str(area / "bin"), - UV_CACHE_DIR=str(area / "cache"), UV_NO_CONFIG="1", UV_PYTHON_DOWNLOADS="never", FCC_TEST_LAUNCHER=str( diff --git a/tests/core/test_lazy_debug_logging.py b/tests/core/test_lazy_debug_logging.py new file mode 100644 index 0000000000..e4bf6e1ead --- /dev/null +++ b/tests/core/test_lazy_debug_logging.py @@ -0,0 +1,177 @@ +"""Check logging levels in a subprocess without changing pytest's global sinks.""" + +import json +import subprocess +import sys +from unittest.mock import AsyncMock, Mock, patch + +import httpx2 +import pytest +from loguru import logger +from openai import AsyncOpenAI + +from free_claude_code.api.handlers import token_count +from free_claude_code.application import execution +from free_claude_code.config.settings import Settings +from free_claude_code.core import trace +from free_claude_code.core.anthropic import TokenCountRequest +from free_claude_code.providers.openai_chat import transport as chat_transport +from tests.application.test_execution import ( + FakeProvider, + ResponsesFakeProvider, + _routed_request, + _routed_responses_request, +) +from tests.providers.test_openai_chat_transport import _consume, _success, _transport + + +async def _probe(level: str, raw: bool) -> None: + logger.remove() + rows = [] + if level != "NONE": + logger.add( + lambda message: rows.append(json.loads(message)), + level=level, + serialize=True, + ) + expected = int(level == "DEBUG") + fields = Mock(return_value={"snapshot": {"text": "visible", "api_key": "hidden"}}) + raw_payload = Mock(return_value={"text": "raw conversation"}) + with ( + patch.object( + trace, "sanitize_trace_value", wraps=trace.sanitize_trace_value + ) as sanitize, + patch.object( + execution, + "anthropic_request_snapshot", + wraps=execution.anthropic_request_snapshot, + ) as ingress, + patch.object( + token_count, + "anthropic_request_snapshot", + wraps=token_count.anthropic_request_snapshot, + ) as counted, + patch.object( + chat_transport, + "provider_chat_body_snapshot", + wraps=chat_transport.provider_chat_body_snapshot, + ) as provider_snapshot, + logger.contextualize(request_id="req_lazy", instance_id="instance_lazy"), + ): + trace.trace_event(fields, stage="unit", event="lazy.probe", source="test") + for wire, routed, provider in ( + ("messages", _routed_request(), FakeProvider()), + ("responses", _routed_responses_request(), ResponsesFakeProvider()), + ): + executor = execution.ProviderExecutor( + AsyncMock(return_value=provider), + progress_timeout_seconds=60, + token_counter=lambda *args: 1, + responses_token_counter=lambda request: 1, + log_raw_payloads=raw, + ) + output = [ + chunk + async for chunk in getattr(executor, f"stream_{wire}")( + routed, raw_log_payload=raw_payload, request_id="req_lazy" + ) + ] + assert output + routed = _routed_request() + handler = token_count.TokenCountHandler( + Settings(), + model_router=Mock(resolve_token_count_request=Mock(return_value=routed)), + token_counter=lambda *args: 7, + ) + assert ( + handler.count( + TokenCountRequest( + model=routed.request.model, messages=routed.request.messages + ) + ).input_tokens + == 7 + ) + async with AsyncOpenAI( + api_key="test", + base_url="https://provider.invalid/v1", + http_client=httpx2.AsyncClient( + transport=httpx2.MockTransport(lambda request: _success()) + ), + ) as client: + assert "hello" in await _consume(_transport(client), "messages") + assert fields.call_count == expected + assert ( + ingress.call_count + == counted.call_count + == provider_snapshot.call_count + == expected + ) + assert raw_payload.call_count == 2 * expected * raw + assert bool(sanitize.call_count) == bool(expected) + if expected: + payloads = [ + row["record"]["extra"]["trace_payload"] + for row in rows + if "trace_payload" in row["record"]["extra"] + ] + probe = next( + payload for payload in payloads if payload["event"] == "lazy.probe" + ) + assert probe == { + "stage": "unit", + "event": "lazy.probe", + "source": "test", + "snapshot": {"text": "visible", "api_key": ""}, + } + ingress_payload = next( + payload + for payload in payloads + if payload["event"] == "free_claude_code.api.request.received" + ) + assert ingress_payload["snapshot"]["model"] == "gateway-model" + assert ( + ingress_payload["snapshot"]["messages"] + == routed.request.model_dump()["messages"] + ) + assert rows[0]["record"]["extra"]["request_id"] == "req_lazy" + assert rows[0]["record"]["extra"]["instance_id"] == "instance_lazy" + else: + assert all(row["record"]["level"]["no"] >= 20 for row in rows) + assert all("trace_payload" not in row["record"]["extra"] for row in rows) + assert ( + sum(row["record"]["message"].startswith("FULL_") for row in rows) + == 2 * expected * raw + ) + + # Adding/removing DEBUG sinks must take effect immediately, with one build per event. + logger.remove() + logger.add(lambda message: None, level="INFO") + fields.reset_mock() + trace.trace_event(fields, stage="unit", event="levels", source="test") + assert fields.call_count == 0 + sinks = [logger.add(lambda message: None, level="DEBUG") for _ in range(2)] + trace.trace_event(fields, stage="unit", event="levels", source="test") + assert fields.call_count == 1 + for sink in sinks: + logger.remove(sink) + trace.trace_event(fields, stage="unit", event="levels", source="test") + assert fields.call_count == 1 + + +@pytest.mark.parametrize("level", ["INFO", "DEBUG", "NONE"]) +@pytest.mark.parametrize("raw", [False, True]) +def test_debug_payloads_are_built_only_when_enabled(level, raw): + result = subprocess.run( + [ + sys.executable, + "-c", + "import asyncio, sys; from tests.core.test_lazy_debug_logging import _probe; asyncio.run(_probe(sys.argv[1], sys.argv[2] == 'True'))", + level, + str(raw), + ], + capture_output=True, + text=True, + encoding="utf-8", + timeout=30, + ) + assert result.returncode == 0, result.stdout + result.stderr diff --git a/tests/core/test_trace.py b/tests/core/test_trace.py index deccfb807b..47bbd7f87b 100644 --- a/tests/core/test_trace.py +++ b/tests/core/test_trace.py @@ -160,12 +160,16 @@ async def test_traced_async_stream_logs_real_exception(tmp_path) -> None: assert interrupted[0]["outcome"] == "error" assert interrupted[0]["exc_type"] == "RuntimeError" close_failed = [ - row for row in rows if row.get("event") == "stream.input.close_failed" + row["record"] + for row in _json_log_rows(log_file) + if row["record"]["extra"].get("event") == "stream.input.close_failed" ] assert len(close_failed) == 1 - assert close_failed[0]["owner"] == "traced_async_stream" - assert close_failed[0]["close_exc_type"] == "RuntimeError" - assert close_failed[0]["preserved_exc_type"] == "RuntimeError" + assert close_failed[0]["level"]["name"] == "WARNING" + assert close_failed[0]["exception"]["value"] == "close boom" + assert close_failed[0]["extra"]["owner"] == "traced_async_stream" + assert close_failed[0]["extra"]["close_exc_type"] == "RuntimeError" + assert close_failed[0]["extra"]["preserved_exc_type"] == "RuntimeError" @pytest.mark.asyncio diff --git a/tests/harnesses/test_claude_desktop_integration.py b/tests/harnesses/test_claude_desktop_integration.py index ad508a9b04..5f58a3939f 100644 --- a/tests/harnesses/test_claude_desktop_integration.py +++ b/tests/harnesses/test_claude_desktop_integration.py @@ -170,6 +170,7 @@ def test_fresh_connection_and_disconnect_preserve_native_library(tmp_path): assert read(meta(root))["appliedId"] == desktop.FCC_ID assert read(mode(root))["deploymentMode"] == "3p" assert read(profile(root))["inferenceGatewayApiKey"] == TOKEN + assert read(profile(root))["coworkEgressAllowedHosts"] == ["*"] before = {p: p.read_bytes() for p in root.rglob("*.json")} assert ( desktop.refresh_connected( @@ -200,6 +201,42 @@ def test_fresh_connection_and_disconnect_preserve_native_library(tmp_path): assert {p: p.read_bytes() for p in root.rglob("*.json")} == after +@pytest.mark.parametrize("reconnect", [False, True]) +@pytest.mark.parametrize( + "egress", + [ + {}, + {"coworkEgressAllowedHosts": []}, + {"coworkEgressAllowedHosts": ["github.com"]}, + ], +) +def test_connection_defaults_missing_egress_and_preserves_explicit_policy( + tmp_path, reconnect, egress +): + record = claude_desktop_disconnect_path() + desktop.configure(tmp_path, URL, TOKEN, True, disconnect_path=record) + config = read(profile(tmp_path)) + config.pop("coworkEgressAllowedHosts", None) + config.update(egress) + write(profile(tmp_path), config) + before = profile(tmp_path).read_bytes() + assert desktop.configure(tmp_path, URL, TOKEN, disconnect_path=record)["connected"] + assert profile(tmp_path).read_bytes() == before + + if reconnect: + desktop.configure(tmp_path, URL, TOKEN, True, disconnect_path=record) + else: + assert desktop.refresh_connected( + tmp_path, URL, TOKEN, disconnect_path=record + ) is (not egress) + + assert read(profile(tmp_path))["coworkEgressAllowedHosts"] == egress.get( + "coworkEgressAllowedHosts", ["*"] + ) + assert desktop.configure(tmp_path, URL, TOKEN, disconnect_path=record)["connected"] + assert not desktop.refresh_connected(tmp_path, URL, TOKEN, disconnect_path=record) + + def test_refresh_rotates_credentials_without_reactivating(tmp_path): desktop.configure( tmp_path, URL, TOKEN, True, disconnect_path=claude_desktop_disconnect_path() diff --git a/tests/providers/test_history_transports.py b/tests/providers/test_history_transports.py index cb6997d0ce..10e32b31f3 100644 --- a/tests/providers/test_history_transports.py +++ b/tests/providers/test_history_transports.py @@ -462,7 +462,7 @@ def reply(request): try: saved = await _saved_reply( executor.stream_messages( - routed, raw_log_payload={}, request_id="actual-fallback" + routed, raw_log_payload=dict, request_id="actual-fallback" ), "messages", ) diff --git a/tests/providers/test_lmstudio.py b/tests/providers/test_lmstudio.py index 6d9de49f1b..cbdf16ce3c 100644 --- a/tests/providers/test_lmstudio.py +++ b/tests/providers/test_lmstudio.py @@ -403,7 +403,7 @@ def handler(request): route = _routed_request if wire == "messages" else _routed_responses_request routed = route(_target("fallback", "fallback-model")) stream = getattr(executor, f"stream_{wire}")( - routed, raw_log_payload={}, request_id="context-timeout" + routed, raw_log_payload=dict, request_id="context-timeout" ) async with asyncio.timeout(2): with pytest.raises(ExecutionFailure) as error: @@ -805,7 +805,7 @@ def handler(request): _routed_request() if wire == "messages" else _routed_responses_request() ) stream = getattr(executor, f"stream_{wire}")( - routed, raw_log_payload={}, request_id="cache-wait-timeout" + routed, raw_log_payload=dict, request_id="cache-wait-timeout" ) async with asyncio.timeout(2): with pytest.raises(ExecutionFailure) as error: @@ -878,7 +878,7 @@ def handler(request): route = _routed_request if wire == "messages" else _routed_responses_request stream = getattr(executor, f"stream_{wire}")( route(_target("fallback", "fallback-model")), - raw_log_payload={}, + raw_log_payload=dict, request_id="context-fallback", ) output = [event async for event in stream] diff --git a/tests/providers/test_opencode.py b/tests/providers/test_opencode.py index aea6338c9e..a6dae6707d 100644 --- a/tests/providers/test_opencode.py +++ b/tests/providers/test_opencode.py @@ -1107,7 +1107,7 @@ def upstream_id(model: str) -> str: resolved=resolved, reasoning=DEFAULT_REASONING_POLICY, ), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_opencode_cross_transport_responses", ) else: @@ -1117,7 +1117,7 @@ def upstream_id(model: str) -> str: resolved=resolved, reasoning=DEFAULT_REASONING_POLICY, ), - raw_log_payload={}, + raw_log_payload=dict, request_id="req_opencode_cross_transport_messages", ) body = "".join([chunk async for chunk in stream]) diff --git a/tests/runtime/test_sqlite_database.py b/tests/runtime/test_sqlite_database.py index a31aa4279d..12577fbf70 100644 --- a/tests/runtime/test_sqlite_database.py +++ b/tests/runtime/test_sqlite_database.py @@ -1,5 +1,6 @@ import asyncio import os +import shutil import sqlite3 import subprocess import sys @@ -252,8 +253,9 @@ def timeout(): @pytest.mark.asyncio @pytest.mark.parametrize("cancel", [False, True]) async def test_startup_failure_disposes_connections_before_releasing_owner( - tmp_path, cancel + tmp_path, cancel, initialized_database ): + shutil.copyfile(initialized_database, tmp_path / "fcc.db") entered, release = threading.Event(), threading.Event() observed = [] @@ -294,6 +296,14 @@ def _initialize(self): await database.close() +@pytest.fixture(scope="module") +def initialized_database(tmp_path_factory): + """Build the schema once, outside the ownership test's synchronization window.""" + path = tmp_path_factory.mktemp("initialized-database") / "fcc.db" + initialize_database(path) + return path + + def test_connection_setup_failure_closes_raw_connection(tmp_path, monkeypatch): connect = sqlite3.connect observed = []