feat(agents): queue only sandbox work; per-user message limits

agents/README.md

@@ -15,7 +15,7 @@ The model is **Claude Haiku 5.5 on Vertex AI** (`claude-haiku-5-5`) by default.
 | `main.py` | The `anyplot-agents` FastAPI service: the `/v1` routes the BFF (`api/routers/agent.py`) calls, the caller check, the run registry behind `409 run_active`, in-memory session and artifact services, and the idle sweeper |
 | `stream.py` | The `anyplot/1` stream translator: ADK events in, sanitised `ready`, `status` (including `queued`), `message`, `plot`, `refusal`, `error` and `done` events out |
 | `Dockerfile`, `cloudbuild.yaml` | The `anyplot-agents` image (the `agents` extra, the source tree with its prompts and fonts, uid 1000) and its deploy: build, a `candidate` revision without traffic, the IAM and `/v1/status` smoke, promote. The config's header lists the owner's bootstrap steps |
-| `anyplot/run_queue.py` | The run queue in front of every `/messages` turn: one run in flight, one start a minute (a turn that cannot render, because the data is unconfirmed, takes no start), a 600-second maximum wait, a `premium` lane that nothing sets yet |
+| `anyplot/run_queue.py` | The run queue in front of the sandbox runs (since 2026-10-14 a chat turn starts at once, and only its `plot_pipeline` or `run_data_code` call waits, mid-turn): one run in flight, one start a minute, a 600-second maximum wait, a `premium` lane that nothing sets yet |
 | `anyplot/theme_render.py` | The theme toggle: renders another theme of a finished version from its stored run form, behind waiting pipeline renders, with no model call |
 | `anyplot/agent.py` | The root agent `anyplot`, the `ALL_AGENTS` registry and `app` (the ADK `App` with its plugins), which `adk web` loads |
 | `anyplot/models.py` | The only place that builds a model or a model client: `make_model`, `make_content_config` and `make_judge_client`, for Claude on Vertex AI and for Gemini |
@@ -82,9 +82,12 @@ Three rules hold for everything here:
 | `AGENT_RENDER_IMAGE` | `anyplot-renderer:dev` | Image the `local` renderer runs, built from `renderer/Dockerfile` |
 | `AGENT_RENDER_CONCURRENCY` | `1` | Theme renders at the same time, for every backend (`render/serial.py`); serial, because one 4 GiB instance holds one sandbox safely (spikes S and S2) |
 | `AGENT_WATERMARK` | `true` | Whether the served PNGs (the artifact route and the feedback bundle) carry the footer strip; `false` serves the raw renders. The gates, the reviewer and `plot.py` always see the raw render |
-| `AGENT_RUN_CONCURRENCY` | `1` | Pipeline runs (whole `/messages` turns) in flight; the run queue holds the rest |
-| `AGENT_RUNS_PER_MINUTE` | `1` | Runs that may start within any 60 seconds (a sliding window) |
-| `AGENT_QUEUE_MAX_WAIT_S` | `600` | Longest wait in the run queue, after which the run ends with `capacity`; the queue holds rate x wait / 60 entries (10) and answers `503 capacity` beyond that. Through the BFF the whole turn, this wait included, ends by `AGENT_TURN_MAX_S` (1,190 s, see `docs/reference/api.md`), which the full wait plus the 540 s run fits into |
+| `AGENT_RUN_CONCURRENCY` | `1` | Sandbox runs in flight (a turn's `plot_pipeline` or `run_data_code` call, holding its entry until the turn ends); the run queue holds the rest. A chat turn without one never enters the queue |
+| `AGENT_RUNS_PER_MINUTE` | `1` | Sandbox runs that may start within any 60 seconds (a sliding window), one per turn |
+| `AGENT_QUEUE_MAX_WAIT_S` | `600` | Longest wait in the run queue, after which the turn ends with `error{code:"capacity"}`; the queue holds rate x wait / 60 entries (10), and a sandbox run beyond that ends its turn the same way. Through the BFF the whole turn, this wait included, ends by `AGENT_TURN_MAX_S` (1,190 s, see `docs/reference/api.md`), which the root's first call, the full wait and the 540 s run fit into |
+| `AGENT_USER_MESSAGES_PER_MINUTE` | `10` | Chat turns (typed messages and actions) per user within any 60 seconds; one more gets the fixed `rate_limited` refusal before the scope judge, so it costs no model call |
+| `AGENT_USER_MESSAGES_PER_DAY` | `200` | Chat turns per user and UTC day; one more gets the fixed `daily_limit` refusal before the scope judge |
+| `AGENT_SESSION_REFUSALS` | `3` | Scope-judge refusals of type `out_of_scope` or `attack` (never `needs_data`) after which a session is closed: that refusal and every later turn answer `session_closed`, and the user reloads the page for a new session |
 | `AGENT_MAX_ATTEMPTS` | `2` | Rounds per pipeline run, 1 to 3: the first and at most one repair round by default (a self-review adds an adapter call to a round, never a round); the eval harness sets 1 or 3 (`--max-attempts`) to measure no repair or a second repair round |
 | `AGENT_REVIEWER_MODE` | `catalogue` | The reviewer's rubric: `catalogue` (the offline scorer's rubric, verdict decided by code from the points) or `checklist` (the narrow checklist used before 2026-10-13) |
 | `AGENT_REVIEW_BAR`, `AGENT_STATUS_BAR` | `1.0`, `0.95` | `catalogue` reviews: the share of technical points (VQ, SC, DQ) below which a render gets the repair, and the ship bar at or above which it shows without notes |
@@ -103,7 +106,7 @@ Three rules hold for everything here:
 | `AGENT_RENDER_TIMEOUT_S` | `60` | Seconds per theme render |
 | `AGENT_DATA_CODE` | `true` | Whether the root's `run_data_code` runs; `false` makes it answer `unavailable`. The tool also answers `unavailable` while the renderer has no `POST /run-data` (`404`) |
 | `AGENT_DATA_RUN_TIMEOUT_S` | `20` | Seconds one data snippet may run, at most 30 |
-| `AGENT_DAILY_DATA_RUNS` | `30` | `run_data_code` runs per user and day, counted before each run; an unconfirmed turn also books a start in the run queue's rate window for its first data run |
+| `AGENT_DAILY_DATA_RUNS` | `30` | `run_data_code` runs per user and day, counted before each run; a data run also takes the run queue's slot and its turn's start in the rate window, like a plot |
 | `AGENT_REQUEST_DEADLINE_S` | `540` | Hard request deadline (180 before 2026-10-13: with a self-review per round, the soft deadline of 140 s stopped the repair round in 11 of 20 runs; the owner's rule is that no plot ships unfinished for lack of time) |
 | `AGENT_SOFT_DEADLINE_S` | `480` | Soft deadline inside the pipeline |
 | `AGENT_JUDGE_TIMEOUT_S` | `4` | Budget of one judge verdict, its one retry and the half-second wait before it included |
@@ -206,7 +209,7 @@ uv run uvicorn agents.main:app --port 8001
 
 The service needs the header `X-Anyplot-User` on every `/v1` route; outside `ENVIRONMENT=development` it also requires the IAM-forwarded ID token (`AGENT_SERVICE_URLS`, `AGENT_ALLOWED_CALLERS`). To drive it from the plot page, run the API in front of it as described under [Drive the chat page](#drive-the-chat-page).
 
-At the defaults only one run may start a minute, so a second "Create plot" within a minute waits in the run queue and the stream shows `status` events with `step: "queued"`. A turn while the data is unconfirmed (the summary after a paste, a data request in step 1) cannot render and takes no start, only the slot, so step 1 never waits for the minute. To iterate faster on your own machine, export `AGENT_RUNS_PER_MINUTE=60`. The theme toggle (`POST /v1/sessions/{sid}/versions/{version}/render {"theme": "dark"}`) never waits in the queue.
+At the defaults only one sandbox run may start a minute, so a second "Create plot" within a minute waits in the run queue: the turn starts at once, and when its pipeline is about to run, the stream shows `status` events with `step: "queued"` until the minute is over. A turn that runs no sandbox (a question, a refusal, the summary after a paste, a data request in step 1) never waits. To iterate faster on your own machine, export `AGENT_RUNS_PER_MINUTE=60`; more than 10 chat turns a minute get the `rate_limited` refusal (`AGENT_USER_MESSAGES_PER_MINUTE`). The theme toggle (`POST /v1/sessions/{sid}/versions/{version}/render {"theme": "dark"}`) never waits in the queue.
 
 ### Drive the chat page
 

agents/anyplot/pipeline.py

@@ -57,7 +57,10 @@
 findings, the G10 line of a render whose text has characters without a glyph, which a
 reviewer `ok` does not erase, a padded canvas, or no review because the budget or the
 deadline ran out);
-`failed` names its reason; `not_ready` comes before any model call. An `ok` or
+`failed` names its reason; `not_ready` comes before any model call and before the run
+queue. A ready run first takes the turn's run-queue slot (`ledger.run_slot`, set by the
+`/messages` route), so only a turn that renders waits in the queue; the soft deadline
+starts after that wait. An `ok` or
 `needs_attention` result is stored as the session's next version before it is
 yielded, and carries that number as `version`. Its `changes` are the notes of every
 adapter call up to the shipped render whose plan changed the code (`note_changes`), so
@@ -986,6 +989,11 @@ async def run_pipeline(ctx: Context, node_input: PipelineArgs) -> AsyncGenerator
         yield _result_event(PlotResult.not_ready(not_ready or "no_dataset"))
         return
     ledger.pipeline_active = True
+    if ledger.run_slot is not None and not await ledger.run_slot():
+        # No slot: the route failed the turn with `capacity`, or the user cancelled the wait. Either way the run is
+        # aborted, a cancelled path, so no result: the stream ends with the error, or with `done` after a cancel.
+        ledger.pipeline_active = False
+        return
     theme = node_input.theme or _default_theme(services, ctx.session.id, view.library, node_input.base)
     run = Run(view=view, dataset=dataset, theme=theme, language=view.language, change_request=node_input.change_request)
     scope = f"pipeline-{secrets.token_hex(6)}"

agents/anyplot/plugins/ledger.py

@@ -15,7 +15,10 @@
 `attack` verdicts of the scope and dataset judges per user, once per distinct message
 or dataset: from `AGENT_ATTACK_STRIKES` of them on, `daily_ok` is false for that user
 for the rest of the day, so a turn gets the budget refusal and a dataset
-`429 rate_limited`. Only a new UTC day, a redeploy or a scale-to-zero resets them.
+`429 rate_limited`. It also counts each user's chat turns, per UTC day
+(`AGENT_USER_MESSAGES_PER_DAY`) and in a sliding minute (`AGENT_USER_MESSAGES_PER_MINUTE`),
+which the `/messages` route checks before the scope judge (`message_limit`). Only a new
+UTC day, a redeploy or a scale-to-zero resets them.
 
 The request, user-day and global-day budgets count cost-weighted tokens: each kind
 weighted by its price relative to an uncached input token (`BUDGET_WEIGHTS`), so a
@@ -31,8 +34,9 @@
 import hashlib
 import json
 import logging
-from collections import OrderedDict
-from collections.abc import Callable
+import time
+from collections import OrderedDict, deque
+from collections.abc import Awaitable, Callable
 from contextvars import ContextVar
 from dataclasses import dataclass, field
 from datetime import UTC, datetime
@@ -105,11 +109,13 @@ class RequestLedger:
     """The change lines of every `prepare_data` call of the turn, in order: the `done` event's `dataset.changes`."""
     data_code_calls: int = 0
     """`run_data_code` calls of this invocation; ToolSafety allows `MAX_DATA_CODE_CALLS`."""
-    rate_start: Callable[[], bool] | None = None
-    """Set by the `/messages` route: books a start in the run queue's rate window for this turn (`RunQueue.rate`),
-    False when the window is full; idempotent per turn. None without a route (`adk web`), whose runs bypass the
-    queue. The one helper for every sandbox run of a turn: `run_data_code` calls it today, and the plot pipeline can
-    take its rated start through it once the queue no longer rates whole turns."""
+    run_slot: Callable[[], Awaitable[bool]] | None = None
+    """Set by the `/messages` route (`TurnSlot.take`): waits for the run queue's slot and the turn's start in the
+    rate window, while the stream sends `status{step:"queued"}`, and holds both until the turn ends. True once the
+    turn holds them (at once on a second call); False when the queue is full, the wait ran out or the turn was
+    cancelled, after the route failed the turn with `capacity` (or ended it). None without a route (`adk web`),
+    whose runs bypass the queue. The one helper for every sandbox run of a turn: `plot_pipeline` and
+    `run_data_code` await it before they render or run code."""
 
     def refuse(self, code: str, text: str) -> None:
         if self.refusal is None and self.error is None:
@@ -156,13 +162,22 @@ class _Day:
     global_tokens: int = 0
     user_strikes: dict[str, set[str]] = field(default_factory=dict)
     user_data_runs: dict[str, int] = field(default_factory=dict)
+    user_messages: dict[str, int] = field(default_factory=dict)
+
+
+MESSAGE_WINDOW_S = 60.0
+"""The sliding window of `AGENT_USER_MESSAGES_PER_MINUTE`."""
+MessageLimit = Literal["rate_limited", "daily_limit"]
+"""Why `message_limit` refuses a turn: the minute's messages, or the day's."""
 
 
 class UsageBook:
-    """Per-user and global daily counters for this instance's lifetime."""
+    """Per-user and global daily counters for this instance's lifetime, and each user's messages of the last minute."""
 
-    def __init__(self) -> None:
+    def __init__(self, clock: Callable[[], float] = time.monotonic) -> None:
         self._day = _Day(today())
+        self._clock = clock
+        self._recent: dict[str, deque[float]] = {}
 
     def _current(self) -> _Day:
         date = today()
@@ -225,6 +240,42 @@ def user_data_runs(self, user_id: str) -> int:
     def data_runs_ok(self, user_id: str, settings: AgentSettings) -> bool:
         return self.user_data_runs(user_id) < settings.daily_data_runs
 
+    def user_messages(self, user_id: str) -> int:
+        """The user's chat turns today (typed messages and actions alike) that passed `message_limit`."""
+        return self._current().user_messages.get(user_id, 0)
+
+    def message_limit(self, user_id: str, settings: AgentSettings) -> MessageLimit | None:
+        """Why the user may not send another chat turn now, or None; checked before the scope judge, so a flood costs
+        no model call. A refused turn is not counted: the window frees itself one minute after the last counted one."""
+        if self.user_messages(user_id) >= settings.user_messages_per_day:
+            return "daily_limit"
+        recent = self._trimmed(user_id)
+        if recent is not None and len(recent) >= settings.user_messages_per_minute:
+            return "rate_limited"
+        return None
+
+    def add_message(self, user_id: str) -> None:
+        """Count one chat turn against the user's minute and day."""
+        day = self._current()
+        day.user_messages[user_id] = day.user_messages.get(user_id, 0) + 1
+        recent = self._trimmed(user_id)
+        if recent is None:
+            recent = self._recent[user_id] = deque()
+        recent.append(self._clock())
+
+    def _trimmed(self, user_id: str) -> deque[float] | None:
+        """The user's message times within the window, oldest first; a user with none left is forgotten."""
+        recent = self._recent.get(user_id)
+        if recent is None:
+            return None
+        now = self._clock()
+        while recent and now - recent[0] >= MESSAGE_WINDOW_S:
+            recent.popleft()
+        if not recent:
+            del self._recent[user_id]
+            return None
+        return recent
+
 
 def request_ok(ledger: RequestLedger, settings: AgentSettings, *, next_calls: int = 1, reserve: int = 0) -> bool:
     """Whether the request may make `next_calls` more LLM calls and still has `reserve` cost-weighted tokens left."""

agents/anyplot/plugins/scope_guard.py

@@ -25,6 +25,12 @@
 * an `attack` verdict is a strike against the user (`UsageBook.add_strike`), once
   per distinct message text: from `AGENT_ATTACK_STRIKES` strikes on, the daily check
   above refuses the user with the budget refusal for the rest of the day;
+* the session's `AGENT_SESSION_REFUSALS`-th refusal of type `out_of_scope` or `attack`
+  (3 by default; `needs_data` never counts, `scope_refusals`) closes the session: that
+  refusal goes out as `session_closed` instead of `out_of_scope`, and the `/messages`
+  route answers every later turn of the session with the same refusal before the
+  judge runs, so the user starts a new session. The strikes above are per user and
+  day and count `attack` only; this count is per session and needs no state of its own;
 * the judge's `lang` for the session's first in-scope typed message becomes the
   session language with the source `detected` (`session_state.LANGUAGE`), unless one
   is set: a later message never changes it, a choice (`chosen`) never yields to it, a
@@ -137,6 +143,29 @@ def _withheld(code: str | None = None) -> types.Content:
     return types.Content(role="user", parts=[types.Part(text=withheld_text(code))])
 
 
+SESSION_CLOSED = "session_closed"
+"""The refusal code (and `refusals.yaml` entry) of a session closed after `AGENT_SESSION_REFUSALS` refusals."""
+COUNTED_PLACEHOLDER = withheld_text("out_of_scope")
+"""The stored form of every `out_of_scope` and `attack` refusal: what `scope_refusals` counts."""
+
+
+def scope_refusals(events: Sequence[Event]) -> int:
+    """The session's scope-judge refusals of type `out_of_scope` or `attack`, counted from its withheld messages.
+
+    Both verdicts store the same placeholder (`withheld_text("out_of_scope")`), a `needs_data`
+    refusal its own, and a refusal sent before the judge (budget, unsupported content, the message
+    limits, a closed session) the bare one or nothing, so only the counted kinds match. The count
+    lives in the session's own events: it ends with the session, and no state is written for it.
+    """
+    count = 0
+    for event in events:
+        if event.author != USER_AUTHOR or not event.content or not event.content.parts:
+            continue
+        if "".join(part.text or "" for part in event.content.parts) == COUNTED_PLACEHOLDER:
+            count += 1
+    return count
+
+
 def session_turns(invocation_context: InvocationContext) -> list[Turn]:
     """The session's text turns so far, oldest first: the user's messages and the replies they got.
 
@@ -267,7 +296,13 @@ async def _judge(
                     KeptRefusal(verdict.verdict, verdict.lang, text, datetime.now(UTC).isoformat(timespec="seconds")),
                 )
             code = refusal_code(verdict.verdict)
-            ledger.refuse(code, refusal(code, verdict.lang))
+            shown = code
+            if code == "out_of_scope":
+                count = scope_refusals(invocation_context.session.events) + 1
+                if count >= settings.session_refusals:
+                    shown = SESSION_CLOSED  # the route refuses every later turn of the session before the judge
+                    attribution("scope_guard", ledger, verdict=SESSION_CLOSED, refusals=count)
+            ledger.refuse(shown, refusal(shown, verdict.lang))
             return _withheld(code)
         await self._detect(invocation_context, ledger, verdict, text)
         return None