This commit is contained in:
GH Action - Upstream Sync committed 2026-07-22 02:43:19 +00:00
commit 2c647d6588
9 files changed
+330 -10

No files matched your search

+26 -3
View File
@@ -246,6 +246,16 @@ 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 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
# attribution), the SSE event, and resumed continuations.
@@ -543,9 +553,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 +607,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 +803,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
@@ -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)
+12 -3
View File
@@ -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,
+6
View File
@@ -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":
+7 -2
View File
@@ -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/"))
+31 -1
View File
@@ -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.
@@ -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
# ---------------------------------------------------------------------------
+44
View File
@@ -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"
)
+69
View File
@@ -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