From c51afe87add12b49bf030190446aab61b55b3a88 Mon Sep 17 00:00:00 2001 From: Alejandro Cantu Date: Mon, 27 Jul 2026 10:16:48 -0700 Subject: [PATCH 1/7] HOTFIX ENG-1081: give the completion verifier room to answer (silent task death, 25% of verdicts) (#278) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(session): give the completion verifier room to answer (ENG-1081) The verdict call is a forced tool call with max_tokens=256. Models that narrate before acting — the open-weight aliases MindsHub serves through Fireworks (mindshub_air/kimi, deepseek) — spend the whole budget on plain prose and never reach the tool call, so `generate_object_code` raises and the fail-safe converts it into a silent "task complete": the turn ends with no message and the agent looks like it died mid-task. Measured in prod Langfuse over 2026-07-14..07-27: 171 of 687 verdict calls returned no tool call (24.9%; 37.1% for external users), 145/147 of them on mindshub_air, every single one with completion_tokens == max_tokens exactly. Not new with ENG-716 — the old free-text verifier also passed max_tokens=256 and 21% of its verdicts were unparseable for the same reason; ENG-716 only changed the consequence from "keep working" to "stop silently". - Verifier budget 256 -> 2048, with one retry at 4096 when truncated. 2048 is measured: mindshub_air spent 1,654 output tokens even with the no-preamble instruction, so 1024 was not enough (1/3 at 1024, 3/3 at 2048). - Verifier prompt now asks for the tool call first, no preamble. Needed alongside the budget, not instead of it (0/3 at 256 even with it). - StructuredOutputError carries `truncated` so a blown budget is retryable while a genuine failure isn't. Detected by output-token count, because the gateway reports finish_reason "stop" at the cap for most aliases (ENG-1082); "length"/"max_tokens" are honoured when reported. - Truncation is logged distinctly from other verifier failures, which all logged identically as "verifier unavailable" before — the reason a 25% failure rate went unseen for 10 days. Co-Authored-By: Claude Opus 5 (1M context) * fix(session): bound the verifier retry to once per session (ENG-1081) Self-review follow-up. If the truncation retry is itself truncated, the model won't fit a verdict at any budget we're willing to pay for — so a chronically-narrating model would otherwise cost 2048+4096 wasted output tokens on every turn instead of the old 256. Latch it after the first proof and stop buying the retry for the rest of the session; the first attempt still runs, so a transient truncation still recovers. Also drops a pointless f-string in the verifier prompt concatenation. Co-Authored-By: Claude Opus 5 (1M context) * fix(llm): share the truncation classifier with the scratchpad path (ENG-1081) Adversarial-review follow-up. Four fixes: 1. The detection lived in `client.py`, so the sync twin in the scratchpad subprocess (`_ScratchpadLLM.generate_object`) still raised a blind `ValueError("LLM did not return structured output.")` — the same silent, causeless failure this branch set out to remove, on the path that is handed to model-written scratchpad code with a caller-chosen max_tokens. `structured.py` exists to keep those two paths in lockstep, so the classifier moves there as `raise_missing_tool_call` and both call it. 2. Drop the per-session retry latch. It assumed a double truncation proves the model can never fit a verdict. Measured: 16 identical verdict calls to mindshub_air spent 245..1654 output tokens — a 6.7x spread for the same request — so one truncation is a tail sample, not proof about the next turn, and latching the retry off on that evidence would reintroduce silent stops. The retry is cheap because it only fires when the first attempt truncates (0 of 12 at 2048). 3. The session tests hand-built StructuredOutputError, so they would have passed even with the detection broken. Added an end-to-end test wiring a fake provider through the real LLMClient into the session; verified by mutation (breaking the classifier fails it, and 4 others). 4. The plan_stream fake alternated on `call_count % 2`, coupling it to the tool loop's internal call count; a third call in a turn would desync it and the test would assert the wrong thing. Now keyed off turn state. Also records the measured distribution next to the budget constant, so the 2048 number carries its evidence (median ~290, max 1654, ~1.24x headroom) instead of a single sample. Co-Authored-By: Claude Opus 5 (1M context) * fix(llm): treat a tool call truncated mid-arguments as truncation too (ENG-1081) Third review pass found the same bug behind a second door. The budget can run out *inside* the forced tool call's JSON, not only in the prose before it. That arrives as a non-empty but incomplete tool call — safe_parse_tool_input repairs what it can and sets parse_error — so the missing-call check never sees it, unwrap_structured_response raises a bare ValidationError, the verifier's `except Exception` treats it as a hard failure, skips the retry, and lands on exactly the silent-stop fail-safe this branch exists to remove. Given output length varies ~6.7x per call, a model that narrates ~2000 tokens and then starts emitting the call is precisely the case that lands here. - `looks_truncated` extracted so the same evidence rule covers both shapes. - `raise_missing_tool_call` -> `raise_unusable_tool_call`, message adapts to "did not return" vs "returned an unusable", and now typed NoReturn. - Both the async client and the scratchpad's sync twin classify a failed unwrap as truncation *only* when the response shows it; a malformed call with budget to spare still propagates as a validation error, so genuine schema bugs stay visible instead of buying a hopeless retry. - Parity test upgraded from a substring grep (which a comment could satisfy) to an AST check that the call really is inside the sync `generate_object`. Tests: 15 in the file, mutation-verified — breaking `looks_truncated` fails 7 of them, including both end-to-end shapes. Full unit suite 1329 passed with the same 2 pre-existing scratchpad-exec failures; all 36 e2e scenarios pass. Co-Authored-By: Claude Opus 5 (1M context) * test(e2e): cover the truncation retry end-to-end; log the verifier's error cause Tool-assisted review pass. - The e2e stub could not express a truncated response (usage.completion_tokens was hardcoded to 10), so the retry had no coverage on the real subprocess path. `_Response.output_tokens` + `queue_verification_truncated()` emulate what the gateway actually returns — prose, no tool call, completion_tokens at the cap, and `finish_reason: "stop"` anyway (ENG-1082). New scenario asserts the session makes a second verdict call with a larger budget and honours the retried verdict instead of stopping silently. - The verifier's generic `except Exception` logged a bare `verdict=ERROR` with no cause — the same undiagnosable swallow this branch exists to fix, and the same thing I flagged on #276. Now logs the exception type and message. Full unit suite 1329 passed; all 37 e2e scenarios pass. Co-Authored-By: Claude Opus 5 (1M context) * fix(session): keep conversation content out of verifier logs (review) @pnewsam, Major on #277: logging `str(exc)` can copy conversation content into ordinary application logs. Correct — a pydantic ValidationError embeds the rejected `input_value`, which for the verifier is model-generated text derived from the user's conversation, and provider exceptions can carry response bodies. It also contradicted this PR's own security claim. - `_safe_error_detail(exc)` emits the exception type plus, for validation errors, the failing field locations and error codes only — via `errors(include_input=False, include_url=False, include_context=False)`, and reading only `loc`/`type` even on the v1 fallback, so `msg`/`input`/`ctx` can never be quoted. Provider errors reduce to type + status_code. - Same leak class found in a pre-existing line: the summary log wrote `reason=%s`, the model's free-text justification, on EVERY verdict — a broader exposure than the one reported. Dropped; status and counters stay, and the reason remains in the Langfuse trace where conversation content belongs. - 3 regression tests, including one asserting a planted value never reaches the log detail. Also shortened the long comment blocks flagged in review (structured.py, session.py budget/prompt constants, client.py, scratchpad_boot.py) — same facts, roughly half the lines. 1332 unit tests pass; all 37 e2e scenarios pass. Co-Authored-By: Claude Opus 5 (1M context) * fix(session): _safe_error_detail must never raise (hotfix hardening) It runs inside the verifier's `except` handler, so an exception escaping it would convert a gracefully-handled verifier failure into a dead turn — the exact failure mode this branch exists to remove. A custom exception can expose `status_code`/`errors` as a property that raises, so the attribute reads are wrapped too. Degrades to the exception type name, never a crash and never the message. Found while assessing hotfix risk for #278. Co-Authored-By: Claude Opus 5 (1M context) --------- Co-authored-by: Claude Opus 5 (1M context) --- anton/core/backends/scratchpad_boot.py | 23 +- anton/core/llm/client.py | 34 +- anton/core/llm/provider.py | 30 ++ anton/core/llm/structured.py | 84 +++- anton/core/session.py | 137 +++++- tests/e2e/scenarios/test_loop_safety.py | 39 ++ tests/e2e/stub_server.py | 27 +- tests/test_verifier_truncation.py | 562 ++++++++++++++++++++++++ 8 files changed, 910 insertions(+), 26 deletions(-) create mode 100644 tests/test_verifier_truncation.py diff --git a/anton/core/backends/scratchpad_boot.py b/anton/core/backends/scratchpad_boot.py index a949752b..d7eaf108 100644 --- a/anton/core/backends/scratchpad_boot.py +++ b/anton/core/backends/scratchpad_boot.py @@ -217,6 +217,8 @@ def generate_object( """ from anton.core.llm.structured import ( build_structured_tool, + looks_truncated, + raise_unusable_tool_call, unwrap_structured_response, ) @@ -233,11 +235,24 @@ def generate_object( ) if not response.tool_calls: - raise ValueError("LLM did not return structured output.") + # Same classification as the async path (ENG-1081). + # Nothing retries here, but the message reaches the model as + # a traceback, so "you ran out of budget" is actionable + # where "did not return structured output" was not. + raise_unusable_tool_call( + response, tool_name=tool["name"], budget=max_tokens + ) - return unwrap_structured_response( - response.tool_calls[0].input, validator_class, is_list - ) + try: + return unwrap_structured_response( + response.tool_calls[0].input, validator_class, is_list + ) + except Exception: + if looks_truncated(response, max_tokens): + raise_unusable_tool_call( + response, tool_name=tool["name"], budget=max_tokens + ) + raise _scratchpad_llm_instance = _ScratchpadLLM() diff --git a/anton/core/llm/client.py b/anton/core/llm/client.py index 88c96629..30b79bf0 100644 --- a/anton/core/llm/client.py +++ b/anton/core/llm/client.py @@ -199,28 +199,41 @@ async def _generate_object_with( """ from anton.core.llm.structured import ( build_structured_tool, + looks_truncated, + raise_unusable_tool_call, unwrap_structured_response, ) tool, validator_class, is_list = build_structured_tool(schema_class) + budget = max_tokens or self._max_tokens + response = await provider.complete( model=model, system=system, messages=messages, tools=[tool], tool_choice={"type": "tool", "name": tool["name"]}, - max_tokens=max_tokens or self._max_tokens, + max_tokens=budget, ) if not response.tool_calls: - raise ValueError( - f"LLM did not return a tool call for forced schema {tool['name']}." - ) + # Shared with the scratchpad's sync twin so both paths classify the + # failure identically — see `raise_unusable_tool_call` (ENG-1081). + raise_unusable_tool_call(response, tool_name=tool["name"], budget=budget) - return unwrap_structured_response( - response.tool_calls[0].input, validator_class, is_list - ) + try: + return unwrap_structured_response( + response.tool_calls[0].input, validator_class, is_list + ) + except Exception: + # The budget can also run out *inside* the tool call's JSON, so + # validation fails here instead. Same cause, same retry (ENG-1081). + if looks_truncated(response, budget): + raise_unusable_tool_call( + response, tool_name=tool["name"], budget=budget + ) + raise async def generate_object( self, @@ -263,8 +276,11 @@ async def generate_object( input was a list annotation. Raises: - ValueError: If the LLM fails to produce a tool call (rare — - forced tool_choice usually prevents this). + StructuredOutputError: If the LLM fails to produce a tool call. + A ``ValueError`` subclass, so callers that catch + ``ValueError`` are unaffected. Check ``.truncated`` to tell a + blown ``max_tokens`` budget (retry with more room) from a + genuine failure (ENG-1081). pydantic.ValidationError: If the tool's input doesn't match the schema. diff --git a/anton/core/llm/provider.py b/anton/core/llm/provider.py index 7831cd2e..719ba67b 100644 --- a/anton/core/llm/provider.py +++ b/anton/core/llm/provider.py @@ -312,6 +312,36 @@ class TokenLimitExceeded(Exception): """Raised when the LLM returns 429 due to billing/token limits.""" +class StructuredOutputError(ValueError): + """Raised when a forced-tool-call structured-output call yields no usable call. + + ``truncated`` says whether a retry can help: ``True`` means the model spent + the ``max_tokens`` budget narrating before it reached the tool call (the + narrating aliases — ``mindshub_air``/``kimi``, ``deepseek``, ``qwen`` — do + this deterministically under a tight budget, ENG-1081); ``False`` means the + provider errored, refused, or returned nothing, and a bigger budget won't + fix it. See `structured.looks_truncated` for how that's decided. + + Subclasses ``ValueError`` so call sites catching the documented ``ValueError`` + from ``generate_object``/``generate_object_code`` keep working. + """ + + def __init__( + self, + message: str, + *, + truncated: bool = False, + output_tokens: int = 0, + max_tokens: int = 0, + stop_reason: str | None = None, + ): + super().__init__(message) + self.truncated = truncated + self.output_tokens = output_tokens + self.max_tokens = max_tokens + self.stop_reason = stop_reason + + class TransientProviderError(ConnectionError): """Raised when the provider fails in a way that a retry might fix. diff --git a/anton/core/llm/structured.py b/anton/core/llm/structured.py index 02372c8e..4f497eb1 100644 --- a/anton/core/llm/structured.py +++ b/anton/core/llm/structured.py @@ -27,7 +27,89 @@ from __future__ import annotations -from typing import Any +from typing import Any, NoReturn + +from .provider import StructuredOutputError + + +def looks_truncated(response, budget: int) -> bool: + """True if `response` was cut off by the `max_tokens` budget. + + Token count first, because the MindsHub gateway reports + ``finish_reason: "stop"`` at the cap for most aliases (ENG-1082) — the + standard ``"length"`` check can't be relied on there. Both dialects are + honoured when reported: OpenAI says ``"length"``, Anthropic ``"max_tokens"``. + No usage information → ``False``; without evidence we don't buy a retry. + """ + usage = getattr(response, "usage", None) + output_tokens = getattr(usage, "output_tokens", 0) or 0 + stop_reason = getattr(response, "stop_reason", None) + return stop_reason in ("length", "max_tokens") or ( + budget > 0 and output_tokens >= budget + ) + + +def raise_unusable_tool_call(response, *, tool_name: str, budget: int) -> NoReturn: + """Raise `StructuredOutputError` explaining *why* the tool call is unusable. + + Covers both shapes of the same underlying problem: + + - **No tool call at all** — the model narrated in plain ``content`` until + the budget ran out. + - **A damaged tool call** — the budget ran out *inside* the call's JSON + arguments, so ``safe_parse_tool_input`` salvaged a partial dict and set + ``parse_error``, and validation then fails. Without this, that case + surfaces as a bare ``ValidationError``, which reads like a schema bug and + never gets the retry a truncation deserves (ENG-1081). + + Lives here, next to `build_structured_tool`/`unwrap_structured_response`, + because both structured-output paths must classify the failure the same + way — the async `LLMClient._generate_object_with` and the sync + `_ScratchpadLLM.generate_object` in the scratchpad subprocess. Keeping it + in one of the two callers is how they drift (ENG-1081). + + A forced tool call can come back empty for two very different reasons: + + - **Truncated** — the model narrated in plain ``content`` and ran out of + ``budget`` before reaching the call. Retrying with more room usually + works. Models served through MindsHub's Fireworks aliases + (``mindshub_air``/``kimi``, ``deepseek``) narrate before acting, so a + tight budget fails them. + - **Anything else** — the provider errored, refused, or returned nothing. + A bigger budget won't help. + + See `looks_truncated` for how truncation is detected. + + Args: + response: The provider's ``LLMResponse``. + tool_name: Name of the forced tool, for the message. + budget: The ``max_tokens`` the call was given. + + Raises: + StructuredOutputError: Always. ``.truncated`` carries the verdict. + """ + usage = getattr(response, "usage", None) + output_tokens = getattr(usage, "output_tokens", 0) or 0 + stop_reason = getattr(response, "stop_reason", None) + truncated = looks_truncated(response, budget) + what = ( + "returned an unusable tool call for" + if getattr(response, "tool_calls", None) + else "did not return a tool call for" + ) + detail = ( + f" (truncated: {output_tokens}/{budget} output tokens spent before the " + "call was complete)." + if truncated + else "." + ) + raise StructuredOutputError( + f"LLM {what} forced schema {tool_name}{detail}", + truncated=truncated, + output_tokens=output_tokens, + max_tokens=budget, + stop_reason=stop_reason, + ) def build_structured_tool(schema_class) -> tuple[dict, type, bool]: diff --git a/anton/core/session.py b/anton/core/session.py index 77eece9a..c6920bba 100644 --- a/anton/core/session.py +++ b/anton/core/session.py @@ -39,6 +39,7 @@ StreamTaskProgress, StreamTextDelta, StreamToolResult, + StructuredOutputError, TokenLimitExceeded, ToolCall, TransientProviderError, @@ -158,6 +159,81 @@ class _VerifierVerdict(BaseModel): reason: str = Field(description="One brief sentence explaining the verdict.") +# Output budgets for the verdict call: first attempt, then the retry used when +# it comes back truncated (ENG-1081). +# +# Models that narrate before acting (`mindshub_air`/`kimi`, `deepseek`, `qwen`) +# spend the budget on prose and never reach the forced tool call. The original +# 256 truncated them on essentially every call — 98.6% of `mindshub_air` verdicts +# in prod returned no tool call, which the fail-safe below turned into a silent +# "task complete". +# +# 2048 is sized from a measured distribution, not one sample: 16 identical calls +# spanned 245–1654 output tokens (median ~290). That 6.7x per-call spread is also +# why there is no "this model can't do it" latch — one truncation is a tail +# sample, not proof about the next turn — and why the 4096 retry exists rather +# than a single bigger budget. 1024 was measurably too small. Nothing pays for +# headroom it doesn't use; first-party models answer in 43–115 tokens either way. +_VERIFIER_TOKEN_BUDGETS = (2048, 4096) + +# Appended to the verifier system prompt. Shortens the preamble on narrating +# models but is not sufficient alone (0/3 at 256 with it), so it pairs with the +# budgets above. Tool name comes off the schema class so it can't go stale. +_VERIFIER_NO_PREAMBLE = ( + f"Call the {_VerifierVerdict.__name__} tool immediately as your first " + "action. Do not think out loud, restate the conversation, or explain your " + "reasoning before calling it — put your one-sentence justification in the " + "tool's `reason` field." +) + + +def _safe_error_detail(exc: BaseException) -> str: + """Describe an exception for logs without copying model or user content. + + Neither ``str(exc)`` nor ``repr(exc)`` is safe here: a Pydantic + ``ValidationError`` embeds the rejected ``input_value`` — for the verifier + that's model-generated text derived from the user's conversation — and + provider exceptions can carry response bodies. Emits the exception type, + plus for validation errors the field locations and error codes only, which + is what actually identifies the failure. + """ + # This runs *inside* an `except` handler, so it must not raise: an exception + # escaping here would turn a gracefully-handled verifier failure into a dead + # turn. Everything below is therefore wrapped, including the attribute reads + # (a custom exception can expose `status_code`/`errors` as a property that + # raises). + try: + name = type(exc).__name__ + except Exception: # pragma: no cover — defensive + return "unavailable" + try: + status = getattr(exc, "status_code", None) + if status is not None: + return f"{name}(status={status})" + errors = getattr(exc, "errors", None) + if callable(errors): + try: + details = errors( + include_input=False, include_url=False, include_context=False + ) + except TypeError: # pydantic v1 has no include_* kwargs + details = errors() + # Only `loc` and `type` are ever read — `msg`/`input`/`ctx` can + # quote the rejected value. + fields = ",".join( + f"{'.'.join(str(p) for p in e.get('loc') or ())}:{e.get('type', '?')}" + for e in details + ) + if fields: + return f"{name}({fields})" + except Exception: + # Anything odd about this exception object — a property that raises, an + # `errors()` that misbehaves — degrades to the type name, never a crash + # and never the message. + pass + return name + + def _render_tool_result_content(content, cap: int) -> str: """Render a tool_result's content as bounded plain text. @@ -2656,25 +2732,64 @@ async def _stream_and_handle_tools( "You are a task-completion verifier. Decide whether the user's " "request is complete, the assistant is waiting on the user, the work " "is unfinished, or the assistant is blocked. Follow the status " - "definitions exactly." + "definitions exactly.\n\n" + # Models that narrate before acting spend the whole budget on + # prose and never reach the tool call (ENG-1081). Asking for the + # call first shortens the preamble; it does not eliminate it, + # which is why the budget below is generous as well. + + _VERIFIER_NO_PREAMBLE ) - try: - verdict = await self._llm.generate_object_code( - _VerifierVerdict, - system=verifier_system, - messages=verify_messages, - max_tokens=256, - ) + verdict = None + for attempt, budget in enumerate(_VERIFIER_TOKEN_BUDGETS): + try: + verdict = await self._llm.generate_object_code( + _VerifierVerdict, + system=verifier_system, + messages=verify_messages, + max_tokens=budget, + ) + break + except StructuredOutputError as exc: + # A truncated verdict is a budget problem, not a verdict: the + # model narrated past `budget` before it reached the tool call + # (ENG-1081). Retry with more room. Any other structured-output + # failure won't be fixed by a bigger budget, so don't pay for it. + retrying = exc.truncated and attempt + 1 < len( + _VERIFIER_TOKEN_BUDGETS + ) + _verifier_log.info( + "completion-verifier verdict=%s budget=%d output_tokens=%d " + "stop_reason=%s retrying=%s", + "TRUNCATED" if exc.truncated else "NO_TOOL_CALL", + budget, exc.output_tokens, exc.stop_reason, retrying, + ) + if not retrying: + break + except Exception as exc: + # Enough to tell the failure modes apart — four used to + # collapse into one "verifier unavailable" line — but never + # the exception message, which can carry conversation + # content (ENG-1081). + _verifier_log.info( + "completion-verifier verdict=ERROR budget=%d error=%s", + budget, _safe_error_detail(exc), + ) + break + + if verdict is not None: status = verdict.status reason = verdict.reason.strip() - except Exception: + else: # Verifier failed — fail safe by treating the turn as done rather # than forcing a continuation the user never asked for. status, reason = "COMPLETE", "verifier unavailable" _verifier_log.info( - "completion-verifier verdict=%s continuation=%d/%d tool_rounds=%d reason=%s", - status, continuation, self._max_continuations, tool_round, reason, + # No `reason` — it's the model's free-text justification, derived + # from the user's conversation, and this is an ordinary app log. + # It stays in the Langfuse trace, where that content belongs. + "completion-verifier verdict=%s continuation=%d/%d tool_rounds=%d", + status, continuation, self._max_continuations, tool_round, ) if status in ("COMPLETE", "WAITING"): diff --git a/tests/e2e/scenarios/test_loop_safety.py b/tests/e2e/scenarios/test_loop_safety.py index 3f83dcfe..642b5738 100644 --- a/tests/e2e/scenarios/test_loop_safety.py +++ b/tests/e2e/scenarios/test_loop_safety.py @@ -50,6 +50,45 @@ def test_continuation_limit_respected(cfg, stub, tmp_path): ), f"Budget-exhausted message not found. Request count: {stub.request_count}" +@pytest.mark.stub_only +def test_truncated_verdict_is_retried_not_silently_dropped(cfg, stub, tmp_path): + # ENG-1081: the verdict call is a forced tool call, and models that narrate + # before acting (mindshub_air/kimi, deepseek) spend the whole budget on prose + # and never reach it. That used to raise, get swallowed as a fake COMPLETE, + # and end the turn with no message at all. The session must retry with a + # bigger budget and then honour the real verdict. + stub.queue_tool_call("scratchpad", {"action": "exec", "name": "c", "code": "print(1)"}) + stub.queue_text("Ran the script. TURN_TEXT") + stub.queue_verification_truncated() # first budget: narrated to the cap + stub.queue_verification_incomplete("the summary table is still missing") + stub.queue_text("Finished the summary table. RETRY_VERDICT_HONOURED") + stub.queue_verification_ok() + result = run_anton(["--folder", str(tmp_path)], ["build me a summary", "exit"], + env=base_env(stub), timeout=cfg.timeout(60)) + + assert_exit_ok(result) + assert_not_output(result, "Traceback (most recent call last)") + # The retried verdict (INCOMPLETE) drove a continuation, so the turn kept + # working instead of stopping silently on a fabricated COMPLETE. + assert_output(result, "RETRY_VERDICT_HONOURED") + verdict_calls = [ + r for r in stub.requests + if "task-completion verifier" in json.dumps(r.get("messages", [])) + or "task-completion verifier" in str(r.get("system", "")) + ] + assert len(verdict_calls) >= 2, ( + f"expected a retried verdict call, saw {len(verdict_calls)}. " + f"Request count: {stub.request_count}" + ) + # The OpenAI provider sends the budget as `max_completion_tokens`. + budgets = [ + r.get("max_completion_tokens") or r.get("max_tokens") for r in verdict_calls[:2] + ] + assert budgets[0] < budgets[1], ( + f"the retry must ask for more room than the first attempt, got {budgets}" + ) + + @pytest.mark.stub_only def test_waiting_verdict_stops_without_continuation(cfg, stub, tmp_path): # ENG-716: a tool-using turn that ends by asking the user a question must diff --git a/tests/e2e/stub_server.py b/tests/e2e/stub_server.py index c8c20125..b97bfeff 100644 --- a/tests/e2e/stub_server.py +++ b/tests/e2e/stub_server.py @@ -29,6 +29,11 @@ class _Response: tool_calls: list[dict] = field(default_factory=list) # None = honour request's `stream` flag; True/False = override force_streaming: bool | None = None + # Reported as `usage.completion_tokens`. Set it equal to the request's + # `max_tokens` to emulate a truncated response — which is how the real + # MindsHub gateway presents one, since it still says + # `finish_reason: "stop"` at the cap (ENG-1082). + output_tokens: int = 10 class StubServer: @@ -84,6 +89,22 @@ def _queue_verdict(self, status: str, reason: str) -> "StubServer": )) return self + def queue_verification_truncated(self, output_tokens: int = 2048) -> "StubServer": + """Queue a verdict call that narrated instead of calling the tool and ran + out of budget doing it (ENG-1081). + + Text, no tool call, and `completion_tokens` equal to the verifier's first + budget — exactly what `mindshub_air`/`kimi`/`deepseek` return. The + session should retry with the larger budget rather than treating this as + a verdict. + """ + self._queue.put(_Response( + content="Let me analyze this conversation carefully. The user asked for", + force_streaming=False, + output_tokens=output_tokens, + )) + return self + def queue_verification_ok(self) -> "StubServer": """Queue a COMPLETE verifier verdict.""" return self._queue_verdict("COMPLETE", "task is done.") @@ -243,7 +264,11 @@ def _send_json(handler: BaseHTTPRequestHandler, resp: _Response) -> None: "message": message, "finish_reason": finish_reason, }], - "usage": {"prompt_tokens": 10, "completion_tokens": 10, "total_tokens": 20}, + "usage": { + "prompt_tokens": 10, + "completion_tokens": resp.output_tokens, + "total_tokens": 10 + resp.output_tokens, + }, } body = json.dumps(data).encode() handler.send_response(200) diff --git a/tests/test_verifier_truncation.py b/tests/test_verifier_truncation.py new file mode 100644 index 00000000..748e6c52 --- /dev/null +++ b/tests/test_verifier_truncation.py @@ -0,0 +1,562 @@ +"""Completion-verifier truncation handling (ENG-1081). + +The verdict call is a forced tool call. Models that narrate before acting +(MindsHub's Fireworks aliases — `mindshub_air`/`kimi`, `deepseek`) spend the +output budget on plain prose and never reach the call, so a tight `max_tokens` +fails them deterministically: 98.6% of `mindshub_air` verdict calls in prod +returned no tool call, and the fail-safe turned each one into a silent +"task complete" with no message to the user. + +Two behaviours are covered here: + +1. `_generate_object_with` reports *why* there was no tool call, distinguishing a + blown budget (retryable) from a genuine failure — by token count, because the + MindsHub gateway reports `finish_reason: "stop"` at the cap (ENG-1082). +2. The verifier retries a truncated verdict once with a bigger budget, and does + NOT spend a retry on a failure a bigger budget can't fix. +""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from tests.conftest import make_mock_llm + +from anton.core.llm.client import LLMClient +from anton.core.llm.provider import ( + LLMResponse, + StreamComplete, + StructuredOutputError, + ToolCall, + Usage, +) +from anton.core.session import ( + _VERIFIER_TOKEN_BUDGETS, + ChatSession, + ChatSessionConfig, + _VerifierVerdict, +) + + +@pytest.fixture() +def workspace(): + # Keep scratchpad venvs inside the repo workspace (pytest runs sandboxed and + # can't write to the real home directory). + base = Path(__file__).resolve().parents[1] / ".pytest-workspace" + base.mkdir(parents=True, exist_ok=True) + return MagicMock(base=base) + + +def _text_response(text: str, output_tokens: int = 20, stop_reason: str = "end_turn") -> LLMResponse: + return LLMResponse( + content=text, + tool_calls=[], + usage=Usage(input_tokens=10, output_tokens=output_tokens), + stop_reason=stop_reason, + ) + + +def _scratchpad_response(text: str, code: str = "print(1)") -> LLMResponse: + return LLMResponse( + content=text, + tool_calls=[ToolCall( + id="tc_1", name="scratchpad", + input={"action": "exec", "name": "main", "code": code}, + )], + usage=Usage(input_tokens=10, output_tokens=20), + stop_reason="tool_use", + ) + + +class _FakeAsyncIter: + def __init__(self, items): + self._items = items + + def __aiter__(self): + return self + + async def __anext__(self): + if not self._items: + raise StopAsyncIteration + return self._items.pop(0) + + +# -------------------------------------------------------------------------- +# 1. The client reports *why* the tool call is missing. +# -------------------------------------------------------------------------- + + +def _client_with_response(response: LLMResponse) -> LLMClient: + provider = MagicMock() + provider.complete = AsyncMock(return_value=response) + return LLMClient( + planning_provider=provider, + planning_model="planner", + coding_provider=provider, + coding_model="coder", + ) + + +async def test_no_tool_call_at_the_cap_is_reported_as_truncated(): + """Prose that spends the whole budget == truncation, even though the + gateway calls it `finish_reason: "stop"` (ENG-1082).""" + llm = _client_with_response( + _text_response("Let me analyze this conversation carefully. The user...", + output_tokens=256, stop_reason="stop") + ) + + with pytest.raises(StructuredOutputError) as exc_info: + await llm.generate_object_code( + _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}], + max_tokens=256, + ) + + exc = exc_info.value + assert exc.truncated is True + assert exc.output_tokens == 256 + assert exc.max_tokens == 256 + # Callers that only know the documented ValueError still catch it. + assert isinstance(exc, ValueError) + + +@pytest.mark.parametrize("stop_reason", ["length", "max_tokens"]) +async def test_both_provider_dialects_for_truncation(stop_reason): + """The gateway/OpenAI dialect says "length"; AnthropicProvider passes + Anthropic's own "max_tokens" through raw. Both mean truncated.""" + llm = _client_with_response(_text_response("narrating…", output_tokens=100, + stop_reason=stop_reason)) + + with pytest.raises(StructuredOutputError) as exc_info: + await llm.generate_object_code( + _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}], + max_tokens=2048, + ) + + assert exc_info.value.truncated is True + + +async def test_stop_reason_length_is_honoured_below_the_cap(): + """Gemini reports truncation honestly and can return almost nothing — + trust `stop_reason` too, not only the token count.""" + llm = _client_with_response(_text_response("", output_tokens=9, stop_reason="length")) + + with pytest.raises(StructuredOutputError) as exc_info: + await llm.generate_object_code( + _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}], + max_tokens=256, + ) + + assert exc_info.value.truncated is True + + +async def test_short_empty_response_is_not_truncated(): + """A provider that returns nothing well inside the budget is a genuine + failure — a bigger budget won't fix it, so it must not be retried.""" + llm = _client_with_response(_text_response("", output_tokens=5, stop_reason="stop")) + + with pytest.raises(StructuredOutputError) as exc_info: + await llm.generate_object_code( + _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}], + max_tokens=256, + ) + + assert exc_info.value.truncated is False + + +def _damaged_tool_call_response(output_tokens: int) -> LLMResponse: + """A forced tool call that ran out of budget *inside* its JSON arguments. + + `safe_parse_tool_input` repairs what it can and sets `parse_error`, so the + call arrives non-empty but incomplete — here missing the required `reason`. + """ + return LLMResponse( + content="", + tool_calls=[ToolCall(id="tc_v", name="_VerifierVerdict", + input={"status": "WAITING"}, + parse_error="Unterminated string starting at: line 1")], + usage=Usage(input_tokens=100, output_tokens=output_tokens), + stop_reason="stop", + ) + + +async def test_tool_call_truncated_mid_arguments_is_retryable(): + """The budget can run out *inside* the tool call, not only before it. + + That arrives as a non-empty but incomplete tool call, so the missing-call + check doesn't see it and validation fails instead — which without this + would read as a schema bug, skip the retry, and land right back on the + silent-stop fail-safe this whole change exists to remove. + """ + llm = _client_with_response(_damaged_tool_call_response(output_tokens=2048)) + + with pytest.raises(StructuredOutputError) as exc_info: + await llm.generate_object_code( + _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}], + max_tokens=2048, + ) + + assert exc_info.value.truncated is True, "must be retryable, not a schema error" + assert "unusable tool call" in str(exc_info.value) + + +async def test_schema_mismatch_under_budget_is_not_disguised_as_truncation(): + """The mirror case: a malformed tool call with budget to spare is a real + schema failure. It must keep propagating as a validation error rather than + being relabelled truncation and buying a retry that cannot help.""" + llm = _client_with_response(_damaged_tool_call_response(output_tokens=60)) + + with pytest.raises(ValueError) as exc_info: + await llm.generate_object_code( + _VerifierVerdict, system="s", messages=[{"role": "user", "content": "m"}], + max_tokens=2048, + ) + + assert not isinstance(exc_info.value, StructuredOutputError), ( + "a genuine schema mismatch must stay a validation error" + ) + + +def test_shared_classifier_is_used_by_both_structured_paths(): + """The async client and the scratchpad's sync twin must classify identically. + + `raise_unusable_tool_call` is the single implementation both call — the whole + point of `structured.py`. Guarded here because the previous version of this + fix put the logic in `client.py`, which silently left the sync path + (exposed to model-written scratchpad code, with a caller-chosen budget) + raising a blind ValueError. + """ + import ast + + from anton.core.llm import structured + + assert hasattr(structured, "raise_unusable_tool_call") + # Parse the source rather than importing it — `scratchpad_boot` is a + # subprocess bootstrap and reads stdin at import time. AST rather than a + # substring search, so a passing mention in a comment or a stale import + # can't satisfy it: the call must be inside `generate_object` itself. + boot_src = ( + Path(__file__).resolve().parents[1] + / "anton" / "core" / "backends" / "scratchpad_boot.py" + ).read_text() + tree = ast.parse(boot_src) + generate_object = next( + (n for n in ast.walk(tree) + if isinstance(n, ast.FunctionDef) and n.name == "generate_object"), + None, + ) + assert generate_object is not None, "sync generate_object not found" + called = { + n.func.id for n in ast.walk(generate_object) + if isinstance(n, ast.Call) and isinstance(n.func, ast.Name) + } + assert "raise_unusable_tool_call" in called, ( + "the sync scratchpad path must call the shared classifier, not raise its own" + ) + assert "LLM did not return structured output." not in boot_src, ( + "the old blind ValueError should be gone from the sync path" + ) + + +def test_classifier_tolerates_a_response_without_usage(): + """Defensive: a provider response missing `usage` must not blow up the + classifier — it just can't prove truncation, which is the safe direction + (no retry bought without evidence).""" + from anton.core.llm.structured import raise_unusable_tool_call + + class _Bare: + content = "some prose" + + with pytest.raises(StructuredOutputError) as exc_info: + raise_unusable_tool_call(_Bare(), tool_name="_VerifierVerdict", budget=2048) + + assert exc_info.value.truncated is False + assert exc_info.value.output_tokens == 0 + + +# -------------------------------------------------------------------------- +# 2. The verifier retries a truncated verdict, once, with more room. +# -------------------------------------------------------------------------- + + +class _ToolThenText: + """`plan_stream` fake: one tool round per turn, then plain text. + + Keyed off "have I already used the tool this turn?" rather than a call + counter — a counter with `% 2` would silently desync if a turn ever made a + third `plan_stream` call (an extra tool round, an internal recovery retry), + and the test would then be asserting something other than it claims. + Call `next_turn()` between turns. + """ + + def __init__(self): + self.tool_used = False + + def next_turn(self) -> None: + self.tool_used = False + + def __call__(self, **kwargs): + if not self.tool_used: + self.tool_used = True + return _FakeAsyncIter([StreamComplete(response=_scratchpad_response("Running."))]) + return _FakeAsyncIter([StreamComplete(response=_text_response("Done."))]) + + +def _session_that_uses_a_tool(mock_llm, workspace) -> ChatSession: + """Session whose turn uses one tool, so the completion verifier runs.""" + mock_llm.plan_stream = _ToolThenText() + return ChatSession(ChatSessionConfig(llm_client=mock_llm, workspace=workspace)) + + +async def test_truncated_verdict_is_retried_with_a_bigger_budget(workspace): + """The narrating-model case: first budget truncates, the retry succeeds, and + the retried verdict is the one that counts — no silent 'COMPLETE'.""" + budgets: list[int] = [] + + async def fake_verdict(_schema, *, system, messages, max_tokens): + budgets.append(max_tokens) + if len(budgets) == 1: + raise StructuredOutputError( + "no tool call", truncated=True, output_tokens=max_tokens, + max_tokens=max_tokens, stop_reason="stop", + ) + return _VerifierVerdict(status="WAITING", reason="asked the user a question") + + mock_llm = make_mock_llm() + mock_llm.generate_object_code = AsyncMock(side_effect=fake_verdict) + + session = _session_that_uses_a_tool(mock_llm, workspace) + try: + async for _ in session.turn_stream("build me a dashboard"): + pass + finally: + await session.close() + + assert budgets == list(_VERIFIER_TOKEN_BUDGETS[:2]), ( + "a truncated verdict must be retried once, with the larger budget" + ) + # The verdict came from the retry (WAITING → a valid stop), so no + # "Continue working" continuation was injected. + assert not any( + "SYSTEM: Task verification determined this task is not yet complete" + in str(m.get("content", "")) + for m in session.history + ) + + +async def test_non_truncated_failure_is_not_retried(workspace): + """A failure a bigger budget can't fix costs exactly one call, then falls + through to the fail-safe.""" + calls: list[int] = [] + + async def fake_verdict(_schema, *, system, messages, max_tokens): + calls.append(max_tokens) + raise StructuredOutputError( + "no tool call", truncated=False, output_tokens=3, + max_tokens=max_tokens, stop_reason="stop", + ) + + mock_llm = make_mock_llm() + mock_llm.generate_object_code = AsyncMock(side_effect=fake_verdict) + + session = _session_that_uses_a_tool(mock_llm, workspace) + try: + async for _ in session.turn_stream("build me a dashboard"): + pass + finally: + await session.close() + + assert calls == [_VERIFIER_TOKEN_BUDGETS[0]], "must not pay for a hopeless retry" + + +async def test_attempts_are_bounded_and_repeat_each_turn(workspace): + """Truncation retries are bounded by the budget list — and are re-tried on a + later turn rather than latched off. + + Output length varies ~6.7x per call for an identical request, so one + truncation is a tail sample, not proof the model can never fit a verdict. + Latching the retry off on that evidence would bring silent stops back. + """ + budgets: list[int] = [] + + async def fake_verdict(_schema, *, system, messages, max_tokens): + budgets.append(max_tokens) + raise StructuredOutputError( + "no tool call", truncated=True, output_tokens=max_tokens, + max_tokens=max_tokens, stop_reason="stop", + ) + + mock_llm = make_mock_llm() + mock_llm.generate_object_code = AsyncMock(side_effect=fake_verdict) + + session = _session_that_uses_a_tool(mock_llm, workspace) + try: + async for _ in session.turn_stream("first turn"): + pass + assert budgets == list(_VERIFIER_TOKEN_BUDGETS), "bounded by the budget list" + + budgets.clear() + mock_llm.plan_stream.next_turn() + async for _ in session.turn_stream("second turn"): + pass + finally: + await session.close() + + assert budgets == list(_VERIFIER_TOKEN_BUDGETS), ( + "a later turn gets a fresh chance — the retry is not latched off" + ) + + +@pytest.mark.parametrize("first_failure", ["no_tool_call", "damaged_tool_call"]) +async def test_truncation_retry_through_the_real_client(workspace, first_failure): + """End-to-end: both shapes of a truncated verdict must flow through the real + `LLMClient` detection into a session-level retry — prose-at-the-cap with no + call at all, and a call cut off inside its own arguments. + + The other session tests raise `StructuredOutputError` by hand, so they would + pass even if the detection in `raise_unusable_tool_call` were wrong. This one + wires a fake *provider* through the real client, so provider → client → + session is the actual code path (ENG-747: don't hand-build error fixtures). + """ + calls: list[int] = [] + + async def fake_complete(*, model, system, messages, tools, tool_choice, max_tokens): + calls.append(max_tokens) + if len(calls) == 1: + if first_failure == "damaged_tool_call": + # Budget ran out *inside* the call's JSON arguments. + return _damaged_tool_call_response(output_tokens=max_tokens) + # Narrated right up to the ceiling, never reached the tool call — + # and the gateway calls that `finish_reason: "stop"` (ENG-1082). + return _text_response("Let me analyze this conversation carefully...", + output_tokens=max_tokens, stop_reason="stop") + return LLMResponse( + content="", + tool_calls=[ToolCall(id="tc_v", name="_VerifierVerdict", + input={"status": "WAITING", "reason": "asked the user"})], + usage=Usage(input_tokens=100, output_tokens=60), + stop_reason="tool_calls", + ) + + provider = MagicMock() + provider.complete = AsyncMock(side_effect=fake_complete) + real_client = LLMClient( + planning_provider=provider, planning_model="planner", + coding_provider=provider, coding_model="coder", + ) + + # Only the verdict call goes through the real client; the turn itself still + # uses the mock so the test stays a unit test. + mock_llm = make_mock_llm() + mock_llm.generate_object_code = real_client.generate_object_code + + session = _session_that_uses_a_tool(mock_llm, workspace) + try: + async for _ in session.turn_stream("build me a dashboard"): + pass + finally: + await session.close() + + assert calls == list(_VERIFIER_TOKEN_BUDGETS), ( + "the real client must classify prose-at-the-cap as truncated and the " + "session must retry it with the larger budget" + ) + # The retried verdict (WAITING) stands: no forced continuation. + assert not any( + "SYSTEM: Task verification determined this task is not yet complete" + in str(m.get("content", "")) + for m in session.history + ) + + +async def test_verifier_prompt_forbids_preamble(workspace): + """The no-preamble instruction reaches the model. It is not sufficient on its + own (0/3 at 256 with it), but it shortens the preamble enough to matter.""" + seen: dict = {} + + async def fake_verdict(_schema, *, system, messages, max_tokens): + seen["system"] = system + return _VerifierVerdict(status="COMPLETE", reason="done") + + mock_llm = make_mock_llm() + mock_llm.generate_object_code = AsyncMock(side_effect=fake_verdict) + + session = _session_that_uses_a_tool(mock_llm, workspace) + try: + async for _ in session.turn_stream("build me a dashboard"): + pass + finally: + await session.close() + + assert "immediately as your first action" in seen["system"] + assert "Do not think out loud" in seen["system"] + + +# -------------------------------------------------------------------------- +# 3. Verifier logs must not carry conversation content (review finding). +# -------------------------------------------------------------------------- + + +def test_error_detail_never_leaks_the_rejected_value(): + """`_safe_error_detail` must identify the failure without quoting content. + + A pydantic ValidationError's message embeds the rejected `input_value` — for + the verifier that's model-generated text derived from the user's + conversation — so `str(exc)` cannot go into ordinary application logs. + """ + import pydantic + + from anton.core.session import _safe_error_detail + + secret = "the user's bank balance is 12345" + try: + _VerifierVerdict.model_validate({"status": secret, "reason": secret}) + except pydantic.ValidationError as exc: + detail = _safe_error_detail(exc) + else: + raise AssertionError("expected a ValidationError") + + assert secret not in detail, f"rejected value leaked into the log detail: {detail!r}" + assert "12345" not in detail + # Still useful: names the exception type and which field failed. + assert "ValidationError" in detail + assert "status" in detail + + +def test_error_detail_of_a_provider_error_is_type_and_status_only(): + from anton.core.session import _safe_error_detail + + class _FakeAPIError(Exception): + status_code = 503 + + detail = _safe_error_detail(_FakeAPIError("upstream said: ")) + assert detail == "_FakeAPIError(status=503)" + assert "response body" not in detail + + +def test_error_detail_falls_back_to_the_type_name(): + from anton.core.session import _safe_error_detail + + detail = _safe_error_detail(RuntimeError("provider hiccup with conversation text")) + assert detail == "RuntimeError" + + +def test_error_detail_never_raises_from_inside_an_except_handler(): + """It runs inside `except`, so raising would turn a handled verifier failure + into a dead turn. Hostile exceptions must still produce a string.""" + from anton.core.session import _safe_error_detail + + class _Hostile(Exception): + @property + def status_code(self): + raise RuntimeError("boom") + + class _HostileErrors(Exception): + def errors(self, **kwargs): + raise RuntimeError("boom") + + # Degrades to the exception type — never raises, never the message. + assert _safe_error_detail(_Hostile()) == "_Hostile" + assert _safe_error_detail(_HostileErrors()) == "_HostileErrors" From 2e39ef6ffc8689805540bbede89fe239f12c6386 Mon Sep 17 00:00:00 2001 From: Hamish Fagg Date: Tue, 28 Jul 2026 16:51:45 +1200 Subject: [PATCH 2/7] feat(cloud_turn): static buffered turn entrypoint --- anton/cloud_turn/__init__.py | 0 anton/cloud_turn/__main__.py | 32 +++++++++++++++++++++++++++ anton/cloud_turn/contract.py | 34 +++++++++++++++++++++++++++++ tests/cloud_turn/__init__.py | 0 tests/cloud_turn/test_entrypoint.py | 21 ++++++++++++++++++ 5 files changed, 87 insertions(+) create mode 100644 anton/cloud_turn/__init__.py create mode 100644 anton/cloud_turn/__main__.py create mode 100644 anton/cloud_turn/contract.py create mode 100644 tests/cloud_turn/__init__.py create mode 100644 tests/cloud_turn/test_entrypoint.py diff --git a/anton/cloud_turn/__init__.py b/anton/cloud_turn/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/anton/cloud_turn/__main__.py b/anton/cloud_turn/__main__.py new file mode 100644 index 00000000..0c4d3582 --- /dev/null +++ b/anton/cloud_turn/__main__.py @@ -0,0 +1,32 @@ +from __future__ import annotations +import asyncio +import sys + +from anton.core.runtime import build_chat_session +from anton.cloud_turn.contract import TurnRequestV1, TurnResultV1 + + +async def run_turn(req: TurnRequestV1) -> TurnResultV1: + try: + session = await build_chat_session( + session_id=req.conversation_id, + workspace_path=req.workspace_path, + model=req.model, + ) + final_text = await session.turn(req.input) + return TurnResultV1(protocol_version=1, kind="turn_completed", final_text=final_text) + except Exception as exc: # dev skeleton: surface as terminal failure + return TurnResultV1(protocol_version=1, kind="turn_failed", error=repr(exc)) + + +def main() -> int: + raw = sys.stdin.read() + req = TurnRequestV1.from_json(raw) + result = asyncio.run(run_turn(req)) + sys.stdout.write(result.to_json() + "\n") + sys.stdout.flush() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/anton/cloud_turn/contract.py b/anton/cloud_turn/contract.py new file mode 100644 index 00000000..3ef3431f --- /dev/null +++ b/anton/cloud_turn/contract.py @@ -0,0 +1,34 @@ +from __future__ import annotations +import json +from dataclasses import dataclass, asdict + + +@dataclass +class TurnRequestV1: + protocol_version: int + conversation_id: str + input: str + workspace_path: str | None = None + model: str | None = None + + @staticmethod + def from_json(raw: str) -> "TurnRequestV1": + d = json.loads(raw) + return TurnRequestV1( + protocol_version=int(d["protocol_version"]), + conversation_id=str(d["conversation_id"]), + input=str(d["input"]), + workspace_path=d.get("workspace_path"), + model=d.get("model"), + ) + + +@dataclass +class TurnResultV1: + protocol_version: int + kind: str # "turn_completed" | "turn_failed" + final_text: str | None = None + error: str | None = None + + def to_json(self) -> str: + return json.dumps(asdict(self)) diff --git a/tests/cloud_turn/__init__.py b/tests/cloud_turn/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/cloud_turn/test_entrypoint.py b/tests/cloud_turn/test_entrypoint.py new file mode 100644 index 00000000..336a0aae --- /dev/null +++ b/tests/cloud_turn/test_entrypoint.py @@ -0,0 +1,21 @@ +import json +import pytest +from anton.cloud_turn.__main__ import run_turn +from anton.cloud_turn.contract import TurnRequestV1 + + +@pytest.mark.asyncio +async def test_run_turn_returns_completed(monkeypatch): + class FakeSession: + async def turn(self, user_input): + return f"echo: {user_input}" + + async def fake_build(**kwargs): + assert kwargs["session_id"] == "conv-1" + return FakeSession() + + monkeypatch.setattr("anton.cloud_turn.__main__.build_chat_session", fake_build) + req = TurnRequestV1(protocol_version=1, conversation_id="conv-1", input="hi") + result = await run_turn(req) + assert result.kind == "turn_completed" + assert result.final_text == "echo: hi" From 96b7dffd363b4f6a7c5d17f95243f79b697249a1 Mon Sep 17 00:00:00 2001 From: Hamish Fagg Date: Tue, 28 Jul 2026 18:10:17 +1200 Subject: [PATCH 3/7] feat(cloud_turn): stream events, read one-line request --- anton/cloud_turn/__main__.py | 31 +++++++++++++++++++---------- tests/cloud_turn/test_entrypoint.py | 31 +++++++++++++++++++++-------- 2 files changed, 43 insertions(+), 19 deletions(-) diff --git a/anton/cloud_turn/__main__.py b/anton/cloud_turn/__main__.py index 0c4d3582..54554de4 100644 --- a/anton/cloud_turn/__main__.py +++ b/anton/cloud_turn/__main__.py @@ -1,30 +1,39 @@ from __future__ import annotations import asyncio +import json import sys +from typing import AsyncIterator from anton.core.runtime import build_chat_session -from anton.cloud_turn.contract import TurnRequestV1, TurnResultV1 +from anton.core.llm.provider import StreamTextDelta +from anton.cloud_turn.contract import TurnRequestV1 -async def run_turn(req: TurnRequestV1) -> TurnResultV1: +async def stream_turn(req: TurnRequestV1) -> AsyncIterator[dict]: try: session = await build_chat_session( session_id=req.conversation_id, workspace_path=req.workspace_path, model=req.model, ) - final_text = await session.turn(req.input) - return TurnResultV1(protocol_version=1, kind="turn_completed", final_text=final_text) - except Exception as exc: # dev skeleton: surface as terminal failure - return TurnResultV1(protocol_version=1, kind="turn_failed", error=repr(exc)) + async for event in session.turn_stream(req.input): + if isinstance(event, StreamTextDelta): + yield {"kind": "delta", "text": event.text} + yield {"kind": "turn_completed"} + except Exception as exc: + yield {"kind": "turn_failed", "error": repr(exc)} + + +async def _run() -> None: + line = sys.stdin.readline() + req = TurnRequestV1.from_json(line) + async for ev in stream_turn(req): + sys.stdout.write(json.dumps(ev) + "\n") + sys.stdout.flush() def main() -> int: - raw = sys.stdin.read() - req = TurnRequestV1.from_json(raw) - result = asyncio.run(run_turn(req)) - sys.stdout.write(result.to_json() + "\n") - sys.stdout.flush() + asyncio.run(_run()) return 0 diff --git a/tests/cloud_turn/test_entrypoint.py b/tests/cloud_turn/test_entrypoint.py index 336a0aae..a7183ce5 100644 --- a/tests/cloud_turn/test_entrypoint.py +++ b/tests/cloud_turn/test_entrypoint.py @@ -1,14 +1,15 @@ -import json import pytest -from anton.cloud_turn.__main__ import run_turn +from anton.core.llm.provider import StreamTextDelta +from anton.cloud_turn.__main__ import stream_turn from anton.cloud_turn.contract import TurnRequestV1 @pytest.mark.asyncio -async def test_run_turn_returns_completed(monkeypatch): +async def test_stream_turn_emits_deltas_then_completed(monkeypatch): class FakeSession: - async def turn(self, user_input): - return f"echo: {user_input}" + async def turn_stream(self, user_input): + yield StreamTextDelta(text="he") + yield StreamTextDelta(text="llo") async def fake_build(**kwargs): assert kwargs["session_id"] == "conv-1" @@ -16,6 +17,20 @@ async def fake_build(**kwargs): monkeypatch.setattr("anton.cloud_turn.__main__.build_chat_session", fake_build) req = TurnRequestV1(protocol_version=1, conversation_id="conv-1", input="hi") - result = await run_turn(req) - assert result.kind == "turn_completed" - assert result.final_text == "echo: hi" + events = [ev async for ev in stream_turn(req)] + assert events == [ + {"kind": "delta", "text": "he"}, + {"kind": "delta", "text": "llo"}, + {"kind": "turn_completed"}, + ] + + +@pytest.mark.asyncio +async def test_stream_turn_failure_is_terminal(monkeypatch): + async def fake_build(**kwargs): + raise RuntimeError("boom") + monkeypatch.setattr("anton.cloud_turn.__main__.build_chat_session", fake_build) + req = TurnRequestV1(protocol_version=1, conversation_id="c", input="hi") + events = [ev async for ev in stream_turn(req)] + assert events[-1]["kind"] == "turn_failed" + assert "boom" in events[-1]["error"] From 154734de4e6ec01f0480c396b563b999cfe2c1d9 Mon Sep 17 00:00:00 2001 From: Hamish Fagg Date: Tue, 28 Jul 2026 22:01:51 +1200 Subject: [PATCH 4/7] build: Dockerfile + CI for minds-anton-scratchpad image Adds the container build for the sandbox pod image (anton + scratchpad_boot + python -m anton.cloud_turn), plus dev/staging/prod GitHub Actions that push to ECR minds-anton-scratchpad via mindsdb/github-actions/build-push-ecr (build + snyk scan; no deploy job, the image has no chart of its own). Installs anton into a UID-1000-owned venv with uv at /usr/local/bin/uv and boto3 (imported by anton but undeclared), and writes /usr/local/bin/scratchpad-boot.sh. Co-Authored-By: Claude Opus 4.8 (1M context) --- .dockerignore | 13 ++++ .github/workflows/scratchpad-dev-build.yml | 68 +++++++++++++++++++ .github/workflows/scratchpad-prod-build.yml | 52 ++++++++++++++ .../workflows/scratchpad-staging-build.yml | 51 ++++++++++++++ Dockerfile | 40 +++++++++++ 5 files changed, 224 insertions(+) create mode 100644 .dockerignore create mode 100644 .github/workflows/scratchpad-dev-build.yml create mode 100644 .github/workflows/scratchpad-prod-build.yml create mode 100644 .github/workflows/scratchpad-staging-build.yml create mode 100644 Dockerfile diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..74527a8b --- /dev/null +++ b/.dockerignore @@ -0,0 +1,13 @@ +# Version is pinned via SETUPTOOLS_SCM_PRETEND_VERSION in the Dockerfile, so .git is not needed. +.git +.venv +**/__pycache__ +**/*.pyc +**/*.pyo +dist +build +*.egg-info +.pytest_cache +.ruff_cache +.mypy_cache +docs diff --git a/.github/workflows/scratchpad-dev-build.yml b/.github/workflows/scratchpad-dev-build.yml new file mode 100644 index 00000000..0a80d8b8 --- /dev/null +++ b/.github/workflows/scratchpad-dev-build.yml @@ -0,0 +1,68 @@ +name: Scratchpad image - Dev build on PR + +# Builds and pushes the minds-anton-scratchpad image (the sandbox pod image) to ECR as +# development- when a PR carries a deploy label. The scratchpad-controller references +# the resulting tag via SCRATCHPAD_CONTROLLER__SCRATCHPAD_IMAGE; there is no Helm chart for +# this image, so this workflow only builds and scans (no deploy job). + +on: + pull_request: + types: [opened, reopened, synchronize, labeled] + +defaults: + run: + shell: bash + +concurrency: + group: ${{ github.workflow_ref }} + cancel-in-progress: true + +jobs: + + get-deploy-labels: + runs-on: mdb-dev + outputs: + deploy-envs: ${{ steps.get-labels.outputs.deploy-envs }} + steps: + - name: Pull MindsDB Github Actions + uses: actions/checkout@v4 + with: + repository: mindsdb/github-actions + path: github-actions + - id: get-labels + uses: ./github-actions/get-deploy-labels + + build: + runs-on: mdb-dev + needs: [get-deploy-labels] + if: needs.get-deploy-labels.outputs.deploy-envs != '[]' + outputs: + image: ${{ steps.build.outputs.image }} + env: + AWS_REGION: us-east-1 + steps: + - name: Checkout + uses: actions/checkout@v4 + - name: Pull MindsDB Github Actions + uses: actions/checkout@v4 + with: + repository: mindsdb/github-actions + path: github-actions + - uses: ./github-actions/setup-env + - id: build + uses: ./github-actions/build-push-ecr + with: + # Override the repo-slug default so the image lands at .../minds-anton-scratchpad. + module-name: minds-anton-scratchpad + build-for-environment: development + + scan: + runs-on: mdb-dev + needs: [build] + name: Scan minds-anton-scratchpad image + steps: + - uses: actions/checkout@v4 + - uses: mindsdb/github-actions/snyk-docker-scan@main + with: + image: ${{ needs.build.outputs.image }} + snyk-token: ${{ secrets.SNYK_TOKEN }} diff --git a/.github/workflows/scratchpad-prod-build.yml b/.github/workflows/scratchpad-prod-build.yml new file mode 100644 index 00000000..46b1ff81 --- /dev/null +++ b/.github/workflows/scratchpad-prod-build.yml @@ -0,0 +1,52 @@ +name: Scratchpad image - Prod build on release + +# Pushes minds-anton-scratchpad as production- when a GitHub Release is published +# (or via manual dispatch). Build + scan only; the scratchpad-controller consumes the tag. + +on: + release: + types: [published] + workflow_dispatch: + +defaults: + run: + shell: bash + +concurrency: + group: ${{ github.workflow_ref }} + cancel-in-progress: true + +jobs: + + build: + runs-on: mdb-dev + outputs: + image: ${{ steps.build.outputs.image }} + env: + AWS_REGION: us-east-1 + steps: + - name: Checkout + uses: actions/checkout@v4 + - name: Pull MindsDB Github Actions + uses: actions/checkout@v4 + with: + repository: mindsdb/github-actions + path: github-actions + - uses: ./github-actions/setup-env + - id: build + uses: ./github-actions/build-push-ecr + with: + module-name: minds-anton-scratchpad + build-for-environment: production + image-ref: ${{ env.CI_REF_SLUG }} + + scan: + runs-on: mdb-dev + needs: [build] + name: Scan minds-anton-scratchpad image + steps: + - uses: actions/checkout@v4 + - uses: mindsdb/github-actions/snyk-docker-scan@main + with: + image: ${{ needs.build.outputs.image }} + snyk-token: ${{ secrets.SNYK_TOKEN }} diff --git a/.github/workflows/scratchpad-staging-build.yml b/.github/workflows/scratchpad-staging-build.yml new file mode 100644 index 00000000..393f9973 --- /dev/null +++ b/.github/workflows/scratchpad-staging-build.yml @@ -0,0 +1,51 @@ +name: Scratchpad image - Staging build on push to main + +# Pushes minds-anton-scratchpad as staging- (plus the moving `staging` and `latest` tags) +# on every push to main. Build + scan only; the scratchpad-controller consumes the tag. + +on: + push: + branches: + - main + +defaults: + run: + shell: bash + +concurrency: + group: ${{ github.workflow_ref }} + cancel-in-progress: true + +jobs: + + build: + runs-on: mdb-dev + outputs: + image: ${{ steps.build.outputs.image }} + env: + AWS_REGION: us-east-1 + steps: + - name: Checkout + uses: actions/checkout@v4 + - name: Pull MindsDB Github Actions + uses: actions/checkout@v4 + with: + repository: mindsdb/github-actions + path: github-actions + - uses: ./github-actions/setup-env + - id: build + uses: ./github-actions/build-push-ecr + with: + module-name: minds-anton-scratchpad + build-for-environment: staging + + scan: + runs-on: mdb-dev + needs: [build] + name: Scan minds-anton-scratchpad image + steps: + - uses: actions/checkout@v4 + - uses: mindsdb/github-actions/snyk-docker-scan@main + with: + image: ${{ needs.build.outputs.image }} + snyk-token: ${{ secrets.SNYK_TOKEN }} diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 00000000..76675180 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,40 @@ +# minds-anton-scratchpad: the sandbox pod image (anton + scratchpad boot + cloud_turn entrypoint). +# Consumed by scratchpad-controller (SCRATCHPAD_CONTROLLER__SCRATCHPAD_IMAGE) and Minds. +# The controller execs `python -m anton.cloud_turn` (whole turn) or +# `/usr/local/bin/scratchpad-boot.sh` (single cell) inside a gVisor pod running as UID 1000. +FROM python:3.12-slim + +# uv is required at RUNTIME: scratchpad_boot shells out to `uv pip install` for missing packages +# via ANTON_UV_PATH=/usr/local/bin/uv, so uv must be present at exactly that path. +COPY --from=ghcr.io/astral-sh/uv:0.6.14 /uv /usr/local/bin/uv + +# hatch-vcs would otherwise need the .git history (excluded from the build context); the image +# only needs a valid version string, so pin one for the build. +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 \ + SETUPTOOLS_SCM_PRETEND_VERSION=2.0.0 \ + VIRTUAL_ENV=/opt/anton-venv \ + PATH=/opt/anton-venv/bin:/usr/local/bin:$PATH \ + UV_LINK_MODE=copy + +WORKDIR /app +COPY . /app + +# Install anton into a venv owned by UID 1000 so the non-root runtime can also `uv pip install` +# missing packages at turn time. boto3 is imported by anton at runtime but not declared as a +# dependency, so add it explicitly. +RUN uv venv "$VIRTUAL_ENV" \ + && uv pip install --no-cache . boto3 \ + && chown -R 1000:1000 "$VIRTUAL_ENV" + +# scratchpad-boot.sh: the single-cell entrypoint the controller execs (reads code + delimiter on stdin). +RUN printf '#!/bin/sh\nexec python -m anton.core.backends.scratchpad_boot\n' \ + > /usr/local/bin/scratchpad-boot.sh \ + && chmod 0755 /usr/local/bin/scratchpad-boot.sh + +# Non-root, matching the pod securityContext (drop ALL caps, no service-account token, fs_group 1000, gVisor). +RUN useradd -u 1000 -m -s /bin/sh scratchpad +USER 1000 + +# The controller always execs an explicit command; this default keeps the image runnable standalone. +CMD ["python", "-m", "anton.cloud_turn"] From a86cdfe86ccd9c9bf338a78e8857984375fe2180 Mon Sep 17 00:00:00 2001 From: Hamish Fagg Date: Tue, 28 Jul 2026 22:04:05 +1200 Subject: [PATCH 5/7] fix CI --- .github/workflows/scratchpad-dev-build.yml | 35 +------------ .github/workflows/scratchpad-prod-build.yml | 52 ------------------- .../workflows/scratchpad-staging-build.yml | 51 ------------------ 3 files changed, 2 insertions(+), 136 deletions(-) delete mode 100644 .github/workflows/scratchpad-prod-build.yml delete mode 100644 .github/workflows/scratchpad-staging-build.yml diff --git a/.github/workflows/scratchpad-dev-build.yml b/.github/workflows/scratchpad-dev-build.yml index 0a80d8b8..002093e9 100644 --- a/.github/workflows/scratchpad-dev-build.yml +++ b/.github/workflows/scratchpad-dev-build.yml @@ -19,23 +19,8 @@ concurrency: jobs: - get-deploy-labels: - runs-on: mdb-dev - outputs: - deploy-envs: ${{ steps.get-labels.outputs.deploy-envs }} - steps: - - name: Pull MindsDB Github Actions - uses: actions/checkout@v4 - with: - repository: mindsdb/github-actions - path: github-actions - - id: get-labels - uses: ./github-actions/get-deploy-labels - build: runs-on: mdb-dev - needs: [get-deploy-labels] - if: needs.get-deploy-labels.outputs.deploy-envs != '[]' outputs: image: ${{ steps.build.outputs.image }} env: @@ -43,26 +28,10 @@ jobs: steps: - name: Checkout uses: actions/checkout@v4 - - name: Pull MindsDB Github Actions - uses: actions/checkout@v4 - with: - repository: mindsdb/github-actions - path: github-actions - - uses: ./github-actions/setup-env + - uses: mindsdb/github-actions/setup-env@main - id: build - uses: ./github-actions/build-push-ecr + uses: mindsdb/github-actions/build-push-ecr@main with: # Override the repo-slug default so the image lands at .../minds-anton-scratchpad. module-name: minds-anton-scratchpad build-for-environment: development - - scan: - runs-on: mdb-dev - needs: [build] - name: Scan minds-anton-scratchpad image - steps: - - uses: actions/checkout@v4 - - uses: mindsdb/github-actions/snyk-docker-scan@main - with: - image: ${{ needs.build.outputs.image }} - snyk-token: ${{ secrets.SNYK_TOKEN }} diff --git a/.github/workflows/scratchpad-prod-build.yml b/.github/workflows/scratchpad-prod-build.yml deleted file mode 100644 index 46b1ff81..00000000 --- a/.github/workflows/scratchpad-prod-build.yml +++ /dev/null @@ -1,52 +0,0 @@ -name: Scratchpad image - Prod build on release - -# Pushes minds-anton-scratchpad as production- when a GitHub Release is published -# (or via manual dispatch). Build + scan only; the scratchpad-controller consumes the tag. - -on: - release: - types: [published] - workflow_dispatch: - -defaults: - run: - shell: bash - -concurrency: - group: ${{ github.workflow_ref }} - cancel-in-progress: true - -jobs: - - build: - runs-on: mdb-dev - outputs: - image: ${{ steps.build.outputs.image }} - env: - AWS_REGION: us-east-1 - steps: - - name: Checkout - uses: actions/checkout@v4 - - name: Pull MindsDB Github Actions - uses: actions/checkout@v4 - with: - repository: mindsdb/github-actions - path: github-actions - - uses: ./github-actions/setup-env - - id: build - uses: ./github-actions/build-push-ecr - with: - module-name: minds-anton-scratchpad - build-for-environment: production - image-ref: ${{ env.CI_REF_SLUG }} - - scan: - runs-on: mdb-dev - needs: [build] - name: Scan minds-anton-scratchpad image - steps: - - uses: actions/checkout@v4 - - uses: mindsdb/github-actions/snyk-docker-scan@main - with: - image: ${{ needs.build.outputs.image }} - snyk-token: ${{ secrets.SNYK_TOKEN }} diff --git a/.github/workflows/scratchpad-staging-build.yml b/.github/workflows/scratchpad-staging-build.yml deleted file mode 100644 index 393f9973..00000000 --- a/.github/workflows/scratchpad-staging-build.yml +++ /dev/null @@ -1,51 +0,0 @@ -name: Scratchpad image - Staging build on push to main - -# Pushes minds-anton-scratchpad as staging- (plus the moving `staging` and `latest` tags) -# on every push to main. Build + scan only; the scratchpad-controller consumes the tag. - -on: - push: - branches: - - main - -defaults: - run: - shell: bash - -concurrency: - group: ${{ github.workflow_ref }} - cancel-in-progress: true - -jobs: - - build: - runs-on: mdb-dev - outputs: - image: ${{ steps.build.outputs.image }} - env: - AWS_REGION: us-east-1 - steps: - - name: Checkout - uses: actions/checkout@v4 - - name: Pull MindsDB Github Actions - uses: actions/checkout@v4 - with: - repository: mindsdb/github-actions - path: github-actions - - uses: ./github-actions/setup-env - - id: build - uses: ./github-actions/build-push-ecr - with: - module-name: minds-anton-scratchpad - build-for-environment: staging - - scan: - runs-on: mdb-dev - needs: [build] - name: Scan minds-anton-scratchpad image - steps: - - uses: actions/checkout@v4 - - uses: mindsdb/github-actions/snyk-docker-scan@main - with: - image: ${{ needs.build.outputs.image }} - snyk-token: ${{ secrets.SNYK_TOKEN }} From febe8cfcf65d9960e6b3d87b0473532ebe835b50 Mon Sep 17 00:00:00 2001 From: ZoranPandovski Date: Tue, 28 Jul 2026 13:43:12 +0200 Subject: [PATCH 6/7] drop skeleton cloud_turn (superseded by reconciled version) --- anton/cloud_turn/__init__.py | 0 anton/cloud_turn/__main__.py | 41 ----------------------------- anton/cloud_turn/contract.py | 34 ------------------------ tests/cloud_turn/__init__.py | 0 tests/cloud_turn/test_entrypoint.py | 36 ------------------------- 5 files changed, 111 deletions(-) delete mode 100644 anton/cloud_turn/__init__.py delete mode 100644 anton/cloud_turn/__main__.py delete mode 100644 anton/cloud_turn/contract.py delete mode 100644 tests/cloud_turn/__init__.py delete mode 100644 tests/cloud_turn/test_entrypoint.py diff --git a/anton/cloud_turn/__init__.py b/anton/cloud_turn/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/anton/cloud_turn/__main__.py b/anton/cloud_turn/__main__.py deleted file mode 100644 index 54554de4..00000000 --- a/anton/cloud_turn/__main__.py +++ /dev/null @@ -1,41 +0,0 @@ -from __future__ import annotations -import asyncio -import json -import sys -from typing import AsyncIterator - -from anton.core.runtime import build_chat_session -from anton.core.llm.provider import StreamTextDelta -from anton.cloud_turn.contract import TurnRequestV1 - - -async def stream_turn(req: TurnRequestV1) -> AsyncIterator[dict]: - try: - session = await build_chat_session( - session_id=req.conversation_id, - workspace_path=req.workspace_path, - model=req.model, - ) - async for event in session.turn_stream(req.input): - if isinstance(event, StreamTextDelta): - yield {"kind": "delta", "text": event.text} - yield {"kind": "turn_completed"} - except Exception as exc: - yield {"kind": "turn_failed", "error": repr(exc)} - - -async def _run() -> None: - line = sys.stdin.readline() - req = TurnRequestV1.from_json(line) - async for ev in stream_turn(req): - sys.stdout.write(json.dumps(ev) + "\n") - sys.stdout.flush() - - -def main() -> int: - asyncio.run(_run()) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/anton/cloud_turn/contract.py b/anton/cloud_turn/contract.py deleted file mode 100644 index 3ef3431f..00000000 --- a/anton/cloud_turn/contract.py +++ /dev/null @@ -1,34 +0,0 @@ -from __future__ import annotations -import json -from dataclasses import dataclass, asdict - - -@dataclass -class TurnRequestV1: - protocol_version: int - conversation_id: str - input: str - workspace_path: str | None = None - model: str | None = None - - @staticmethod - def from_json(raw: str) -> "TurnRequestV1": - d = json.loads(raw) - return TurnRequestV1( - protocol_version=int(d["protocol_version"]), - conversation_id=str(d["conversation_id"]), - input=str(d["input"]), - workspace_path=d.get("workspace_path"), - model=d.get("model"), - ) - - -@dataclass -class TurnResultV1: - protocol_version: int - kind: str # "turn_completed" | "turn_failed" - final_text: str | None = None - error: str | None = None - - def to_json(self) -> str: - return json.dumps(asdict(self)) diff --git a/tests/cloud_turn/__init__.py b/tests/cloud_turn/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/tests/cloud_turn/test_entrypoint.py b/tests/cloud_turn/test_entrypoint.py deleted file mode 100644 index a7183ce5..00000000 --- a/tests/cloud_turn/test_entrypoint.py +++ /dev/null @@ -1,36 +0,0 @@ -import pytest -from anton.core.llm.provider import StreamTextDelta -from anton.cloud_turn.__main__ import stream_turn -from anton.cloud_turn.contract import TurnRequestV1 - - -@pytest.mark.asyncio -async def test_stream_turn_emits_deltas_then_completed(monkeypatch): - class FakeSession: - async def turn_stream(self, user_input): - yield StreamTextDelta(text="he") - yield StreamTextDelta(text="llo") - - async def fake_build(**kwargs): - assert kwargs["session_id"] == "conv-1" - return FakeSession() - - monkeypatch.setattr("anton.cloud_turn.__main__.build_chat_session", fake_build) - req = TurnRequestV1(protocol_version=1, conversation_id="conv-1", input="hi") - events = [ev async for ev in stream_turn(req)] - assert events == [ - {"kind": "delta", "text": "he"}, - {"kind": "delta", "text": "llo"}, - {"kind": "turn_completed"}, - ] - - -@pytest.mark.asyncio -async def test_stream_turn_failure_is_terminal(monkeypatch): - async def fake_build(**kwargs): - raise RuntimeError("boom") - monkeypatch.setattr("anton.cloud_turn.__main__.build_chat_session", fake_build) - req = TurnRequestV1(protocol_version=1, conversation_id="c", input="hi") - events = [ev async for ev in stream_turn(req)] - assert events[-1]["kind"] == "turn_failed" - assert "boom" in events[-1]["error"] From f37661c390073636c6158f3ddda1cbe8000141eb Mon Sep 17 00:00:00 2001 From: Zoran Pandovski Date: Tue, 28 Jul 2026 14:24:42 +0200 Subject: [PATCH 7/7] Feat/cloud turn (#282) * Initial entrypoint * Turn errors and fixtures * Turn protocol * Add turn runner and session builder * Plug the turn into backends and session * Cloud turn tests * Cloud turn tests * Fix tests * Rmv all old contract logic and tests * NEw contract * Update session and new tests --- anton/cloud_turn/__init__.py | 13 ++ anton/cloud_turn/__main__.py | 127 ++++++++++++ anton/cloud_turn/contract.py | 45 +++++ anton/cloud_turn/session.py | 130 ++++++++++++ anton/core/artifacts/store.py | 14 +- anton/core/backends/local.py | 172 +++++++++++----- anton/core/session.py | 19 ++ anton/workspace.py | 7 +- pyproject.toml | 1 + tests/cloud_turn_fake_entry.py | 112 +++++++++++ tests/test_cloud_turn_entrypoint.py | 92 +++++++++ tests/test_cloud_turn_process.py | 170 ++++++++++++++++ tests/test_cloud_turn_session.py | 296 ++++++++++++++++++++++++++++ 13 files changed, 1146 insertions(+), 52 deletions(-) create mode 100644 anton/cloud_turn/__init__.py create mode 100644 anton/cloud_turn/__main__.py create mode 100644 anton/cloud_turn/contract.py create mode 100644 anton/cloud_turn/session.py create mode 100644 tests/cloud_turn_fake_entry.py create mode 100644 tests/test_cloud_turn_entrypoint.py create mode 100644 tests/test_cloud_turn_process.py create mode 100644 tests/test_cloud_turn_session.py diff --git a/anton/cloud_turn/__init__.py b/anton/cloud_turn/__init__.py new file mode 100644 index 00000000..653c54ec --- /dev/null +++ b/anton/cloud_turn/__init__.py @@ -0,0 +1,13 @@ +"""Cloud turn: run one full anton turn inside a sandbox pod. + +`python -m anton.cloud_turn` reads a :class:`TurnRequestV1` as one JSON line on +stdin, runs the turn to completion against the mounted workspace with a +cloud-safe session, and emits `delta` / `turn_completed` / `turn_failed` JSONL +events on stdout (diagnostics to stderr). It is the headless counterpart of the +desktop CLI host: the same ChatSession, built cloud-safe. +""" + +from anton.cloud_turn.contract import TurnRequestV1 +from anton.cloud_turn.session import build_cloud_chat_session + +__all__ = ["TurnRequestV1", "build_cloud_chat_session"] diff --git a/anton/cloud_turn/__main__.py b/anton/cloud_turn/__main__.py new file mode 100644 index 00000000..fc0e631c --- /dev/null +++ b/anton/cloud_turn/__main__.py @@ -0,0 +1,127 @@ +"""`python -m anton.cloud_turn` - the sandbox-pod turn entrypoint. + +Contract (matches scratchpad-controller + cowork-server): + stdin : ONE newline-terminated TurnRequestV1 JSON line (controller closes stdin) + stdout: JSONL events - `delta` / `turn_completed` / `turn_failed`, nothing else + stderr: diagnostic logs + full tracebacks + exit : 0 (the controller detects the terminal from the event, not the code) + +stdout is isolated at the OS file-descriptor level: the real FD 1 is duplicated +to a private, non-inheritable descriptor used only for protocol events, and +process FD 1 is redirected to stderr. So `print()`, `os.write(1, ...)`, +native-library writes, and child-process stdout can never corrupt the stream. +""" + +from __future__ import annotations + +import asyncio +import contextlib +import inspect +import json +import logging +import os +import sys + +from anton.cloud_turn.contract import TurnRequestV1 +from anton.cloud_turn.session import build_cloud_chat_session + +logger = logging.getLogger(__name__) + +#: Bound the single request line so a malformed/huge stdin can't exhaust memory. +MAX_REQUEST_BYTES = 10 * 1024 * 1024 +#: Keep wire error strings short and log-safe. +MAX_ERROR_MESSAGE_CHARS = 300 + + +def _scrub(exc: Exception) -> str: + """Short, credential-scrubbed error string for the wire. Full traceback + stays on stderr (logged by the caller).""" + from anton.utils.datasources import scrub_credentials + + text = scrub_credentials(f"{type(exc).__name__}: {exc}") + if len(text) > MAX_ERROR_MESSAGE_CHARS: + text = text[: MAX_ERROR_MESSAGE_CHARS - 1] + "…" + return text + + +@contextlib.contextmanager +def _isolated_protocol_stdout(): + """OS-level stdout isolation. Yields ``emit(event: dict)`` writing JSONL to + the saved protocol descriptor; everything else (FD 1) goes to stderr.""" + sys.stdout.flush() + sys.stderr.flush() + stderr_fd = sys.stderr.fileno() + + protocol_fd = os.dup(1) + os.set_inheritable(protocol_fd, False) # children never inherit the protocol channel + os.dup2(stderr_fd, 1) # any write to fd 1 now lands on stderr + saved_sys_stdout = sys.stdout + sys.stdout = sys.stderr + logging.basicConfig(stream=sys.stderr, level=logging.INFO) + + def emit(event: dict) -> None: + data = (json.dumps(event) + "\n").encode("utf-8") + view = memoryview(data) + while view: # os.write may partial-write; loop until fully flushed + n = os.write(protocol_fd, view) + view = view[n:] + + try: + yield emit + finally: + with contextlib.suppress(Exception): + sys.stderr.flush() + sys.stdout = saved_sys_stdout + os.close(protocol_fd) + + +async def _close(session) -> None: + close = getattr(session, "close", None) + if close is None: + return + try: + result = close() + if inspect.isawaitable(result): + await result + except Exception: + logger.warning("cloud session close failed (non-fatal)", exc_info=True) + + +async def stream_turn(raw_line: str, emit, session_builder=None) -> None: + """Parse the request, run one turn, and emit exactly one terminal event. + + Streaming: assistant text is emitted as ``delta`` events as it arrives, then + a bare ``turn_completed``. Any failure (parse or turn) -> one ``turn_failed`` + with a scrubbed error string. + """ + from anton.core.llm.provider import StreamTextDelta + + builder = session_builder or build_cloud_chat_session + session = None + try: + req = TurnRequestV1.from_json(raw_line) + session = builder(req) + async for event in session.turn_stream(req.input): + if isinstance(event, StreamTextDelta): + emit({"kind": "delta", "text": event.text or ""}) + emit({"kind": "turn_completed"}) + except Exception as exc: + # Full traceback -> stderr only; wire carries a short scrubbed string. + logger.exception("cloud turn failed") + emit({"kind": "turn_failed", "error": _scrub(exc)}) + finally: + if session is not None: + await _close(session) + + +def main(argv: list[str] | None = None) -> int: + with _isolated_protocol_stdout() as emit: + # One bounded line (the controller writes a single JSON line + \n, then + # closes stdin). ``readline`` returns on the newline without blocking. + raw_line = sys.stdin.readline(MAX_REQUEST_BYTES + 1) + asyncio.run(stream_turn(raw_line, emit)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/anton/cloud_turn/contract.py b/anton/cloud_turn/contract.py new file mode 100644 index 00000000..833d05c6 --- /dev/null +++ b/anton/cloud_turn/contract.py @@ -0,0 +1,45 @@ +"""Wire contract between the scratchpad-controller and the pod entrypoint. + +Matches what the controller sends (`scratchpad_controller.anton_turn.request_line`) +and what cowork-server consumes off the reply stream. Intentionally minimal and +data-only: the entrypoint reads ONE newline-terminated JSON line on stdin. + +Events written back on stdout (JSONL) are the three the controller translates: + {"kind": "delta", "text": "..."} - streamed assistant text, one per chunk + {"kind": "turn_completed"} - terminal success (no payload) + {"kind": "turn_failed", "error": "..."} - terminal failure (scrubbed string) +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field + + +@dataclass +class TurnRequestV1: + """One turn to run in the pod. Sent as a single JSON line on stdin.""" + + protocol_version: int + conversation_id: str + input: str + #: Mount path the controller passes; the pod uses its own trusted mount and + #: does not act on this value (kept for wire-compatibility). See session.py. + workspace_path: str | None = None + #: Optional model override; None uses the settings default. + model: str | None = None + #: DB-authoritative ordered history ({"role","content"} dicts). The pod never + #: loads its own history; cowork-server owns persistence. + history: list = field(default_factory=list) + + @staticmethod + def from_json(raw: str) -> "TurnRequestV1": + d = json.loads(raw) + return TurnRequestV1( + protocol_version=int(d["protocol_version"]), + conversation_id=str(d["conversation_id"]), + input=str(d["input"]), + workspace_path=d.get("workspace_path"), + model=d.get("model"), + history=d.get("history") or [], + ) diff --git a/anton/cloud_turn/session.py b/anton/cloud_turn/session.py new file mode 100644 index 00000000..ae7b24eb --- /dev/null +++ b/anton/cloud_turn/session.py @@ -0,0 +1,130 @@ +"""Cloud-safe ChatSession builder. + +Assembles a :class:`~anton.core.session.ChatSession` scoped to the tenant's +mounted workspace with the desktop-only, tenant-leaky behaviours OFF. It does +NOT call the desktop ``build_chat_session`` (which loads workspace ``.env``, +uses ``~/.anton`` personal memory, and injects vault creds into ``os.environ``). + +Safety posture (all internal — nothing here is on the wire): + +* Trusted pod-side workspace mount, never taken from the request. +* No dotenv loading (``AntonSettings(_env_file=None)``, shared into Workspace). +* Personal memory / connectors / data-vault / disk history OFF. +* Scratchpad subprocess env built from a non-secret allowlist, so generated + code can't read the provider key (interim until the Plan-5 gateway removes + the key from the pod entirely). +* Only reviewed, headless-safe tools are exposed (scratchpad + artifacts). +""" + +from __future__ import annotations + +import logging +import os +from pathlib import Path +from typing import TYPE_CHECKING + +from anton.cloud_turn.contract import TurnRequestV1 + +if TYPE_CHECKING: + from anton.core.session import ChatSession + +logger = logging.getLogger(__name__) + +#: Trusted mount path — pod-side config, never from the wire request. +DEFAULT_CLOUD_WORKSPACE_PATH = "/workspace" +#: Operator/CI override for the mount path (pod-side env var, not request data). +_WORKSPACE_PATH_ENV = "ANTON_CLOUD_WORKSPACE_PATH" + +#: The only tools exposed in a cloud turn: scratchpad + the workspace-scoped +#: artifact tools. Everything else core registers is dropped. +CLOUD_TOOL_ALLOWLIST = frozenset( + { + "scratchpad", + "create_artifact", + "list_artifacts", + "open_artifact", + "update_artifact", + } +) + + +def resolve_trusted_workspace_path() -> Path: + """Resolve the trusted workspace mount path (never from the wire request). + + Reads :data:`_WORKSPACE_PATH_ENV` or falls back to + :data:`DEFAULT_CLOUD_WORKSPACE_PATH`; rejects relative paths and ``..``, + then canonicalises so downstream containment checks compare a real path. + """ + raw = os.environ.get(_WORKSPACE_PATH_ENV) or DEFAULT_CLOUD_WORKSPACE_PATH + if not os.path.isabs(raw): + raise ValueError( + f"trusted workspace path must be absolute, got {raw!r} " + f"(set {_WORKSPACE_PATH_ENV} to an absolute path)" + ) + if ".." in Path(raw).parts: + raise ValueError(f"trusted workspace path must not contain '..': {raw!r}") + resolved = Path(raw).resolve() + resolved.mkdir(parents=True, exist_ok=True) + if not resolved.is_dir(): + raise ValueError(f"trusted workspace path is not a directory: {resolved}") + return resolved + + +def build_cloud_chat_session(request: TurnRequestV1) -> "ChatSession": + """Assemble a cloud-safe ChatSession for one turn. + + History is DB-authoritative (from the request). The workspace path is the + trusted pod mount, NOT ``request.workspace_path``. + """ + from anton.config.settings import AntonSettings + from anton.core.backends.local import sanitized_scratchpad_runtime_factory + from anton.core.llm.client import LLMClient + from anton.core.session import ChatSession, ChatSessionConfig + from anton.workspace import Workspace + + base = resolve_trusted_workspace_path() + + # `_env_file=None`: never load the AntonSettings .env chain (~/.anton/.env, + # ~/.cowork/.env, /workspace/.env). Same object passed to Workspace so it + # doesn't build a second, dotenv-loading one. + settings = AntonSettings(_env_file=None) + settings.resolve_workspace(str(base)) + if request.model: + settings.planning_model = request.model + # Skills stay in the workspace, never the pod-shared ~/.anton. + settings.skills_root = base / ".anton" / "skills" + + workspace = Workspace(base, settings=settings) + workspace.initialize() + # No apply_env_to_process(): loading workspace .env into the process env + # would expose tenant secrets to cell code. + + llm_client = LLMClient.from_settings(settings) + + config = ChatSessionConfig( + llm_client=llm_client, + settings=settings, + workspace=workspace, + session_id=request.conversation_id, + harness="cloud", + # DB-authoritative history; the pod never loads its own. + initial_history=list(request.history) if request.history else None, + console=None, # headless + cortex=None, # personal memory OFF + episodic=None, + self_awareness=None, + data_vault=None, # connectors OFF + history_store=None, # disk history OFF (DB authoritative) + tools=[], # no host connector/publish tools + tool_allowlist=CLOUD_TOOL_ALLOWLIST, # only reviewed tools survive the build + runtime_factory=sanitized_scratchpad_runtime_factory, # secret-free scratchpad env + web_search_enabled=False, + web_fetch_enabled=False, + ) + + session = ChatSession(config) + logger.info( + "cloud session built conversation=%s workspace=%s tools=%s", + request.conversation_id, base, sorted(CLOUD_TOOL_ALLOWLIST), + ) + return session diff --git a/anton/core/artifacts/store.py b/anton/core/artifacts/store.py index 19cb4672..135d60b1 100644 --- a/anton/core/artifacts/store.py +++ b/anton/core/artifacts/store.py @@ -117,6 +117,13 @@ def ensure_root(self) -> Path: return self._root def folder_for(self, slug: str) -> Path: + # `open`/`update` take the slug straight from tool input. Reject anything + # that resolves outside the root (``..``, absolute, or symlink escape); + # nested-but-contained paths are fine (create never emits them). + candidate = (self._root / slug).resolve() + root = self._root.resolve() + if candidate != root and root not in candidate.parents: + raise ValueError(f"artifact slug escapes the workspace: {slug!r}") return self._root / slug def metadata_path(self, slug: str) -> Path: @@ -385,7 +392,12 @@ def _save(self, artifact: Artifact) -> None: self.readme_path(artifact.slug).write_text(readme, encoding="utf-8") def _load_silent(self, slug: str) -> Artifact | None: - path = self.metadata_path(slug) + try: + path = self.metadata_path(slug) + except ValueError: + # Escaping slug → treat as "no such artifact" so open/update stay graceful. + logger.warning("Rejected out-of-workspace artifact slug %r", slug) + return None if not path.is_file(): return None try: diff --git a/anton/core/backends/local.py b/anton/core/backends/local.py index d73ee85c..3bfbffc6 100644 --- a/anton/core/backends/local.py +++ b/anton/core/backends/local.py @@ -77,6 +77,47 @@ def _utf8_env(base: "os._Environ[str] | dict[str, str]") -> dict[str, str]: return env +# Sanitized scratchpad env contract: in sanitize_env mode the child inherits +# ONLY these non-secret names (and only if the parent has them) — never a +# provider key, ANTON_* credential, gateway token, or datasource secret. +# Python runtime vars (PYTHONUTF8, PYTHONPATH) are set explicitly in start(), +# never inherited, so nothing can redirect imports. +_SCRATCHPAD_ENV_ALLOWLIST = frozenset( + { + # process / tooling + "PATH", + "HOME", + "USER", + "LOGNAME", + "SHELL", + # temp dirs + "TMPDIR", + "TMP", + "TEMP", + # locale / tty + "LANG", + "LC_ALL", + "LC_CTYPE", + "LANGUAGE", + "TZ", + "TERM", + # TLS trust roots (HTTPS verification from cell code) + "SSL_CERT_FILE", + "SSL_CERT_DIR", + "REQUESTS_CA_BUNDLE", + "CURL_CA_BUNDLE", + # Windows: required for subprocess/socket startup + "SYSTEMROOT", + "SYSTEMDRIVE", + } +) + + +def _sanitized_parent_env() -> dict[str, str]: + """Parent env reduced to the non-secret allowlist.""" + return {k: v for k, v in os.environ.items() if k in _SCRATCHPAD_ENV_ALLOWLIST} + + _MAX_OUTPUT = 10_000 @@ -95,6 +136,7 @@ def __init__( coding_base_url: str, cells: list[Cell] | None = None, workspace_path: Path | None = None, + sanitize_env: bool = False, _venvs_base: Path | None = None, ) -> None: super().__init__( @@ -114,6 +156,9 @@ def __init__( # is a "where to put scratchpad venvs" hint; the explicit # arg is "the agent's project, when known". self._explicit_workspace_path: Path | None = workspace_path + # Cloud/headless: build the child env from the non-secret allowlist. + # Default False = desktop behaviour (full inherited env) unchanged. + self._sanitize_env: bool = sanitize_env self._proc: asyncio.subprocess.Process | None = None self._boot_path: str | None = None self._venv_dir: str | None = None @@ -390,59 +435,64 @@ async def start(self) -> None: os.close(fd) self._boot_path = path - # Force UTF-8 mode in the child so its I/O never depends on the host - # code page (ENG-824). - env = _utf8_env(os.environ) + # Force UTF-8 in the child (ENG-824). In sanitized mode the base is the + # non-secret allowlist, not the full parent env. + env = _utf8_env( + _sanitized_parent_env() if self._sanitize_env else os.environ + ) if self._coding_model: env["ANTON_SCRATCHPAD_MODEL"] = self._coding_model if self._coding_provider: env["ANTON_SCRATCHPAD_PROVIDER"] = self._coding_provider - if "ANTHROPIC_API_KEY" not in env and "ANTON_ANTHROPIC_API_KEY" in env: - env["ANTHROPIC_API_KEY"] = env["ANTON_ANTHROPIC_API_KEY"] - if "OPENAI_API_KEY" not in env and "ANTON_OPENAI_API_KEY" in env: - env["OPENAI_API_KEY"] = env["ANTON_OPENAI_API_KEY"] - if "OPENAI_BASE_URL" not in env and "ANTON_OPENAI_BASE_URL" in env: - env["OPENAI_BASE_URL"] = env["ANTON_OPENAI_BASE_URL"] - if ( - "OPENAI_API_KEY" not in env - and "ANTON_MINDS_API_KEY" in env - and self._coding_provider == "openai-compatible" - ): - env["OPENAI_API_KEY"] = env["ANTON_MINDS_API_KEY"] - if ( - "OPENAI_BASE_URL" not in env - and "ANTON_MINDS_URL" in env - and self._coding_provider == "openai-compatible" - ): - # Host-aware (ENG-436): api.mindshub.ai serves /v1, legacy - # mdb.ai serves /api/v1. The previous hardcoded /api/v1 was - # wrong for mindshub. Mirrors config/settings.py + - # cowork-server minds_chat_base_url. - _minds_base = env["ANTON_MINDS_URL"].rstrip("/") - if _minds_base.endswith("/v1"): - env["OPENAI_BASE_URL"] = _minds_base - elif "mdb.ai" in _minds_base: - env["OPENAI_BASE_URL"] = f"{_minds_base}/api/v1" - else: - env["OPENAI_BASE_URL"] = f"{_minds_base}/v1" - if self._coding_api_key: - sdk_key = { - "anthropic": "ANTHROPIC_API_KEY", - "openai": "OPENAI_API_KEY", - "openai-compatible": "OPENAI_API_KEY", - }.get(self._coding_provider, "") - if sdk_key: - env[sdk_key] = self._coding_api_key - if self._coding_provider in ("openai", "openai-compatible"): - base_url = ( - self._coding_base_url - or env.get("ANTON_OPENAI_BASE_URL") - or env.get("OPENAI_BASE_URL") - or "" - ) - if base_url: - env["OPENAI_BASE_URL"] = base_url - env["ANTON_OPENAI_BASE_URL"] = base_url + # Provider-credential propagation is desktop-only; sanitized mode + # withholds every model key (nested get_llm() is off in the cloud pod). + if not self._sanitize_env: + if "ANTHROPIC_API_KEY" not in env and "ANTON_ANTHROPIC_API_KEY" in env: + env["ANTHROPIC_API_KEY"] = env["ANTON_ANTHROPIC_API_KEY"] + if "OPENAI_API_KEY" not in env and "ANTON_OPENAI_API_KEY" in env: + env["OPENAI_API_KEY"] = env["ANTON_OPENAI_API_KEY"] + if "OPENAI_BASE_URL" not in env and "ANTON_OPENAI_BASE_URL" in env: + env["OPENAI_BASE_URL"] = env["ANTON_OPENAI_BASE_URL"] + if ( + "OPENAI_API_KEY" not in env + and "ANTON_MINDS_API_KEY" in env + and self._coding_provider == "openai-compatible" + ): + env["OPENAI_API_KEY"] = env["ANTON_MINDS_API_KEY"] + if ( + "OPENAI_BASE_URL" not in env + and "ANTON_MINDS_URL" in env + and self._coding_provider == "openai-compatible" + ): + # Host-aware (ENG-436): api.mindshub.ai serves /v1, legacy + # mdb.ai serves /api/v1. The previous hardcoded /api/v1 was + # wrong for mindshub. Mirrors config/settings.py + + # cowork-server minds_chat_base_url. + _minds_base = env["ANTON_MINDS_URL"].rstrip("/") + if _minds_base.endswith("/v1"): + env["OPENAI_BASE_URL"] = _minds_base + elif "mdb.ai" in _minds_base: + env["OPENAI_BASE_URL"] = f"{_minds_base}/api/v1" + else: + env["OPENAI_BASE_URL"] = f"{_minds_base}/v1" + if self._coding_api_key: + sdk_key = { + "anthropic": "ANTHROPIC_API_KEY", + "openai": "OPENAI_API_KEY", + "openai-compatible": "OPENAI_API_KEY", + }.get(self._coding_provider, "") + if sdk_key: + env[sdk_key] = self._coding_api_key + if self._coding_provider in ("openai", "openai-compatible"): + base_url = ( + self._coding_base_url + or env.get("ANTON_OPENAI_BASE_URL") + or env.get("OPENAI_BASE_URL") + or "" + ) + if base_url: + env["OPENAI_BASE_URL"] = base_url + env["ANTON_OPENAI_BASE_URL"] = base_url uv = self._find_uv() if uv: env["ANTON_UV_PATH"] = uv @@ -817,3 +867,27 @@ def local_scratchpad_runtime_factory( cells=cells, workspace_path=workspace_path, ) + + +def sanitized_scratchpad_runtime_factory( + *, + name: str, + coding_provider: str, + coding_model: str, + coding_api_key: str, + coding_base_url: str, + cells: list[Cell] | None, + workspace_path: Path | None, +) -> ScratchpadRuntime: + """Cloud/headless factory: like the local factory but with + ``sanitize_env=True`` so the child env carries no secrets.""" + return LocalScratchpadRuntime( + name=name, + coding_provider=coding_provider, + coding_model=coding_model, + coding_api_key=coding_api_key, + coding_base_url=coding_base_url, + cells=cells, + workspace_path=workspace_path, + sanitize_env=True, + ) diff --git a/anton/core/session.py b/anton/core/session.py index c6920bba..6a5c27c0 100644 --- a/anton/core/session.py +++ b/anton/core/session.py @@ -375,6 +375,10 @@ class ChatSessionConfig: # settings' `router_enabled` (ANTON_ROUTER_ENABLED); hosts pass an # explicit bool to override per session. router_enabled: bool | None = None + # When set, only these tool names survive the build; ``None`` = full desktop + # set. Applied on every ``_build_tools`` call so a lazy rebuild can't leak a + # non-allowlisted tool. + tool_allowlist: frozenset[str] | None = None class ChatSession: @@ -423,6 +427,7 @@ def __init__(self, config: ChatSessionConfig) -> None: self._act_first = config.act_first self._started_at = config.started_at self._extra_tools = config.tools + self._tool_allowlist = config.tool_allowlist self._workspace = config.workspace self._data_vault = config.data_vault self._console = config.console @@ -955,6 +960,20 @@ def _build_tools(self) -> list[dict]: self._build_core_tools() for tool in self._extra_tools: self.tool_registry.register_tool(tool) + # Enforce the allowlist on every build (None = full desktop set). + if self._tool_allowlist is not None: + built = {t.name for t in self.tool_registry.get_tool_defs()} + # Fail loud on a name that matches no built tool (typo / unavailable + # here) rather than silently dropping it. + unknown = set(self._tool_allowlist) - built + if unknown: + raise ValueError( + "tool_allowlist names not registered in this session: " + + ", ".join(sorted(unknown)) + + f" (available: {', '.join(sorted(built))})" + ) + for name in built - set(self._tool_allowlist): + self.tool_registry.unregister_tool(name) return self.tool_registry.dump() def _build_core_tools(self) -> None: diff --git a/anton/workspace.py b/anton/workspace.py index 1626f751..d3258f21 100644 --- a/anton/workspace.py +++ b/anton/workspace.py @@ -27,14 +27,17 @@ class Workspace: """Manages the .anton/ workspace directory and its files.""" - def __init__(self, base: Path) -> None: + def __init__(self, base: Path, settings: AntonSettings | None = None) -> None: self._base = base self._anton_dir = base / ".anton" self._anton_md = self._anton_dir / "anton.md" self._env_file = self._anton_dir / ".env" self._anton_md_last_read: datetime | None = None - settings = AntonSettings() + # Reuse a caller's settings so the cloud pod's dotenv-disabled + # AntonSettings isn't bypassed by a second one built here. Only + # `artifacts_dir` is read; None = desktop behaviour unchanged. + settings = settings or AntonSettings() self._artifacts_dir = self._anton_dir / settings.artifacts_dir @property diff --git a/pyproject.toml b/pyproject.toml index 4901fe87..a394d5b8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,7 @@ asyncio_mode = "auto" pythonpath = ["."] markers = [ "stub_only: test requires stub server (skipped when --live is passed)", + "slow: test spawns a real scratchpad venv/subprocess (seconds)", ] [tool.hatch.build.targets.wheel] diff --git a/tests/cloud_turn_fake_entry.py b/tests/cloud_turn_fake_entry.py new file mode 100644 index 00000000..bf7b6ffa --- /dev/null +++ b/tests/cloud_turn_fake_entry.py @@ -0,0 +1,112 @@ +"""Deterministic subprocess entrypoint for cloud-turn E2E / FD tests. + +Run as ``python tests/cloud_turn_fake_entry.py`` — it installs a deterministic +seam, then calls the REAL ``anton.cloud_turn.__main__.main()`` so the full +process boundary (FD isolation, stdin parse, runner lifecycle, scratchpad) runs +for real. NOT collected by pytest (no ``test_`` prefix). + +Modes (env ``CLOUD_TURN_FAKE_MODE``): +* ``model`` — real ChatSession + real scratchpad, but the LLM is a fake provider + scripted by ``CLOUD_TURN_FAKE_SCRIPT`` (JSON list of steps). No network. +* ``stray`` — replace the session builder with one whose turn writes stray + output to stdout/FD 1/logging, then completes (proves FD isolation). +* ``stray_fail`` — like ``stray`` but the turn raises after the stray output. +""" + +from __future__ import annotations + +import json +import logging +import os +import sys + + +def _install_fake_model() -> None: + import anton.core.llm.client as client_mod + from anton.core.llm.client import LLMClient + from anton.core.llm.provider import ( + LLMProvider, + LLMResponse, + ProviderConnectionInfo, + ToolCall, + Usage, + ) + + script = json.loads(os.environ.get("CLOUD_TURN_FAKE_SCRIPT", "[]")) + + class _FakeProvider(LLMProvider): + name = "fake" + + def __init__(self) -> None: + self._i = 0 + + async def complete(self, *, model, system, messages, tools=None, + tool_choice=None, max_tokens=4096, native_web_tools=None): + step = script[min(self._i, len(script) - 1)] if script else {"text": ""} + self._i += 1 + tool_calls = [] + if "tool" in step: + t = step["tool"] + tool_calls = [ToolCall(id=t.get("id", "t1"), name=t["name"], input=t["input"])] + return LLMResponse( + content=step.get("text", ""), + tool_calls=tool_calls, + usage=Usage(context_pressure=0.0), + ) + + def export_connection_info(self): + return ProviderConnectionInfo(provider="fake", api_key="fake") + + def _fake_from_settings(cls, settings): + prov = _FakeProvider() + return LLMClient( + planning_provider=prov, planning_model="fake-model", + coding_provider=prov, coding_model="fake-model", + ) + + client_mod.LLMClient.from_settings = classmethod(_fake_from_settings) + + +def _install_stray_session(fail: bool) -> None: + import anton.cloud_turn.__main__ as entry_mod + + class _StraySession: + def __init__(self) -> None: + self.history = [] + self.closed = False + + async def turn_stream(self, user_input, **kwargs): + # Every channel that must NOT reach the protocol stream: + print("STRAY via print()") # Python stdout + sys.stdout.write("STRAY via sys.stdout.write\n") # Python stdout + os.write(1, b"STRAY via os.write(1)\n") # direct FD 1 (native-style) + logging.getLogger("some.library").warning("STRAY via logging") + if fail: + raise RuntimeError("boom after stray output") + if False: # make this an async generator + yield + + def close(self): + self.closed = True + + entry_mod.build_cloud_chat_session = lambda request: _StraySession() + + +def main() -> int: + mode = os.environ.get("CLOUD_TURN_FAKE_MODE", "model") + if mode == "model": + _install_fake_model() + elif mode == "stray": + _install_stray_session(fail=False) + elif mode == "stray_fail": + _install_stray_session(fail=True) + else: + raise SystemExit(f"unknown CLOUD_TURN_FAKE_MODE={mode!r}") + + from anton.cloud_turn.__main__ import main as real_main + + return real_main([]) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/test_cloud_turn_entrypoint.py b/tests/test_cloud_turn_entrypoint.py new file mode 100644 index 00000000..b3f30170 --- /dev/null +++ b/tests/test_cloud_turn_entrypoint.py @@ -0,0 +1,92 @@ +"""Entrypoint wire contract: request parsing + streaming event emission. + +Offline: a fake session (scripted stream events) replaces ChatSession, so no LLM +key or scratchpad is needed. Mirrors the controller/cowork contract: +`delta` -> ... -> `turn_completed` | `turn_failed`. +""" + +from __future__ import annotations + +import asyncio +import json + +from anton.cloud_turn.contract import TurnRequestV1 +from anton.cloud_turn.__main__ import stream_turn +from anton.core.llm.provider import StreamTextDelta + + +# ── contract parsing ───────────────────────────────────────────────────────── + +def test_from_json_parses_full_request(): + req = TurnRequestV1.from_json(json.dumps({ + "protocol_version": 1, "conversation_id": "c", "input": "hi", + "workspace_path": "/workspace", "model": "m", + "history": [{"role": "user", "content": "prev"}], + })) + assert req.conversation_id == "c" + assert req.input == "hi" + assert req.history == [{"role": "user", "content": "prev"}] + + +def test_from_json_history_defaults_empty(): + req = TurnRequestV1.from_json('{"protocol_version":1,"conversation_id":"c","input":"hi"}') + assert req.history == [] + assert req.workspace_path is None and req.model is None + + +# ── streaming event emission ───────────────────────────────────────────────── + +class _FakeSession: + def __init__(self, deltas=(), raise_on_stream=None): + self._deltas = list(deltas) + self._raise = raise_on_stream + self.closed = False + + async def turn_stream(self, user_input, **kwargs): + if self._raise: + raise self._raise + for d in self._deltas: + yield StreamTextDelta(text=d) + + def close(self): + self.closed = True + + +def _drive(session, req_json='{"protocol_version":1,"conversation_id":"c","input":"hi"}'): + events = [] + asyncio.run(stream_turn(req_json, events.append, session_builder=lambda r: session)) + return events + + +def test_emits_deltas_then_completed(): + session = _FakeSession(deltas=["he", "llo"]) + events = _drive(session) + assert events == [ + {"kind": "delta", "text": "he"}, + {"kind": "delta", "text": "llo"}, + {"kind": "turn_completed"}, + ] + assert session.closed is True # session closed on success + + +def test_text_only_no_deltas_still_completes(): + events = _drive(_FakeSession(deltas=[])) + assert events == [{"kind": "turn_completed"}] + + +def test_turn_failure_is_terminal_and_scrubbed(): + session = _FakeSession(raise_on_stream=RuntimeError("boom sk-ant-" + "A" * 80)) + events = _drive(session) + assert len(events) == 1 + assert events[0]["kind"] == "turn_failed" + assert "boom" in events[0]["error"] + assert "sk-ant-" + "A" * 80 not in events[0]["error"] # credential scrubbed + assert session.closed is True # closed even on failure + + +def test_bad_request_is_a_single_turn_failed(): + events = [] + asyncio.run(stream_turn("{not valid json", events.append)) + assert len(events) == 1 + assert events[0]["kind"] == "turn_failed" + assert events[0]["error"] # non-empty, scrubbed diff --git a/tests/test_cloud_turn_process.py b/tests/test_cloud_turn_process.py new file mode 100644 index 00000000..27862ad7 --- /dev/null +++ b/tests/test_cloud_turn_process.py @@ -0,0 +1,170 @@ +"""Real-subprocess tests for the cloud-turn process boundary. + +Runs the actual entrypoint (via cloud_turn_fake_entry.py, which calls the real +main()) as a child process, so FD-level stdout isolation, fresh +process/scratchpad state, and the deterministic E2E are proven end to end. +Event kinds match the controller contract: delta / turn_completed / turn_failed. +""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + +_HARNESS = str(Path(__file__).parent / "cloud_turn_fake_entry.py") + + +def _run_cli(request, *, workspace, mode="model", script=None, timeout=60): + """Run the entrypoint as a subprocess; return (exit_code, events, stdout, stderr). + Parsing stdout as JSONL is itself the assertion that stdout is clean.""" + env = os.environ.copy() + env["ANTON_CLOUD_WORKSPACE_PATH"] = str(workspace) + env["CLOUD_TURN_FAKE_MODE"] = mode + if script is not None: + env["CLOUD_TURN_FAKE_SCRIPT"] = json.dumps(script) + + stdin = request if isinstance(request, str) else json.dumps(request) + proc = subprocess.run( + [sys.executable, _HARNESS], + input=stdin.encode("utf-8"), + stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=env, timeout=timeout, + ) + lines = [ln for ln in proc.stdout.decode("utf-8").splitlines() if ln.strip()] + events = [json.loads(ln) for ln in lines] # raises if stdout isn't clean JSONL + return proc.returncode, events, proc.stdout.decode(), proc.stderr.decode() + + +def _req(**over): + body = {"protocol_version": 1, "conversation_id": "c", "input": "hi"} + body.update(over) + return body + + +# ── FD-level stdout isolation ──────────────────────────────────────────────── + +def test_stray_stdout_never_corrupts_protocol(tmp_path): + """print(), sys.stdout.write, os.write(1, …) and library logging during the + turn must all land on stderr, leaving stdout a clean protocol stream.""" + _, events, stdout, stderr = _run_cli(_req(), workspace=tmp_path, mode="stray") + assert [e["kind"] for e in events] == ["turn_completed"] + assert "STRAY" not in stdout # nothing stray on the protocol channel + assert "STRAY via os.write(1)" in stderr # direct FD-1 write redirected to stderr + assert "STRAY via print()" in stderr + + +def test_stray_then_failure_stays_clean(tmp_path): + _, events, stdout, _ = _run_cli(_req(), workspace=tmp_path, mode="stray_fail") + assert [e["kind"] for e in events] == ["turn_failed"] + assert "STRAY" not in stdout + assert "Traceback" not in stdout # no traceback leaks onto the wire + assert "boom" in events[-1]["error"] + + +def test_malformed_input_clean_protocol(tmp_path): + _, events, stdout, _ = _run_cli("{not valid json", workspace=tmp_path) + assert len(events) == 1 and events[0]["kind"] == "turn_failed" + assert events[0]["error"] + + +# ── deterministic E2E (real CLI, fake model, no network) ───────────────────── + +def test_e2e_text_only_turn(tmp_path): + """Full path: JSON on stdin -> parse -> cloud-safe session -> streaming + delta events -> turn_completed.""" + _, events, *_ = _run_cli( + _req(input="What is 2 + 2?"), + workspace=tmp_path, mode="model", script=[{"text": "The answer is 4."}], + ) + kinds = [e["kind"] for e in events] + assert kinds[-1] == "turn_completed" + text = "".join(e["text"] for e in events if e["kind"] == "delta") + assert text == "The answer is 4." + + +# ── fresh process + scratchpad state per invocation ────────────────────────── +# Tool output is not on the wire (his contract), so we observe via workspace +# files: Turn A sets a variable + writes a file; Turn B (new process, same +# workspace) reports whether the variable survived. Files persist, runtime does not. + +_CELL_A = ( + "X_SENTINEL = 4242\n" + "open('a_done.txt', 'w').write('a-ok')\n" + "print('set')\n" +) +_CELL_B = ( + "open('b_result.txt', 'w').write('has_x=' + str('X_SENTINEL' in dir()))\n" + "print('done')\n" +) + + +def _scratchpad_step(code): + return {"tool": {"name": "scratchpad", "input": { + "action": "exec", "name": "main", "code": code, + "one_line_description": "test cell", + }}} + + +@pytest.mark.slow +def test_fresh_scratchpad_state_across_processes(tmp_path): + codeA, eventsA, *_ = _run_cli( + _req(conversation_id="A", input="set state"), + workspace=tmp_path, mode="model", timeout=180, + script=[_scratchpad_step(_CELL_A), {"text": "did A"}], + ) + assert eventsA[-1]["kind"] == "turn_completed", eventsA + # Turn A's scratchpad ran and its file persists in the workspace. + assert (tmp_path / "a_done.txt").read_text() == "a-ok" + + codeB, eventsB, *_ = _run_cli( + _req(conversation_id="B", input="read state"), + workspace=tmp_path, mode="model", timeout=180, + script=[_scratchpad_step(_CELL_B), {"text": "did B"}], + ) + assert eventsB[-1]["kind"] == "turn_completed", eventsB + # New process => fresh scratchpad namespace: Turn A's variable is gone. + assert (tmp_path / "b_result.txt").read_text() == "has_x=False" + + +# ── session.close() terminates the inner scratchpad (no orphans) ───────────── + +async def _real_cloud_session(tmp_path, monkeypatch): + import anton.core.llm.client as llm_client_mod + from unittest.mock import AsyncMock, MagicMock + + from anton.core.llm.provider import ProviderConnectionInfo + from anton.cloud_turn.contract import TurnRequestV1 + from anton.cloud_turn.session import build_cloud_chat_session + + monkeypatch.setenv("ANTON_CLOUD_WORKSPACE_PATH", str(tmp_path)) + + def _mk(cls, settings): + llm = AsyncMock() + llm.coding_provider = MagicMock() + llm.coding_provider.export_connection_info = MagicMock( + return_value=ProviderConnectionInfo(provider="anthropic", api_key="test")) + llm.coding_model = "m" + llm.planning_provider = MagicMock() + llm.planning_provider.native_web_tools = MagicMock(return_value=set()) + return llm + + monkeypatch.setattr(llm_client_mod.LLMClient, "from_settings", classmethod(_mk)) + req = TurnRequestV1(protocol_version=1, conversation_id="c", input="hi") + return build_cloud_chat_session(req) + + +@pytest.mark.slow +async def test_session_close_terminates_scratchpad(tmp_path, monkeypatch): + session = await _real_cloud_session(tmp_path, monkeypatch) + pad = await session._scratchpads.get_or_create("main") + await pad.execute("x = 1") + proc = pad._proc + assert proc is not None and proc.returncode is None # alive + + await session.close() + assert pad._proc is None # manager released it + assert proc.returncode is not None # OS process terminated diff --git a/tests/test_cloud_turn_session.py b/tests/test_cloud_turn_session.py new file mode 100644 index 00000000..1954c96a --- /dev/null +++ b/tests/test_cloud_turn_session.py @@ -0,0 +1,296 @@ +"""Safety tests for the cloud-safe ChatSession builder. + +Two layers: +* Config-level (fast, offline): the ChatSessionConfig handed to ChatSession has + every cross-tenant hazard off and the right allowlist/factory. +* Enforcement-level (real code paths): drive the real lazy _build_tools() and a + real sanitized scratchpad subprocess so a claimed property is proven, not just + configured. +""" + +from __future__ import annotations + +import pytest + +import anton.core.llm.client as llm_client_mod +import anton.core.session as session_mod +from anton.cloud_turn.contract import TurnRequestV1 +from anton.cloud_turn.session import ( + CLOUD_TOOL_ALLOWLIST, + _WORKSPACE_PATH_ENV, + build_cloud_chat_session, + resolve_trusted_workspace_path, +) +from anton.core.backends.local import sanitized_scratchpad_runtime_factory + + +class _FakeSession: + """Placeholder return value for the mocked ChatSession - config-level tests + only inspect the captured ChatSessionConfig, never the session.""" + + +def _build(tmp_path, monkeypatch, **req_overrides): + captured: dict = {} + + def fake_chat_session(config): + captured["config"] = config + return _FakeSession() + + monkeypatch.setattr(session_mod, "ChatSession", fake_chat_session) + monkeypatch.setattr( + llm_client_mod.LLMClient, "from_settings", + classmethod(lambda cls, settings: object()), + ) + monkeypatch.setenv(_WORKSPACE_PATH_ENV, str(tmp_path)) + + body = dict(protocol_version=1, conversation_id="conv_1", input="hello") + body.update(req_overrides) + session = build_cloud_chat_session(TurnRequestV1(**body)) + return session, captured["config"] + + +# ── config-level safety ────────────────────────────────────────────────────── + +def test_all_cross_tenant_hazards_off(tmp_path, monkeypatch): + _, cfg = _build(tmp_path, monkeypatch) + assert cfg.cortex is None # personal memory OFF + assert cfg.episodic is None + assert cfg.self_awareness is None + assert cfg.data_vault is None # connectors / vault OFF + assert cfg.history_store is None # disk history OFF + assert cfg.console is None # headless + assert cfg.tools == [] # no host connector/publish tools + assert cfg.web_search_enabled is False + assert cfg.web_fetch_enabled is False + + +def test_scratchpad_uses_sanitized_factory_and_is_workspace_bound(tmp_path, monkeypatch): + _, cfg = _build(tmp_path, monkeypatch) + assert cfg.runtime_factory is sanitized_scratchpad_runtime_factory + assert cfg.workspace is not None + assert cfg.harness == "cloud" + assert cfg.session_id == "conv_1" + + +def test_db_history_is_seeded_not_loaded(tmp_path, monkeypatch): + _, cfg = _build( + tmp_path, monkeypatch, + history=[{"role": "user", "content": "prior turn"}], + ) + assert cfg.initial_history == [{"role": "user", "content": "prior turn"}] + + +def test_config_uses_explicit_tool_allowlist(tmp_path, monkeypatch): + _, cfg = _build(tmp_path, monkeypatch) + assert cfg.tool_allowlist == CLOUD_TOOL_ALLOWLIST + assert "launch_backend" not in cfg.tool_allowlist + assert "scratchpad" in cfg.tool_allowlist + + +def test_model_override_applied(tmp_path, monkeypatch): + _, cfg = _build(tmp_path, monkeypatch, model="claude-opus-4-8") + assert cfg.settings.planning_model == "claude-opus-4-8" + + +# ── trusted workspace path (never from the wire) ───────────────────────────── + +def test_request_workspace_path_is_ignored_trusted_mount_used(tmp_path, monkeypatch): + # A request may carry workspace_path (cowork sends "/workspace"), but the pod + # uses its OWN trusted mount, not the wire value. + monkeypatch.setenv(_WORKSPACE_PATH_ENV, str(tmp_path)) + _, cfg = _build(tmp_path, monkeypatch, workspace_path="/etc/evil") + assert cfg.workspace.base == tmp_path.resolve() + + +def test_resolver_uses_env_override(tmp_path, monkeypatch): + monkeypatch.setenv(_WORKSPACE_PATH_ENV, str(tmp_path)) + assert resolve_trusted_workspace_path() == tmp_path.resolve() + + +def test_resolver_rejects_relative_path(monkeypatch): + monkeypatch.setenv(_WORKSPACE_PATH_ENV, "not/absolute") + with pytest.raises(ValueError, match="absolute"): + resolve_trusted_workspace_path() + + +def test_resolver_rejects_parent_traversal(monkeypatch): + monkeypatch.setenv(_WORKSPACE_PATH_ENV, "/workspace/../etc") + with pytest.raises(ValueError, match=r"\.\."): + resolve_trusted_workspace_path() + + +def test_artifact_tools_cannot_escape_workspace(tmp_path): + from anton.core.artifacts.store import ArtifactStore + + store = ArtifactStore(tmp_path / "artifacts") + for bad in ("../../etc", "/etc/passwd", "a/../../b"): + with pytest.raises(ValueError, match="escapes"): + store.folder_for(bad) + assert store.open("../../../etc/passwd") is None + assert store.folder_for("my-report").parent == (tmp_path / "artifacts") + + +# ── dotenv never loaded ────────────────────────────────────────────────────── + +def test_apply_env_to_process_never_called(tmp_path, monkeypatch): + import anton.workspace as ws_mod + calls = {"apply_env": 0, "init": 0} + real_init = ws_mod.Workspace.initialize + real_apply = ws_mod.Workspace.apply_env_to_process + + monkeypatch.setattr(ws_mod.Workspace, "initialize", + lambda self: (calls.__setitem__("init", calls["init"] + 1), real_init(self))[1]) + monkeypatch.setattr(ws_mod.Workspace, "apply_env_to_process", + lambda self: (calls.__setitem__("apply_env", calls["apply_env"] + 1), real_apply(self))[1]) + + _build(tmp_path, monkeypatch) + assert calls["init"] == 1 + assert calls["apply_env"] == 0 # .env NOT loaded into process env + + +def test_cloud_settings_ignore_dotenv_files(tmp_path, monkeypatch): + from anton.config.settings import AntonSettings + + monkeypatch.delenv("ANTON_ANTHROPIC_API_KEY", raising=False) + monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) + + envf = tmp_path / "sentinel.env" + envf.write_text("ANTON_ANTHROPIC_API_KEY=DOTENV_SENTINEL\n") + assert AntonSettings(_env_file=str(envf)).anthropic_api_key == "DOTENV_SENTINEL" + + monkeypatch.chdir(tmp_path) + (tmp_path / ".env").write_text("ANTON_ANTHROPIC_API_KEY=DOTENV_SENTINEL\n") + _, cfg = _build(tmp_path, monkeypatch) + assert cfg.settings.anthropic_api_key != "DOTENV_SENTINEL" + + +# ── real enforcement paths ─────────────────────────────────────────────────── + +def _mock_llm(): + from unittest.mock import AsyncMock, MagicMock + + from anton.core.llm.provider import ProviderConnectionInfo + + llm = AsyncMock() + llm.coding_provider = MagicMock() + llm.coding_provider.export_connection_info = MagicMock( + return_value=ProviderConnectionInfo(provider="anthropic", api_key="test") + ) + llm.coding_model = "claude-sonnet-4-6" + llm.planning_provider = MagicMock() + llm.planning_provider.native_web_tools = MagicMock(return_value=set()) + return llm + + +def _real_session_with_workspace(tmp_path, **cfg_overrides): + from unittest.mock import MagicMock + + from anton.core.session import ChatSession, ChatSessionConfig + from anton.workspace import Workspace + + ws = Workspace(tmp_path) + ws.initialize() + session = ChatSession( + ChatSessionConfig(llm_client=_mock_llm(), workspace=ws, **cfg_overrides) + ) + session._scratchpads = MagicMock(available_packages=[]) + return session + + +def test_final_tool_set_equals_allowlist_after_real_build(tmp_path, monkeypatch): + """The registry is built LAZILY at turn time, so the allowlist must be + enforced by the real _build_tools() - assert the EXACT final tool set.""" + from unittest.mock import MagicMock + + monkeypatch.setenv(_WORKSPACE_PATH_ENV, str(tmp_path)) + monkeypatch.setattr( + llm_client_mod.LLMClient, "from_settings", + classmethod(lambda cls, settings: _mock_llm()), + ) + req = TurnRequestV1(protocol_version=1, conversation_id="c", input="hi") + session = build_cloud_chat_session(req) # REAL ChatSession + session._scratchpads = MagicMock(available_packages=[]) + + session._build_tools() + names = {t.name for t in session.tool_registry.get_tool_defs()} + assert names == set(CLOUD_TOOL_ALLOWLIST), ( + f"cloud tool set drifted from the allowlist: {sorted(names)}" + ) + + +def test_tool_allowlist_none_preserves_desktop_tools(tmp_path): + session = _real_session_with_workspace(tmp_path) # allowlist defaults to None + assert session._tool_allowlist is None + session._build_tools() + names = {t.name for t in session.tool_registry.get_tool_defs()} + assert {"launch_backend", "select_path", "scratchpad", "create_artifact"} <= names + + +def test_sanitized_env_contract(monkeypatch): + from anton.core.backends.local import _SCRATCHPAD_ENV_ALLOWLIST, _sanitized_parent_env + + required = {"PATH": "/usr/bin:/bin", "HOME": "/home/anton", "TMPDIR": "/tmp/x", + "LANG": "en_US.UTF-8", "SSL_CERT_FILE": "/etc/ssl/cert.pem"} + for k, v in required.items(): + assert k in _SCRATCHPAD_ENV_ALLOWLIST, f"{k} must be in the contract" + monkeypatch.setenv(k, v) + secrets = {"ANTHROPIC_API_KEY": "sk-ant-x", "ANTON_MINDS_API_KEY": "anton-x", + "ANTON_GATEWAY_TOKEN": "gw-x", "DS_POSTGRES_MAIN__PASSWORD": "ds-x", + "OPENAI_API_KEY": "sk-oai-x", "SOME_RANDOM_SECRET": "nope"} + for k, v in secrets.items(): + monkeypatch.setenv(k, v) + + env = _sanitized_parent_env() + for k, v in required.items(): + assert env.get(k) == v + for k in secrets: + assert k not in env + assert set(env) <= set(_SCRATCHPAD_ENV_ALLOWLIST) + + +async def test_sanitized_scratchpad_cannot_read_secret_env(tmp_path, monkeypatch): + """Real subprocess: with the sanitized factory, secret sentinels in the + parent env must be UNREADABLE from cell code.""" + import json + + monkeypatch.setenv("ANTHROPIC_API_KEY", "sk-ant-SENTINEL") + monkeypatch.setenv("ANTON_GATEWAY_TOKEN", "GATEWAY-SENTINEL") + monkeypatch.setenv("DS_POSTGRES_MAIN__PASSWORD", "DATASOURCE-SENTINEL") + + pad = sanitized_scratchpad_runtime_factory( + name="cloud-sanitize", coding_provider="anthropic", coding_model="", + coding_api_key="", coding_base_url="", cells=None, workspace_path=tmp_path, + ) + await pad.start() + try: + cell = await pad.execute( + "import os, json\n" + "keys = ['ANTHROPIC_API_KEY','ANTON_GATEWAY_TOKEN','DS_POSTGRES_MAIN__PASSWORD']\n" + "print(json.dumps({k: os.environ.get(k) for k in keys}))" + ) + assert cell.error is None, cell.error + seen = json.loads(cell.stdout.strip()) + assert all(v is None for v in seen.values()), f"secret leaked: {seen}" + finally: + await pad.close() + + +async def test_desktop_scratchpad_env_unchanged(tmp_path, monkeypatch): + """Regression: sanitize_env=False (desktop default) STILL inherits the + parent env - the sanitizer is strictly opt-in.""" + from anton.core.backends.local import local_scratchpad_runtime_factory + + monkeypatch.setenv("MY_DESKTOP_SENTINEL", "DESKTOP-VISIBLE") + pad = local_scratchpad_runtime_factory( + name="desktop-env", coding_provider="anthropic", coding_model="", + coding_api_key="", coding_base_url="", cells=None, workspace_path=tmp_path, + ) + await pad.start() + try: + cell = await pad.execute( + "import os; print(os.environ.get('MY_DESKTOP_SENTINEL', 'MISSING'))" + ) + assert cell.error is None, cell.error + assert cell.stdout.strip() == "DESKTOP-VISIBLE" + finally: + await pad.close()