Use serve mode instead of headless mode agent.

Author: beacoder

README.md

@@ -8,20 +8,21 @@
 
 The trusted side of an agent platform: API server, controller, auth,
 billing, secrets.  The untrusted side — the agent runtime itself — is
-`python-agent-harness`, executed as a **subprocess** with
-`python-agent-harness headless --json` and consumed as a JSON-lines
-event stream.  The harness is never imported; the repos stay decoupled
-(the harness only needs to be on PATH of whatever runs the agent).
+`python-agent-harness`, executed as a **resident subprocess** per
+sandbox: `python-agent-harness serve`, a bidirectional JSON-lines
+protocol over stdin/stdout.  The harness is never imported; the repos
+stay decoupled (the harness only needs to be on PATH of whatever runs
+the agent).
 
 ```
 Coding Space Server
         │
  ┌──────▼─────────┐
  │ Agent Controller│   app.controller
  └──────┬─────────┘
-        │ subprocess: python-agent-harness headless --json
+        │ resident subprocess: python-agent-harness serve (JSONL over pipes)
  ┌──────▼─────────┐
- │    Sandbox      │   app.controller.runner (LocalRunner now, Docker later)
+ │    Sandbox      │   app.controller.runner (ServerRunner now, Docker later)
  │  agent process  │
  │  tools, bash, … │
  └─────────────────┘
@@ -33,7 +34,7 @@ Controller                 agent-generated commands
 Auth (JWT)                 repository, builds
 Billing (usage ledger)
 Secrets (Fernet)
-Runner (sandbox manager contract; LocalRunner today, Docker later)
+Runner (sandbox manager contract; ServerRunner today, Docker later)
 ```
 
 ## Layout
@@ -49,13 +50,14 @@ app/
   schemas.py       Pydantic request/response models
   routes/
     auth.py        POST /auth/register /auth/login /auth/refresh /auth/me
-    conversations.py  CRUD + POST /{id}/runs (start) + GET /{id}/stream (SSE)
+    conversations.py  CRUD + POST /{id}/runs (start) + POST /{id}/answer
+                      + GET /{id}/stream (SSE)
     billing.py     usage summary (token ledger)
     secrets.py     CRUD (write-only read: value never returned)
   controller/
-    protocol.py    Parse harness --json lines (seq/run_id/result/usage)
-    manager.py     Controller: start_run, event pump, subscribe, cancel
-    runner.py      Runner protocol + LocalRunner (subprocess exec, cancel)
+    protocol.py    Parse harness event lines (seq/run_id/result/usage)
+    manager.py     Controller: start_run, event pump, subscribe, cancel, answer
+    runner.py      Runner protocol + ServerRunner (resident serve process)
   static/index.html  Minimal chat UI (EventSource -> runs, fetch -> API)
 ```
 
@@ -71,6 +73,26 @@ uvicorn app.main:app --reload    # http://127.0.0.1:8000 (UI at /)
 `PAW_HARNESS__CMD` at the absolute binary if it is not.  Auth: create
 a user via `/auth/register`, then log in; the UI does this for you.
 
+## The serve protocol (harness side)
+
+One sandbox = one long-lived `serve` process.  The web side writes
+ops, the harness answers with events:
+
+```
+host → harness: {"op": "submit", "prompt": ..., "run_id": ...}
+                {"op": "answer", "run_id": ..., "answers": [...]}
+                {"op": "cancel", "run_id": ...} / {"op": "ping"} / {"op": "shutdown"}
+harness → host: {"type": "ready"} then per-run
+                start/delta/notify/log lines and a terminal
+                {"type": "result", "answer": ..., "usage": ..., "cancelled": ...}
+```
+
+Because the process is resident: conversation history persists across
+turns (multi-turn memory), no per-turn interpreter spawn, `answer`
+delivers the user's reply to a pending mid-run question (the agent's
+Question tool / PlanExit confirm), and cancel is a protocol message —
+no signal semantics.
+
 ## Configuration (env, prefix `PAW_`)
 
 | var | default | note |
@@ -80,23 +102,24 @@ a user via `/auth/register`, then log in; the UI does this for you.
 | `PAW_ACCESS_TOKEN_MINUTES` | `30` | JWT access TTL |
 | `PAW_REFRESH_TOKEN_DAYS` | `14` | JWT refresh TTL |
 | `PAW_HARNESS__CMD` | `python-agent-harness` | harness binary |
-| `PAW_HARNESS__CWD` | `""` | agent workspace dir per run |
-| `PAW_HARNESS__MAX_ROUNDS` | unset | round budget forwarded to `--max-rounds` |
-| `PAW_HARNESS__TIMEOUT` | unset | wall-clock budget forwarded to `--timeout` |
-| `PAW_RUNNER` | `local` | `local` only; docker later |
+| `PAW_HARNESS__CWD` | `""` | agent workspace dir per sandbox |
+| `PAW_HARNESS__TIMEOUT` | unset | host-side wall-clock budget for one run |
+| `PAW_RUNNER` | `server` | `server` only; docker later |
 | `PAW_SANDBOX__TTL_SECONDS` | `300` | idle reaper TTL |
 
 ## Design notes
 
 - **Decoupling**: the harness is a black-box binary driven by its
-  documented `headless --json` protocol (`start`/`delta`/`notify`/
-  `log`/`result` lines, `seq` for ordering, `run_id` for correlation,
-  `usage` for billing).  No imports, no shared state; the web side
-  can be versioned and deployed independently.
-- **Runs are processes**: one conversation turn = one harness exec =
-  one `Run` row; events stream to subscribers over SSE exactly as the
-  harness emitted them (plus run lifecycle events), and the `result`
-  line lands in the DB.
+  documented JSONL protocols (the same `start`/`delta`/`notify`/
+  `log`/`result` line shapes on the resident `serve` pipe and the
+  one-shot `headless --json` pipe; `seq` for ordering, `run_id` for
+  correlation, `usage` for billing).  No imports, no shared state; the
+  web side can be versioned and deployed independently.
+- **Runs are protocol turns, not process lifecycles**: one
+  conversation turn = one `op:submit` = one `Run` row; the resident
+  process survives the run and serves the next turn.  Events stream to
+  subscribers over SSE exactly as the harness emitted them (plus run
+  lifecycle events), and the `result` line lands in the DB.
 - **Secrets** are Fernet-encrypted at rest and never returned by the
   API; they are meant to be injected into the sandbox environment by
   the sandbox manager (not exposed to agents via the API).

app/controller/__init__.py

@@ -2,15 +2,15 @@
 
 from .manager import Controller
 from .protocol import ProtocolEvent, RunOutcome, apply_event, parse_line, parse_stream
-from .runner import ExecResult, LocalRunner, Runner, get_runner
+from .runner import ExecResult, Runner, ServerRunner, get_runner
 
 __all__ = [
     "Controller",
     "ExecResult",
-    "LocalRunner",
     "ProtocolEvent",
     "RunOutcome",
     "Runner",
+    "ServerRunner",
     "apply_event",
     "get_runner",
     "parse_line",

app/controller/manager.py

@@ -133,10 +133,10 @@ def _worker() -> None:
         return run
 
     def cancel_run(self, run_id: str) -> bool:
-        """Best-effort graceful cancel: signal the harness process via
-        the runner (SIGINT path; tools get their salvage window) and
-        mark the intent on the run row.  The run's own exit (result
-        line with ``cancelled: true``, or error) finalizes the row."""
+        """Best-effort graceful cancel: send the protocol cancel op to
+        the resident harness process and mark the intent on the run
+        row.  The run's own exit (result line with ``cancelled: true``,
+        or error) finalizes the row."""
         with self._lock:
             active = self._active.get(run_id)
         if active is None:
@@ -150,6 +150,22 @@ def cancel_run(self, run_id: str) -> bool:
                 db.commit()
         return True
 
+    def deliver_answer(self, run_id: str, answers: list[str]) -> bool:
+        """Forward a user's answer to a pending mid-run question.
+
+        Returns False when the run is not live (unknown/finished); the
+        route maps that to 409.  The runner forwards the answer as an
+        ``op:answer`` protocol message to the resident harness process.
+        """
+        with self._lock:
+            active = self._active.get(run_id)
+        if active is None:
+            return False
+        delivered = getattr(self._runner, "deliver_answer", None)
+        if delivered is None:
+            return False
+        return bool(delivered(active.sandbox_id, run_id, answers))
+
     # -- internals ---------------------------------------------------------
 
     def _ensure_sandbox(self, db: Session, *, user_id: str, conversation_id: str) -> Sandbox:

app/controller/protocol.py

@@ -1,11 +1,12 @@
-"""Protocol types and parsing for the harness ``headless --json`` stream.
-
-The harness emits one JSON object per line on stdout
-(``start``/``delta``/``notify``/``log``/``result``, each carrying
-``seq`` and ``run_id``; ``result`` carries ``answer``, ``errors``,
-``usage``, ``model``, ``cancelled``).  This module turns raw output
-into typed events — the *only* coupling point with the harness repo,
-and purely a data contract.
+"""Protocol types and parsing for the harness event stream.
+
+The harness emits one JSON object per line on stdout (``start``/
+``delta``/``notify``/``log``/``result``, each carrying ``seq`` and
+``run_id``; ``result`` carries ``answer``, ``errors``, ``usage``,
+``model``, ``cancelled``) — the same line shapes on both the one-shot
+``headless --json`` pipe and the resident ``serve`` protocol.  This
+module turns raw lines into typed events — the *only* coupling point
+with the harness repo, and purely a data contract.
 """
 
 from __future__ import annotations
@@ -14,7 +15,7 @@
 from dataclasses import dataclass, field
 from typing import Any
 
-LINE_TYPES = {"start", "delta", "notify", "log", "result"}
+LINE_TYPES = {"start", "delta", "notify", "log", "result", "error"}
 
 
 @dataclass
@@ -87,7 +88,12 @@ def apply_event(outcome: RunOutcome, event: ProtocolEvent) -> None:
         outcome.answer = str(event.data.get("answer", ""))
         errors = event.data.get("errors")
         if isinstance(errors, list):
-            outcome.errors = [str(e) for e in errors]
+            # the result line is canonical for its own errors, but must
+            # not wipe protocol errors folded from earlier lines (e.g.
+            # a rejected mid-run answer): merge, preserving order
+            for e in (str(e) for e in errors):
+                if e not in outcome.errors:
+                    outcome.errors.append(e)
         usage = event.data.get("usage")
         outcome.usage = dict(usage) if isinstance(usage, dict) else None
         outcome.model = event.data.get("model")
@@ -96,6 +102,13 @@ def apply_event(outcome: RunOutcome, event: ProtocolEvent) -> None:
         message = event.data.get("data")
         if message is not None and str(message) not in outcome.errors:
             outcome.errors.append(str(message))
+    elif event.type == "error":
+        # harness protocol-level failure (unknown op, stale answer, a
+        # cancel racing the run's finish): keep it in the run's error
+        # trail so it surfaces instead of vanishing
+        message = event.data.get("error")
+        if message is not None and str(message) not in outcome.errors:
+            outcome.errors.append(str(message))
 
 
 def parse_stream(text: str) -> list[ProtocolEvent]: