feat(cloud_turn): static buffered turn entrypoint

.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

.github/workflows/scratchpad-dev-build.yml

@@ -0,0 +1,37 @@
+name: Scratchpad image - Dev build on PR
+
+# Builds and pushes the minds-anton-scratchpad image (the sandbox pod image) to ECR as
+# development-<sha> 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:
+
+  build:
+    runs-on: mdb-dev
+    outputs:
+      image: ${{ steps.build.outputs.image }}
+    env:
+      AWS_REGION: us-east-1
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+      - uses: mindsdb/github-actions/setup-env@main
+      - id: build
+        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

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"]

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"]

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())

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 [],
+        )