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

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

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: