From fd110e80d408f0c3368daa0aba8a69b9d3c64522 Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 21 Jul 2026 09:43:22 +0100 Subject: [PATCH 1/4] fix: handle invalidated sandboxes in daytona gracefully --- application/sandbox/artifacts_capture.py | 9 +++- application/sandbox/daytona.py | 32 ++++++++++- tests/sandbox/test_artifacts_capture.py | 44 +++++++++++++++ tests/sandbox/test_daytona.py | 69 ++++++++++++++++++++++++ 4 files changed, 151 insertions(+), 3 deletions(-) diff --git a/application/sandbox/artifacts_capture.py b/application/sandbox/artifacts_capture.py index e484e8d4..fd3fef7e 100644 --- a/application/sandbox/artifacts_capture.py +++ b/application/sandbox/artifacts_capture.py @@ -122,7 +122,10 @@ def snapshot_signatures(manager: Any, session_id: str) -> Dict[str, Tuple[int, O try: files = manager.list_files(session_id) except Exception: - logger.exception("artifacts_capture: pre-exec listing failed") + # Best-effort: a listing failure (sandbox auto-stopped/deleted) just means no + # pre-image, so change detection falls back to "capture everything". Swallowed + # and recoverable -> WARN, not ERROR (an ERROR here false-alarms monitoring). + logger.warning("artifacts_capture: pre-exec listing failed", exc_info=True) return signatures candidates = sorted(f for f in files if not f.startswith("inputs/") and not _is_scratch(f)) if len(candidates) > MAX_SCANNED_FILES: @@ -164,7 +167,9 @@ def capture_artifacts( try: post_files = set(manager.list_files(session_id)) except Exception: - logger.exception("artifacts_capture: post-exec listing failed") + # Best-effort: without a post-image there is nothing to capture, so return + # empty. Swallowed and recoverable -> WARN, not ERROR (see snapshot_signatures). + logger.warning("artifacts_capture: post-exec listing failed", exc_info=True) return [] produced = (f for f in post_files if not f.startswith("inputs/")) diff --git a/application/sandbox/daytona.py b/application/sandbox/daytona.py index 688c146c..3ce1605d 100644 --- a/application/sandbox/daytona.py +++ b/application/sandbox/daytona.py @@ -376,14 +376,44 @@ class DaytonaSandbox(CodeSandbox): return self._to_result(handle.sandbox.process.code_run(wrapped, timeout=wall)) except Exception: # noqa: BLE001 - second failure -> error result below pass - return ExecResult( + result = ExecResult( status="error", error_name=type(exc).__name__, error_value=str(exc) or "code_run failed", exit_code=-1, ) + # An auto-DELETED sandbox (vs a merely stopped one, handled above) can't be + # woken and would fail every retry on this cached handle. Drop the dead + # handle and flag the runtime invalid so SandboxManager cold-starts a fresh + # sandbox on the next call, rather than looping on the tombstone until the + # idle-TTL reap. Mirrors the Jupyter backend's kernel-death handling. + if self._sandbox_gone(handle): + self._forget_handle(session_id, handle) + result.runtime_invalidated = True + return result return self._to_result(response) + def _sandbox_gone(self, handle: "_Handle") -> bool: + """True when the sandbox behind ``handle`` no longer exists in the cloud. + + Separates an auto-DELETED sandbox (unrecoverable — a fresh one must be created) + from a transient transport blip on a still-live sandbox (retryable on the same + handle). A ``get`` that cannot resolve the id, or a terminal state, means gone. + """ + try: + fresh = self._client.get(handle.sandbox_id) + except Exception: # noqa: BLE001 - unresolvable id => the sandbox is gone + return True + state = getattr(fresh, "state", None) + state_value = getattr(state, "value", state) + return state_value in (None, "destroyed", "deleted", "error", "archived") + + def _forget_handle(self, session_id: str, handle: "_Handle") -> None: + """Drop the cached handle if it is still the registered one (ABA-safe).""" + with self._lock: + if self._handles.get(session_id) is handle: + self._handles.pop(session_id, None) + @staticmethod def _with_workspace_cwd(workspace: str, code: str) -> str: """Prepend a chdir into the session workspace so relative paths resolve there. diff --git a/tests/sandbox/test_artifacts_capture.py b/tests/sandbox/test_artifacts_capture.py index f519b1df..059afa4e 100644 --- a/tests/sandbox/test_artifacts_capture.py +++ b/tests/sandbox/test_artifacts_capture.py @@ -3,6 +3,7 @@ from __future__ import annotations import hashlib +import logging import pytest @@ -127,3 +128,46 @@ class TestReadSweepCap: captured = ac.capture_artifacts(mgr, "sid", pre, user_id="u") assert captured == [] # nothing changed -> nothing persisted assert mgr.reads == ac.MAX_SCANNED_FILES # but the sweep is still bounded + + +class _ListingRaisesMgr: + """A manager whose workspace listing fails (sandbox auto-stopped/deleted).""" + + def list_files(self, _sid): + # Mirrors the generic IOError the Daytona/Jupyter backends raise when the + # workspace can't be listed because the runtime is gone. + raise IOError("list_files failed: RuntimeError") + + +@pytest.mark.unit +class TestListingFailureLogLevel: + """Regression: a swallowed, recoverable workspace-listing failure must log at + WARNING, not ERROR. + + ``snapshot_signatures`` / ``capture_artifacts`` catch a failed listing, log it, + and return empty (best-effort — the exec result is still delivered). Logging it + via ``logger.exception`` stamps it ERROR, which inflated the production error + stream: on 2026-07-20 these two lines were 7 of the day's ~43 real-app ERRORs + and false-alarmed the error review (ledger #28, Bug B). A no-op capture is not + an error. + """ + + def test_pre_exec_listing_failure_logs_warning_not_error(self, caplog): + with caplog.at_level(logging.DEBUG, logger="application.sandbox.artifacts_capture"): + sigs = ac.snapshot_signatures(_ListingRaisesMgr(), "sid") + assert sigs == {} # swallowed, best-effort + recs = [r for r in caplog.records if "pre-exec listing failed" in r.getMessage()] + assert recs, "expected a pre-exec listing-failed log record" + assert all(r.levelno == logging.WARNING for r in recs), ( + "a swallowed, recoverable listing failure must not log at ERROR" + ) + + def test_post_exec_listing_failure_logs_warning_not_error(self, caplog): + with caplog.at_level(logging.DEBUG, logger="application.sandbox.artifacts_capture"): + captured = ac.capture_artifacts(_ListingRaisesMgr(), "sid", {}, user_id="u") + assert captured == [] # swallowed, best-effort + recs = [r for r in caplog.records if "post-exec listing failed" in r.getMessage()] + assert recs, "expected a post-exec listing-failed log record" + assert all(r.levelno == logging.WARNING for r in recs), ( + "a swallowed, recoverable listing failure must not log at ERROR" + ) diff --git a/tests/sandbox/test_daytona.py b/tests/sandbox/test_daytona.py index 57137ead..658607e8 100644 --- a/tests/sandbox/test_daytona.py +++ b/tests/sandbox/test_daytona.py @@ -753,3 +753,72 @@ def test_factory_forwards_max_sandboxes(fake_sdk, monkeypatch): backend = sc.SandboxCreator.create_backend("daytona") assert backend._max_sandboxes == 7 sc.SandboxCreator.reset() + + +# --- Regression: a DELETED sandbox must invalidate the runtime (Bug A) ---- +# +# Daytona auto-STOPS an idle sandbox (auto_stop_interval) and later auto-DELETES +# it (auto_delete_interval). A stop is recoverable — ``_ensure_started`` wakes it +# and retries (covered above). A delete is NOT: the cached handle points at a +# tombstone, ``code_run`` raises "container not found", and ``client.get`` can no +# longer resolve it. The Jupyter backend handles the equivalent kernel-death by +# dropping its handle and setting ``runtime_invalidated=True`` so the manager +# cold-starts on the next call; the Daytona backend must do the same. +# +# Without this the manager keeps a dead handle and EVERY subsequent call in the +# session fails until the idle-TTL reap — observed in prod as one sandbox failing +# five times across ~18 minutes (error-review ledger #28, Bug A). + + +def test_exec_invalidates_runtime_when_sandbox_deleted(sandbox): + """A code_run against an auto-DELETED sandbox drops the handle and flags the + runtime invalid so the next open cold-starts.""" + sandbox.open("conv-1") + _, created = sandbox._client.created[0] + # Deleted under the still-cached handle: exec fails AND the sandbox can no + # longer be resolved (get raises => gone, not merely stopped). + created.process.code_run.side_effect = RuntimeError("No such container: sbx-1") + sandbox._client.get = mock.Mock(side_effect=KeyError("sbx-1")) + + res = sandbox.exec("conv-1", "print('x')") + + assert res.status == "error" + assert res.runtime_invalidated is True # signals the manager to cold-start + assert "conv-1" not in sandbox._handles # tombstone dropped; next open re-creates + + +def test_exec_does_not_invalidate_live_sandbox_on_transient_error(sandbox): + """A blip on a still-live (started) sandbox is NOT an invalidation: the warm + handle stays so a later call can reuse it.""" + sandbox.open("conv-1") + _, created = sandbox._client.created[0] + created.process.code_run.side_effect = RuntimeError("transient transport blip") + + res = sandbox.exec("conv-1", "print('x')") + + assert res.status == "error" + assert res.runtime_invalidated is False + assert "conv-1" in sandbox._handles + + +def test_manager_cold_starts_after_daytona_sandbox_deleted(fake_sdk): + """End-to-end (SandboxManager + DaytonaSandbox): once the cloud sandbox is + deleted, the next run transparently creates a FRESH sandbox instead of reusing + the dead handle. This is the user-facing recovery contract Bug A breaks.""" + from application.sandbox.daytona import DaytonaSandbox + from application.sandbox.manager import SandboxManager + + backend = DaytonaSandbox(api_key="k") + mgr = SandboxManager(backend, max_ttl=600) + mgr.open("conv-1") + _, created = backend._client.created[0] + created.process.code_run.side_effect = RuntimeError("No such container: sbx-1") + backend._client.get = mock.Mock(side_effect=KeyError("sbx-1")) # deleted => gone + + res = mgr.exec("conv-1", "print('x')") + + assert res.status == "error" + assert not mgr.has_session("conv-1") # manager dropped the invalidated session + + mgr.open("conv-1") # recovery + assert backend._client.create.call_count == 2 # a NEW sandbox, not the tombstone From 196e4e86d2802ef32a8955fe89004a0161022253 Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 21 Jul 2026 12:03:44 +0100 Subject: [PATCH 2/4] fix: handle wierd empty finish content as error --- .../answer/services/conversation_service.py | 38 +++++++++++++++++++ application/llm/openai.py | 6 +++ 2 files changed, 44 insertions(+) diff --git a/application/api/answer/services/conversation_service.py b/application/api/answer/services/conversation_service.py index f116a274..97f6d54b 100644 --- a/application/api/answer/services/conversation_service.py +++ b/application/api/answer/services/conversation_service.py @@ -33,6 +33,25 @@ TERMINATED_RESPONSE_PLACEHOLDER = ( ) +def _final_finish_reason(llm: Any) -> Optional[str]: + """Best-effort finish_reason of the round that actually produced the answer. + + Read from the responding provider — the fallback instance when a fallback + served the stream, else the primary. Set by the Responses path, the + non-streaming path, and (now) the chat-completions stream. ``None`` when + unavailable (e.g. the stream was cut before any finish frame). + """ + if llm is None: + return None + fallback = getattr(llm, "_fallback_llm", None) + responding = getattr(llm, "_responding_provider", None) + if fallback is not None and responding == getattr(fallback, "provider_name", None): + fr = getattr(fallback, "_last_finish_reason", None) + if fr is not None: + return fr + return getattr(llm, "_last_finish_reason", None) + + class ConversationService: def get_conversation( self, conversation_id: str, user_id: str @@ -112,6 +131,25 @@ class ConversationService: if metadata: message_payload["metadata"] = metadata + # Classify empty answers at the point of persistence. A tool-request + # turn (finish_reason=tool_calls — the /v1 client-tool loop) and a + # genuine dead-end (finish_reason=stop with no visible content) both + # persist an empty ``response`` and are otherwise indistinguishable in + # the stored row; log the discriminators so they can be told apart. + if not (response or "").strip(): + logger.info( + "empty_answer_persisted", + extra={ + "model": model_id, + "finish_reason": _final_finish_reason(llm), + "has_tool_calls": bool(tool_calls), + "tool_call_count": len(tool_calls or []), + "thought_len": len(thought or ""), + "responding_provider": getattr(llm, "_responding_provider", None), + "conversation_id": conversation_id, + }, + ) + if conversation_id is not None and index is not None: with db_session() as conn: repo = ConversationsRepository(conn) diff --git a/application/llm/openai.py b/application/llm/openai.py index e078a839..8f62f1a5 100644 --- a/application/llm/openai.py +++ b/application/llm/openai.py @@ -536,6 +536,7 @@ class OpenAILLM(BaseLLM): request_params["stream_options"] = stream_options self._last_usage = None self._stream_reached_finish = False + self._last_finish_reason = None response = self.client.chat.completions.create(**request_params) try: @@ -564,6 +565,11 @@ class OpenAILLM(BaseLLM): # this to refuse restreaming a delivered answer when a # trailing frame fails. self._stream_reached_finish = True + # Capture the reason itself (not just the bool) so the save + # path can classify an empty answer: a ``stop`` with no + # content is a genuine dead-end, ``tool_calls`` is a tool + # request. Previously the chat path discarded this. + self._last_finish_reason = finish_reason # Yield non-content chunks only when needed for tool-call handling. if has_tool_calls or finish_reason == "tool_calls": From ad6d9a83e0d1f63cb68858a72685de07f4327d21 Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 21 Jul 2026 13:00:15 +0100 Subject: [PATCH 3/4] fix: more durable pause --- application/api/answer/routes/base.py | 26 +++++++++++++++++++++++--- 1 file changed, 23 insertions(+), 3 deletions(-) diff --git a/application/api/answer/routes/base.py b/application/api/answer/routes/base.py index c4817462..0c4d046d 100644 --- a/application/api/answer/routes/base.py +++ b/application/api/answer/routes/base.py @@ -246,6 +246,13 @@ class BaseAnswerResource: structured_chunks = [] query_metadata: Dict[str, Any] = {} paused = False + # A ``tool_calls_pending`` event is held back and only flushed after + # continuation state is committed (or the stateless finalize path is + # reached): the v1 translator turns it into ``finish_reason:"tool_calls"``, + # and a client that resumes on that signal would otherwise race + # ``save_state``, miss the pending state, and fall back to a stateless + # ``persist=False`` round that never persists the final answer. + pending_pause_event: Optional[dict] = None # One id shared across the WAL row, primary LLM (token_usage # attribution), the SSE event, and resumed continuations. @@ -543,9 +550,12 @@ class BaseAnswerResource: yield _emit({"type": "thought", "thought": line["thought"]}) elif "type" in line: if line.get("type") == "tool_calls_pending": - # Save continuation state and end the stream + # Hold the pause event; it is flushed in the ``paused`` + # block below only once continuation state is durable, + # so a fast client's resume can never arrive before + # the state it needs to claim. paused = True - yield _emit(line) + pending_pause_event = line elif line.get("type") == "error": # An event flagged ``user_facing`` already carries a curated, # actionable message (e.g. an artifact-quota notice). Passing it @@ -594,8 +604,13 @@ class BaseAnswerResource: # ``complete`` here (recording the emitted tool_calls) and end # the stream, so the reconciler never sees a non-terminal row. # The client still gets ``finish_reason:"tool_calls"`` + the - # calls from the ``tool_calls_pending`` event yielded above. + # calls from the ``tool_calls_pending`` event flushed below. if finalize_tool_pause_as_complete: + # No continuation state is written on this path; flush the + # held pause event now, in its original position ahead of + # the terminal id/end events. + if pending_pause_event is not None: + yield _emit(pending_pause_event) yield from self._finalize_stateless_tool_pause( continuation=continuation, reserved_message_id=reserved_message_id, @@ -785,6 +800,11 @@ class BaseAnswerResource: }, ) + # Continuation state (and any first-turn conversation row) is + # committed above; only now flush the held pause event so the + # client's resume request can never beat the saved state. + if pending_pause_event is not None: + yield _emit(pending_pause_event) yield _emit({"type": "id", "id": str(conversation_id)}) yield _emit({"type": "end"}) # Drain the terminal ``end`` so a reconnecting client From ce0402a00380d2e77884420d317031fbb6eeacd2 Mon Sep 17 00:00:00 2001 From: Alex Date: Tue, 21 Jul 2026 14:28:19 +0100 Subject: [PATCH 4/4] fix: better tool persistance --- application/api/answer/routes/base.py | 7 +- application/api/v1/routes.py | 15 ++- .../api/v1/test_v1_tool_pause_finalization.py | 98 ++++++++++++++++++- 3 files changed, 114 insertions(+), 6 deletions(-) diff --git a/application/api/answer/routes/base.py b/application/api/answer/routes/base.py index 0c4d046d..380b6f13 100644 --- a/application/api/answer/routes/base.py +++ b/application/api/answer/routes/base.py @@ -250,8 +250,11 @@ class BaseAnswerResource: # continuation state is committed (or the stateless finalize path is # reached): the v1 translator turns it into ``finish_reason:"tool_calls"``, # and a client that resumes on that signal would otherwise race - # ``save_state``, miss the pending state, and fall back to a stateless - # ``persist=False`` round that never persists the final answer. + # ``save_state``, miss the pending state, and fall back to the + # transcript-rebuild round. That fallback still persists the final + # answer when a conversation is mapped, but only as an appended + # empty-prompt turn — it cannot finalize the reserved WAL row, which + # would be stranded non-terminal for the reconciler. pending_pause_event: Optional[dict] = None # One id shared across the WAL row, primary LLM (token_usage diff --git a/application/api/v1/routes.py b/application/api/v1/routes.py index 620f15a1..f8a266fd 100644 --- a/application/api/v1/routes.py +++ b/application/api/v1/routes.py @@ -298,9 +298,18 @@ def chat_completions(): internal_data["tool_actions"], ) # A missing/expired durable continuation has no reserved WAL - # row to finalize. Run it statelessly without appending a blank - # sibling turn to the mapped conversation. - internal_data["persist"] = False + # row to finalize, so nothing mid-loop is persisted — but the + # FINAL turn of this round must still land in the mapped + # conversation, else a loop whose first resume produces the + # answer (one pause, then stop — the common single-search + # shape) persists only the empty tool-call row and the answer + # is lost upstream. With persist on, the finished answer is + # APPENDED as a new empty-prompt turn (there is no reserved + # row to finalize into); that's the accepted shape here. No + # worthless rows appear along the way: continuation rounds + # reserve no WAL row and the pause path never appends. Only a + # round with no mapped conversation at all stays stateless. + internal_data["persist"] = bool(internal_data.get("conversation_id")) continuation = { "messages": messages, "tools_dict": tools_dict, diff --git a/tests/api/v1/test_v1_tool_pause_finalization.py b/tests/api/v1/test_v1_tool_pause_finalization.py index bedc734b..5fb20dfe 100644 --- a/tests/api/v1/test_v1_tool_pause_finalization.py +++ b/tests/api/v1/test_v1_tool_pause_finalization.py @@ -711,9 +711,105 @@ class TestV1ContinuationRoutesStatelessly: assert continuation["reserved_message_id"] is None assert continuation["request_id"] is None assert captured.get("finalize_tool_pause_as_complete") is False - # Missing durable state runs without appending a blank sibling turn. + # A mapped conversation keeps persistence on: the final turn of the + # round must land in it (as an appended answer turn — there is no + # reserved WAL row to finalize), else the answer is silently lost. assert captured.get("conversation_id") == "conv-1" + assert captured.get("should_persist") is True + + def test_continuation_without_conversation_id_stays_stateless(self): + """No mapped conversation → the round runs with persistence off.""" + app = self._build_app() + + def _fake_translate(data, api_key): + # Translator behavior for a stateless continuation (opencode + # et al.): no conversation_id, persist defaults to False. + return { + "question": "", + "tool_actions": [{"call_id": "c1", "result": "r"}], + "persist": False, + "messages": data["messages"], + } + + fake_processor = MagicMock() + fake_processor.decoded_token = {"sub": "owner"} + fake_processor.conversation_id = None + fake_processor.agent_config = {"user_api_key": "k"} + fake_processor.agent_id = None + fake_processor.model_id = "m" + fake_processor.model_user_id = None + fake_processor.reserved_message_id = None + fake_processor.request_id = None + fake_processor.build_continuation_from_messages.return_value = ( + MagicMock(), [], {}, [{"call_id": "c1"}], [{"call_id": "c1", "result": "r"}], "" + ) + + captured: Dict[str, Any] = {} + + def _capture_complete_stream(**kw): + captured.update(kw) + return iter(['data: {"type": "end"}']) + + fake_helper = MagicMock() + fake_helper.check_usage.return_value = None + fake_helper.complete_stream.side_effect = _capture_complete_stream + fake_helper.process_response_stream.return_value = { + "error": None, + "conversation_id": None, + "answer": "done", + "sources": [], + "tool_calls": [], + "thought": "", + } + + @contextmanager + def _yield_conn(): + yield MagicMock() + + with patch( + "application.api.v1.routes._lookup_agent", + return_value={"id": "agent-1", "name": "Agent", "user_id": "owner"}, + ), patch( + "application.api.v1.routes.translate_request", + side_effect=_fake_translate, + ), patch( + "application.api.v1.routes.StreamProcessor", + return_value=fake_processor, + ), patch( + "application.api.v1.routes._V1AnswerHelper", + return_value=fake_helper, + ), patch( + "application.api.v1.routes.db_readonly", + _yield_conn, + ), patch( + "application.api.v1.routes.translate_response", + return_value={"id": "x", "choices": []}, + ): + with app.test_client() as c: + resp = c.post( + "/v1/chat/completions", + headers={"Authorization": "Bearer x"}, + json={ + "messages": [ + {"role": "user", "content": "hi"}, + { + "role": "assistant", + "tool_calls": [{"id": "c1", "function": {}}], + }, + {"role": "tool", "tool_call_id": "c1", "content": "r"}, + ], + }, + ) + + assert resp.status_code == 200 + fake_processor.build_continuation_from_messages.assert_called_once() + fake_processor.resume_from_tool_actions.assert_not_called() + # No mapped conversation: nothing to land the answer in, so the + # round stays fully stateless (no orphan empty-question rows), and + # a further pause finalizes as complete instead of writing state. + assert captured.get("conversation_id") is None assert captured.get("should_persist") is False + assert captured.get("finalize_tool_pause_as_complete") is True # ---------------------------------------------------------------------------