fix(environment): bound data-plane replies, keep batch order, cancel intake on stop

- One /agentenv response is held to 64 MiB: replies are measured as they
  are produced and one past the budget becomes a -32000 error stating it.
- Every message of a batch enters the arrival-order queue on arrival, so
  another request's reset never lands between them.
- Stopping the server, or a caller leaving, cancels the active download
  and answers queued calls `cancelled` without running them.
- file:// URIs name only this host: another host or a network share is
  refused before anything is opened.
- A loaded view value may not give an entity another identity.
- List answers refuse a member lacking a field every recording carried,
  as item answers do.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

docs/agent-environments.md

@@ -388,7 +388,7 @@ Each read tool answers according to its view in the state:
 | View       | Answer                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
 | ---------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
 | `item`     | The entity the path arguments name, rendered into the recorded envelope of `envelope_ordinal`. Arguments match as replay matches them: each collection segment equals the held collection's segment as written or after percent-decoding, and the identity is the last argument percent-decoded, then as written; values are never re-split on `/`. Any entity the state holds answers, including one `data/add` loaded; query arguments only decide `replay.exact_match`. An entity the state does not hold, or one lacking a field every recording carried, is a tool error; a body is never invented. |
-| `list`     | When the arguments name one of the lines the view checked exactly (its latest recordings before the first related write; every path and query value equal to the recording's, and every recorded query parameter supplied), that recorded envelope with its array rendered from the state: every entity of the collection in state order for `membership: "all"`, or the recorded members still held.                                                                                                                                                                                                    |
+| `list`     | When the arguments name one of the lines the view checked exactly (its latest recordings before the first related write; every path and query value equal to the recording's, and every recorded query parameter supplied), that recorded envelope with its array rendered from the state: every entity of the collection in state order for `membership: "all"`, or the recorded members still held. A member lacking a field every recording carried makes the call a tool error rather than a partial list.                                                                                           |
 | `document` | When the arguments name the document's checked line exactly, the state's value for it; a tool error when the state holds none.                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
 | `recorded` | The recording, as replay answers it.                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
 
@@ -412,21 +412,42 @@ The data plane serves three JSON-RPC methods:
 | `data/add`   | `{}` after replacing the whole state with one valid `rea-state/v1` document                               |
 | `data/get`   | `{"parts": [{"kind": "data", "data": <rea-state/v1 document>}]}` with `state_evidence_id` and the `types` |
 
-Calls run one at a time in arrival order, across requests as within a batch,
+Calls run one at a time in arrival order across requests. Every message of a
-so a reset sent while a download is loading applies after that load. `data/add`
+batch is queued when the batch arrives, so another request's call never lands
+between them and each message sees the changes of those before it; a reset
+sent while a download is loading applies after that load. `data/add`
 takes exactly one part: a `data` part whose `data` is the document, or a
 `file` part whose `file` carries `bytes` (base64 of UTF-8 JSON) or a `uri`,
-either a `file://` URL of a local regular file or an `https://` URL fetched
+either a `file://` URL of a local regular file, whose host is empty or
-without following redirects and within 30 seconds, including reading the body;
+`localhost`, or an `https://` URL fetched without following redirects and
-`mimeType`, when given, is `application/json`. A loaded document is at most 32
+within 30 seconds, including reading the body; `mimeType`, when given, is
+`application/json`. A `file://` URL naming another host, or a UNC path such as
+`file:////server/share/state.json`, is refused before anything is opened, so a
+load never reaches a network share. A loaded document is at most 32
 MiB; an inline part also stays within the 4 MiB request body bound, so larger
 states arrive by `uri`. Anything else, including `{"parts": []}`, is `-32602
 invalid params`; a download that fails or answers a non-2xx status is
 `-32000`. Errors never echo the uri or the content; a file or download failure
 names its category in `error.data.reason`: `not-found`, `permission`,
-`not-regular-file`, `changed` or `read-failed` for files, and `timeout`,
+`not-regular-file`, `changed` or `read-failed` for files, `timeout`,
 `http-status` or `network` (connection, TLS or a refused redirect) for
-downloads.
+downloads, and `cancelled` for a download the server's stop or its caller's
+departure ended.
+
+Stopping the server cancels every pending call, and a caller that leaves before
+its reply cancels its own: a download in progress ends, and calls not yet run
+are answered `-32000` with `error.data.reason` `cancelled` without running. A
+cancelled call leaves the state unchanged, and shutdown never waits for a
+download.
+
+One `/agentenv` response carries at most 64 MiB of replies, twice the 32 MiB
+a loaded document may hold: room for a `data/get` of a loaded state, while a
+batch of reads cannot multiply a large state into one body. Replies are
+measured in order as they are produced; one that would take the response past
+the budget is answered instead with `-32000` whose `error.data` gives
+`max_response_bytes` and the `reply_bytes` it would have taken. That call still
+ran, so repeat such a read in another request. A reply no larger than that
+error, such as a write's `{}`, is always kept.
 
 ```json
 {
@@ -449,7 +470,9 @@ Each entity's `type` is a state type's `collection_template`, its `collection`
 matches that template segment by segment (a template parameter matches any one
 segment), its identity field's value is its `id`, and its type, collection and
 id are unique; `view_fields` is optional and names item or list views of its
-type. Spell a collection as requests spell it, as `data/get` reports it: list
+type. Views render their values over the entity's fields, so a view's values
+must not set the type's identity field to a value whose text differs from the
+entity's `id`. Spell a collection as requests spell it, as `data/get` reports it: list
 calls find entities under the checked line's recorded path. Each document
 names a `document` view once, at its initial document's `path`. The `version`,
 and the `state_evidence_id` and `types` that `data/get` reports, must match

docs/specs/agent-environments/state-serve/plan.md

@@ -21,11 +21,13 @@
 5. Declare the state data plane on the card in `environmentTools.ts`, taking
    the data plane in an options object so the discovery budget reserves the
    card and descriptions actually served.
-6. Add `stateIntake.ts` for parts, files and downloads (injectable fetcher and
+6. Add `stateIntake.ts` for parts, files on this host and downloads
-   timeout, coarse failure categories) and `environmentDataPlane.ts` for the
+   (injectable fetcher and timeout, cancellation, coarse failure categories)
-   three methods, run one at a time in arrival order; let `serveEnvironment.ts`
+   and `environmentDataPlane.ts` for the three methods, run one at a time in
-   serve the shared catalog and answer function and run data-plane methods,
+   arrival order and cancelled by the server's stop or the caller's
-   batches in order and notifications.
+   departure; let `serveEnvironment.ts` serve the shared catalog and answer
+   function, queue every message of a batch at arrival, bound the replies of
+   one response and stop the data plane before closing the listener.
 7. Make `environment-serve` prepare the state under its served name after
    loading the source, add `--replay-only`, log an unavailable state as a
    warning before binding, and report `state` in the stopped document.

docs/specs/agent-environments/state-serve/spec.md

@@ -59,8 +59,11 @@ this increment: write tools are not served from the state.
   view's `checked_record_ordinals` exactly (every path and query value equal and
   no recorded query parameter unsupplied), choosing among them as replay
   chooses: every collection entity in state order for `membership: "all"`,
-  otherwise the recorded members the state still holds. Lines recorded after a
+  otherwise the recorded members the state still holds. A member lacking one
-  write or with other filters were never checked and answer from recordings.
+  of the view's `required_fields` makes the call a tool error stating that the
+  state holds no complete member, never a partial body, as for items. Lines
+  recorded after a write or with other filters were never checked and answer
+  from recordings.
 - Answer document views with the state's value when the arguments name the
   document's checked line exactly.
 - Answer recorded views, tools without a view and unrecorded requests from the
@@ -72,14 +75,30 @@ this increment: write tools are not served from the state.
   (`bytes`, a `file://` regular file or an `https://` download with a 30 s
   timeout covering the body, no redirects and a streamed 32 MiB cap;
   `mimeType` when given is `application/json`) and `data/get` (`{parts:
-  [{kind: "data", data: rea_state}]}`). Anything else is `-32602`, including
+  [{kind: "data", data: rea_state}]}`). A `file://` uri whose host is neither
-  empty `parts`; a failed download is `-32000`; errors never echo a uri or
+  empty nor `localhost`, or whose path is a UNC path, is `-32602` before
-  content and name a coarse failure category in `error.data.reason`. Calls run
+  anything is opened. Anything else is `-32602`, including empty `parts`; a
-  one at a time in arrival order across requests, batches in order, and
+  failed download is `-32000`; errors never echo a uri or content and name a
+  coarse failure category in `error.data.reason`. Calls run one at a time in
+  arrival order across requests; every message of a batch is queued when the
+  batch arrives, so no other request's call lands between them, and
   notifications run without a reply.
+- Stopping the server (`close` or `prepareShutdown`) aborts the data plane's
+  stop signal, and a request's own signal aborts when its client leaves: the
+  download in progress ends with reason `cancelled`, queued calls are answered
+  `-32000` with reason `cancelled` without running, and a `data/add` whose
+  file read finished after the cancellation does not replace the state.
+- Bound one `/agentenv` response to 64 MiB of replies (twice the intake
+  limit). Each reply is serialized and counted as it is produced, in message
+  order; one that would pass the budget is replaced by `-32000` with
+  `data: {max_response_bytes, reply_bytes}`, unless that error is no smaller
+  than the reply, so the body is never built past the budget and small write
+  replies are kept.
 - Validate `rea-state/v1` documents against the served types and views: known
   types, collections that instantiate their template, ids equal to the identity
-  field, unique keys, `view_fields` naming item or list views of the type,
+  field, unique keys, `view_fields` naming item or list views of the type
+  without setting the type's identity field to a value whose scalar text
+  differs from the entity's id (refused at that view's `fields`),
   documents naming document views once at the initial path, and the optional
   `state_evidence_id` and `types` that `data/get` reports equal to the served
   ones. Refuse numbers without an exact native value and `__proto__` members.
@@ -140,14 +159,18 @@ every answer path: item (including percent-encoded collection segments and
 ids), list `all`, list `recorded`, lines recorded after a write or with other
 filters, a line replay's tie-break would not pick, document, recorded view, tool
 without a view, unrecorded request, missing entity, missing required field and
-the body limit. Adapters tests cover the data-plane shapes, `-32602` for empty
+the body limit, a view value that changes an entity's identity, and a list
-and malformed parts and invalid documents, inexact numbers, bytes, `file://`
+member missing a required field. Adapters tests cover the data-plane shapes,
-and its failure categories, injected https downloads (status, declared and
+`-32602` for empty and malformed parts and invalid documents, inexact numbers,
-streamed size, the timeout whether or not the fetcher honours its signal), the
+bytes, `file://` and its failure categories, `file://` uris naming another
-default fetch's refused redirects, arrival-order application across requests,
+host or a UNC path, injected https downloads (status, declared and streamed
-batches and notifications, the card declaration, absent write tools, a served
+size, the timeout whether or not the fetcher honours its signal), the default
-query `POST`, the per-mode discovery reservation and a `data/add` that changes
+fetch's refused redirects, arrival-order application across requests, a batch
-the next tool answer over MCP. The mcp-boundary tests serve the state the
+another request's reset cannot split, cancellation by the server's stop and by
+a departed caller, a server that closes while a download is stalled, the
+response budget, batches and notifications, the card declaration, absent write
+tools, a served query `POST`, the per-mode discovery reservation and a
+`data/add` that changes the next tool answer over MCP. The mcp-boundary tests serve the state the
 built CLI derives from only a manifest and a capture, and `--replay-only` with
 every manifest tool, `-32601` data-plane answers and a card whose SHA-256
 equals that of the card main serves for the fixture; the process-boundary test checks the stopped document's

docs/specs/agent-environments/state-serve/tasks.md

@@ -24,4 +24,9 @@
       exceeds the discovery budget, check the bundle name before preparing the
       state, keep replay-only out of export plans, test a check the state
       answers differently from replay, and pin main's replay card digest.
+- [x] Address review: bound one `/agentenv` response's replies, queue a
+      batch's messages together at arrival, cancel the running download and
+      queued calls when the server stops or the caller leaves, refuse
+      `file://` uris naming another host or a UNC path, refuse view values
+      that change an entity's identity, and apply list views' required fields.
 - [x] Run static, formatting, duplication, documentation and test gates.

docs/specs/agent-environments/state-serve/verification.md

@@ -1,78 +1,63 @@
 # Verification
 
-Verified base: e96484b10a1ae3dad8541d748599da32b447495e (main with #5319 and
+Verified base: 6676582675e08574146a812ae58cff67e2e53282 (this branch's head),
-#5349 merged into this branch), with the changes of this pass uncommitted.
+with the review fixes of this pass uncommitted.
 
-Executed on the implementation worktree with Node v22.23.1 (the repository pins
+Executed on the implementation worktree on macOS with Node v22.23.1 (the
-24.18.0 in `.nvmrc`, which was not installed). Every check below ran as written
+repository pins 24.18.0 in `.nvmrc`, which was not installed). Every check
-after the last source change, in this order.
+below ran as written after the last source change, in this order.
 
 - `npm run format` then `npm run format:check`: passed (4330 files).
-- `npm run check:fast`: typecheck and lint passed (0 lint errors). The lint
+- `npm run check:fast`: typecheck and lint passed. The lint report holds 0
-  report holds no diagnostic in any file this change touches.
+  errors and no diagnostic in any file this pass touches.
 - `npx knip`: passed (exit 0, no findings).
 - `npm run jscpd`: 3.37% duplicated lines over 3373 files (2800 clones), under
-  the 3.5% gate. Five clones touch a changed file. The same jscpd arguments on
+  the 3.5% gate; no clone involves a file this pass changed.
-  a copy of the base tree's `src/` and `tests/` (identical to `HEAD` for every
-  changed file) report 3.38% (2802 clones) with seven clones touching these
-  files, which include all five, so this change adds none; the bundle tests'
-  name-refusal pair and their oversized-manifest pair with
-  `environmentCapacity.test.ts` are gone.
 - `npm run scan:todos`: passed, no markers.
 - `npm run docs:check`: the built documentation checks (6 tasks) and the README
   consistency conformance test (5 tests) passed.
-- `npx vitest run --project domain --project services`: 347 files, 5946 tests
+- `npx vitest run --project domain --project services`: 347 files, 5948 tests
   passed, 5 pre-existing skips.
-- `npm run build:cached`, then `npm run test:focused --` the serve CLI, MCP and
+- `npm run build:cached` (the built output already included this pass's
-  process boundaries, the export CLI boundary, the CLI startup imports, the
+  sources), then `npm run test:focused --` the serve CLI, MCP and process
-  agentenv verifier identity boundary, the bundle package pin conformance test,
+  boundaries, the export CLI boundary, the CLI startup imports, the agentenv
-  the CI change scope boundary, the analysis-view CLI and MCP boundaries, and
+  verifier identity boundary, the bundle package pin conformance test, the CI
-  the `serveEnvironment`, `serveEnvironmentState`, `environmentCapacity` and
+  change scope boundary, the analysis-view CLI and MCP boundaries, and the
-  `stateIntake` adapters tests: 14 files, 329 tests passed.
+  `serveEnvironment`, `serveEnvironmentState`, `environmentCapacity` and
-- After this file was written: `npm run format`, `npm run format:check` and
+  `stateIntake` adapters tests: 14 files, 335 tests passed.
-  `npm run docs:check` passed again.
+- After this file was written: `npm run format`, `npm run format:check`,
+  `npm run scan:todos` and `npm run docs:check` passed again.
 
-A state catalog over the discovery budget, through the built CLI: a capture of
+The review fixes are covered by these tests, all passing in the runs above:
-380 distinct `GET` paths projected with no template rules gives a 380-tool
-read-only manifest whose replay catalog fits the reservation and whose state
-catalog does not (300 to 360 such tools fit both, 380 to 460 only replay).
-`environment-serve` with no flag logged the warning
-`"code":"resource_constraint"` before binding, served 380 tools and stopped
-with `state: {"served": false, "reason": "unavailable", "code":
-"resource_constraint", …}`; with `--replay-only` it served the same 380 tools
-and reported `replay-only`. `environment-export` of the same pair wrote a
-bundle with `tool_count` 380, that `state`, `TOOL_COUNT = 380`, a replay check
-and a README naming the code. The same export with `--name con` was refused
-with `invalid_request` and created nothing.
 
-`--replay-only` against main: a script loaded main's (`3fca2e0c`) sources for
+- Response budget: a batch of four `data/get`s over a loaded state of about a
-the Grafana-shaped fixture in-process and compared them with the built CLI
+  third of the budget keeps two results, answers the other two with `-32000`
-serving the same files with `--replay-only` and `ENVIRONMENT_NAME=renamed`. The
+  and `max_response_bytes`, keeps the trailing reset's `{}`, and its body stays
-card was byte-identical (11,131 bytes; its SHA-256, computed from main's
+  within the budget.
-sources, is the one the mcp-boundary test now pins), `tools/list` matched
+- Batch order: a batch `[data/add (gated download), data/get]` with a
-main's 19 tool names, descriptions and input schemas, five tool calls
+  `data/reset` from another request sent while the download is held reports
-(`get_users_by_id` with `2` and `999`, `get_annotations`,
+  the loaded state from its `data/get`, and the reset applies last.
-`get_folders_by_uid`, `post_folders`) returned main's replay results, and
+- Cancellation: aborting the data plane's stop signal ends a stalled download
-`data/get` answered `-32601`.
+  (its fetcher's signal is aborted) and answers the queued reset as
+  `cancelled` without running it; a call made after the stop does not run; a
+  caller whose request signal aborts gets `cancelled` and the state is
+  unchanged. Over loopback HTTP, `close()` of a served state returned while a
+  download that never answers was in progress, and that download's signal was
+  aborted.
+- File hosts: `file://remote.example.test/...` and
+  `file:////remote.example.test/...` are refused with the network-share
+  message and no failure category; a `file://localhost/...` uri loads.
+- Loaded identity: `view_fields` setting `id` to `3`, `"3"` or `null` on user 2
+  is refused at `/rea_state/entities/7/view_fields/0/fields`; `"2"` loads.
+- List required fields: a folder without `title` makes `get_folders` a tool
+  error naming the incomplete entity.
 
-Startup, measured by starting the built CLI, timing until the listener
+That each test fails without its fix is source-derived; the fixes were not
-announcement and reading `process.resourceUsage()` at exit, with a load average
+reverted to show it.
-of 6 to 7 on a shared host. Medians of six interleaved runs:
 
-| Capture                           | Mode                   | Listen ms | CPU ms | Peak RSS MiB |
+Not covered here: a Windows host (the UNC refusal ran on macOS, where
-| --------------------------------- | ---------------------- | --------- | ------ | ------------ |
+`file:////host/share` converts to `//host/share`; the host check runs before
-| Grafana fixture (25 KB)           | `--replay-only`        | 542       | 819    | 265          |
+any path conversion on every platform), a real `https://` download (tests
-| Grafana fixture (25 KB)           | default, derived state | 546       | 822    | 266          |
+inject the fetcher), the memory and time of a budgeted batch (not measured),
-| Synthetic, 1,530 records (5.2 MB) | `--replay-only`        | 610       | 934    | 277          |
+agent token usage, a real AgentEnv framework run, a real-world capture, and
-| Synthetic, 1,530 records (5.2 MB) | default, derived state | 789       | 1234   | 303          |
+write effects, which are a later increment.
-
-Deriving the state costs about 180 ms, 300 ms CPU and 26 MiB peak RSS on the
-5.2 MB capture and is within noise on the fixture. The state catalog check runs
-before deriving and takes about 1.4 ms in-process for the 350-tool manifest of
-the services test, which it refuses without deriving.
-
-Not covered here: agent token usage (no agent task was run), a real AgentEnv
-framework run against a bundle or a state-served environment
-(`npm run verify:agentenv:environment` needs Docker and the pinned framework),
-a real-world capture, a real `https://` download (tests inject the fetcher or
-stub `fetch`), and write effects, which are a later increment.

src/application/environment/environmentState.test.ts

@@ -422,6 +422,29 @@ describe("rea-state/v1 validation", () => {
     expect(problem.message).not.toContain("never echoed");
   });
 
+  it("refuses view values that give an entity another identity, and accepts its own id spelled as either scalar", () => {
+    const index = state.entities.findIndex(
+      (entity) => entity.type === "/api/users" && entity.id === "2",
+    );
+    const withView = (fields: Record<string, unknown>) =>
+      documentWith((document) => {
+        document.rea_state.entities[index]!.view_fields = [
+          { tool: "get_users_by_id", fields },
+        ];
+      });
+    for (const id of [3, "3", null])
+      expect(firstProblem(withView({ id, login: "other" }))).toEqual({
+        path: `rea_state/entities/${String(index)}/view_fields/0/fields`,
+        message:
+          "sets the type's id field to a value other than the entity's id",
+      });
+    const store = new EnvironmentStateStore(state);
+    expect(store.replace(withView({ id: "2", login: "other" })).ok).toBe(true);
+    expect(store.entity("/api/users", "/api/users", "2")?.view_fields).toEqual([
+      { tool: "get_users_by_id", fields: { id: "2", login: "other" } },
+    ]);
+  });
+
   it("refuses values parsing would change at the member holding them, without naming caller data", () => {
     const marker = Symbol("not exact JSON");
     const problem = firstProblem(

src/application/environment/environmentStateAnswer.test.ts

@@ -265,6 +265,29 @@ describe("list answers", () => {
     });
   });
 
+  it("refuses a list whose member lacks a field every recording carried, never answering a partial body", () => {
+    const serving = servingOf();
+    load(serving, (entities) =>
+      entities.map((entity) => {
+        if (entity.type !== "/api/folders" || entity.id !== "fg1jo91e57e2ob")
+          return entity;
+        const { title: _title, ...fields } = entity.fields as Record<
+          string,
+          unknown
+        >;
+        return { ...entity, fields };
+      }),
+    );
+    expect(
+      refusal(serving, "get_folders", { limit: "100", page: "1" }),
+    ).toMatchObject({
+      code: "invalid_request",
+      message: expect.stringContaining(
+        "holds no complete /api/folders entity fg1jo91e57e2ob in /api/folders: it lacks fields this tool's responses always carried: title",
+      ),
+    });
+  });
+
   it("answers an unrecorded request line from its recording", () => {
     const serving = servingOf();
     load(serving, () => []);

src/application/environment/environmentStateAnswer.ts

@@ -7,6 +7,7 @@ import {
   replaceAt,
   scalarText,
   segmentsOf,
+  type JsonObject,
   type RecordedViewReason,
   type StateEntity,
   type StateItemView,
@@ -86,6 +87,27 @@ const refused = (
     { path: ["tool"], reason: "invalid_value", expected: tool.name },
   ]);
 
+/** One entity as an item or list view shows it, or a refusal when it lacks a field the view's every recording carried; a partial body is never answered. */
+const shownEntity = (
+  tool: EnvironmentManifestTool,
+  view: StateItemView | StateListView,
+  entity: StateEntity,
+  volatile: ReadonlySet<string>,
+): Result<JsonObject, AnalysisError> => {
+  const shown = renderStateEntity(entity, view, volatile);
+  const missing = view.required_fields.filter(
+    (name) => !Object.hasOwn(shown, name),
+  );
+  return missing.length === 0
+    ? ok(shown)
+    : err(
+        refused(
+          tool,
+          `The state holds no complete ${view.type} entity ${entity.id} in ${entity.collection}: it lacks fields this tool's responses always carried: ${missing.join(", ")}.`,
+        ),
+      );
+};
+
 /** The recorded answer, labelled with why the state did not produce it. */
 const fromRecording = (
   source: EnvironmentSource,
@@ -229,18 +251,13 @@ const answerItem = (
         `The state holds no ${view.type} entity ${call.id} in /${call.segments.map(({ text }) => text).join("/")}.`,
       ),
     );
-  const volatile = new Set(store.type(view.type)?.volatile_fields);
+  const shown = shownEntity(
-  const shown = renderStateEntity(entity, view, volatile);
+    tool,
-  const missing = view.required_fields.filter(
+    view,
-    (name) => !Object.hasOwn(shown, name),
+    entity,
+    new Set(store.type(view.type)?.volatile_fields),
   );
-  if (missing.length > 0)
+  if (!shown.ok) return shown;
-    return err(
-      refused(
-        tool,
-        `The state's ${view.type} entity ${entity.id} lacks fields this tool's responses always carried: ${missing.join(", ")}.`,
-      ),
-    );
   const base = source.exchanges.get(view.envelope_ordinal);
   const envelope = base === undefined ? undefined : recordedJson(base);
   if (base === undefined || envelope === undefined)
@@ -253,7 +270,7 @@ const answerItem = (
   return stateAnswer(
     serving,
     { exchange: base, exact },
-    replaceAt(envelope, view.pointer, shown),
+    replaceAt(envelope, view.pointer, shown.value),
     {
       source: "state",
       view: "item",
@@ -315,9 +332,12 @@ const answerList = (
           return entity === undefined ? [] : [entity];
         });
   const volatile = new Set(type?.volatile_fields);
-  const items = members.map((entity) =>
+  const items: JsonObject[] = [];
-    renderStateEntity(entity, view, volatile),
+  for (const entity of members) {
-  );
+    const shown = shownEntity(tool, view, entity, volatile);
+    if (!shown.ok) return shown;
+    items.push(shown.value);
+  }
   return stateAnswer(
     serving,
     { exchange: base, exact: true },

src/application/environment/environmentStateDocument.ts

@@ -179,6 +179,39 @@ interface EntityCheck {
 const entityKey = (entity: Pick<StateEntity, "type" | "collection" | "id">) =>
   compositeKey([entity.type, entity.collection, entity.id]);
 
+/** The first problem with an entity's view values: each names an item or list view of its type once and keeps its identity. */
+const viewFieldsProblem = (
+  state: EnvironmentState,
+  type: StateType,
+  entity: ParsedEntity,
+  path: Path,
+): AnalysisInputIssue | undefined => {
+  const tools = new Set<string>();
+  for (const [index, view] of (entity.view_fields ?? []).entries()) {
+    const plan = state.views.get(view.tool);
+    if (
+      (plan?.shape !== "item" && plan?.shape !== "list") ||
+      plan.type !== entity.type ||
+      tools.has(view.tool)
+    )
+      return problem(
+        [...path, "view_fields", index, "tool"],
+        "is not an item or list view of the entity's type, or repeats one",
+      );
+    tools.add(view.tool);
+    // Views render these values over the entity's fields, so one carrying another id would answer under this entity's key with another entity's identity.
+    if (
+      Object.hasOwn(view.fields, type.id_field) &&
+      scalarText(view.fields[type.id_field]) !== entity.id
+    )
+      return problem(
+        [...path, "view_fields", index, "fields"],
+        `sets the type's ${type.id_field} field to a value other than the entity's id`,
+      );
+  }
+  return undefined;
+};
+
 /** The first problem with one entity, given the keys already seen. */
 const entityProblem = (
   { state, initial, seen }: EntityCheck,
@@ -210,21 +243,7 @@ const entityProblem = (
       [...path, "record_ordinals"],
       "cites records the served Evidence does not cite for this entity",
     );
-  const tools = new Set<string>();
+  return viewFieldsProblem(state, type, entity, path);
-  for (const [index, view] of (entity.view_fields ?? []).entries()) {
-    const plan = state.views.get(view.tool);
-    if (
-      (plan?.shape !== "item" && plan?.shape !== "list") ||
-      plan.type !== entity.type ||
-      tools.has(view.tool)
-    )
-      return problem(
-        [...path, "view_fields", index, "tool"],
-        "is not an item or list view of the entity's type, or repeats one",
-      );
-    tools.add(view.tool);
-  }
-  return undefined;
 };
 
 /** The first problem with the documents: each names a document view once, at the initial document's path. */

src/server/environment/environmentDataPlane.ts

@@ -1,6 +1,7 @@
 import type { EnvironmentStateStore } from "../../application/environment/environmentStateStore.js";
 import type { AnalysisInputError } from "../../domain/analysisErrorCore.js";
 import {
+  CANCELLED,
   DEFAULT_STATE_INTAKE,
   invalidParams,
   partDocument,
@@ -13,10 +14,11 @@ export type RpcOutcome =
   | { readonly result: unknown }
   | { readonly error: RpcError };
 
-/** The data-plane methods one environment serves, by JSON-RPC method name. */
+/** The data-plane methods one environment serves, by JSON-RPC method name; a call is queued when made, and `signal` cancels it. */
 export type RpcMethods = (
   method: string,
   params: unknown,
+  signal?: AbortSignal,
 ) => Promise<RpcOutcome | undefined>;
 
 const isRecord = (value: unknown): value is Readonly<Record<string, unknown>> =>
@@ -37,57 +39,76 @@ const documentError = (error: AnalysisInputError): RpcError => {
   );
 };
 
-/** `data/add`: exactly one part, whose document replaces the whole state only when it is valid. */
+/** What one queued call runs with: the live state, how documents arrive and what cancels it. */
+interface CallContext {
+  readonly store: EnvironmentStateStore;
+  readonly intake: StateIntake;
+  readonly cancel: AbortSignal;
+}
+
+/** `data/add`: exactly one part, whose document replaces the whole state only when it is valid and the call was not cancelled. */
 const addState = async (
-  store: EnvironmentStateStore,
+  { store, intake, cancel }: CallContext,
   params: unknown,
-  intake: StateIntake,
 ): Promise<RpcOutcome> => {
   const parts = isRecord(params) ? params.parts : undefined;
   if (!Array.isArray(parts) || parts.length !== 1)
     return { error: invalidParams("data/add takes exactly one part") };
-  const document = await partDocument(parts[0], intake);
+  const document = await partDocument(parts[0], intake, cancel);
   if (!document.ok) return { error: document.error };
+  // A file read is not interruptible, so a cancellation during it is honoured here, before the state changes.
+  if (cancel.aborted) return { error: CANCELLED };
   const replaced = store.replace(document.value);
   return replaced.ok
     ? { result: {} }
     : { error: documentError(replaced.error) };
 };
 
-/** One data-plane method over the live state, or undefined when the method is not served. */
+type Method = (
-const runMethod = (
+  context: CallContext,
-  store: EnvironmentStateStore,
-  intake: StateIntake,
-  method: string,
   params: unknown,
-): Promise<RpcOutcome> | undefined => {
+) => RpcOutcome | Promise<RpcOutcome>;
-  switch (method) {
+
-    case "data/reset":
+const METHODS: ReadonlyMap<string, Method> = new Map<string, Method>([
+  [
+    "data/reset",
+    ({ store }) => {
       store.reset();
-      return Promise.resolve({ result: {} });
+      return { result: {} };
-    case "data/add":
+    },
-      return addState(store, params, intake);
+  ],
-    case "data/get":
+  ["data/add", addState],
-      return Promise.resolve({
+  [
-        result: { parts: [{ kind: "data", data: store.snapshot() }] },
+    "data/get",
-      });
+    ({ store }) => ({
-    default:
+      result: { parts: [{ kind: "data", data: store.snapshot() }] },
-      return undefined;
+    }),
-  }
+  ],
-};
+]);
 
 /**
  * `data/reset`, `data/add` and `data/get` over one live state. Calls run one
- * at a time in arrival order across requests, so a slow load never overwrites
+ * at a time in the order they are made, across requests, so a slow load never
- * a later reset and at most one loaded document is parsed at once.
+ * overwrites a later reset and at most one loaded document is parsed at once.
+ * Once `stopped` or a call's own signal aborts, its download ends and a call
+ * not yet run is answered as cancelled.
  */
 export const stateDataPlane = (
   store: EnvironmentStateStore,
   intake: StateIntake = DEFAULT_STATE_INTAKE,
+  stopped: AbortSignal = new AbortController().signal,
 ): RpcMethods => {
   let previous: Promise<unknown> = Promise.resolve();
-  return (method, params) => {
+  return (method, params, signal) => {
-    const next = previous.then(() => runMethod(store, intake, method, params));
+    const run = METHODS.get(method);
+    if (run === undefined) return Promise.resolve(undefined);
+    const cancel =
+      signal === undefined ? stopped : AbortSignal.any([stopped, signal]);
+    const next = previous.then(() =>
+      cancel.aborted
+        ? { error: CANCELLED }
+        : run({ store, intake, cancel }, params),
+    );
     previous = next.catch(() => undefined);
     return next;
   };

src/server/environment/serveEnvironment.ts

@@ -40,14 +40,12 @@ import {
   UNTRUSTED_CONTENT_INSTRUCTIONS,
 } from "../toolResultEncoding.js";
 import { RequestTrackingMcpServer } from "../RequestTrackingMcpServer.js";
-import {
+import { stateDataPlane, type RpcMethods } from "./environmentDataPlane.js";
-  stateDataPlane,
-  type RpcMethods,
-  type RpcOutcome,
-} from "./environmentDataPlane.js";
 import {
   DEFAULT_STATE_INTAKE,
   exactJsonReviver,
+  SERVER_ERROR,
+  STATE_INTAKE_LIMIT_BYTES,
   type RpcError,
   type StateFetcher,
 } from "./stateIntake.js";
@@ -139,10 +137,14 @@ type RpcReply = {
   readonly id: string | number | null;
 } & ({ readonly result: unknown } | { readonly error: RpcError });
 
-/** The reply to one JSON-RPC message, or none for a notification; a method no table serves is not found. */
+/** At most this many UTF-8 bytes of replies in one `/agentenv` response: twice the largest document `data/add` loads, so a `data/get` of a loaded state fits but a batch cannot multiply it. */
-const rpcReply = async (
+export const MAX_RPC_RESPONSE_BYTES = 2 * STATE_INTAKE_LIMIT_BYTES;
+
+/** Start one JSON-RPC message now, so its call is queued at arrival; resolves to its reply, or none for a notification. */
+const startRpc = (
   message: unknown,
   methods: RpcMethods | undefined,
+  signal: AbortSignal,
 ): Promise<RpcReply | undefined> => {
   const id =
     isRecord(message) &&
@@ -150,26 +152,96 @@ const rpcReply = async (
       ? message.id
       : null;
   if (!isRpcMessage(message))
-    return {
+    return Promise.resolve({
       jsonrpc: "2.0",
       id,
       error: { code: INVALID_REQUEST, message: "invalid request" },
-    };
+    });
   const method = String(message.method);
   // A notification still runs its method; it only gets no reply.
-  const outcome: RpcOutcome | undefined =
+  const outcome =
-    methods === undefined ? undefined : await methods(method, message.params);
+    methods === undefined
-  if (!("id" in message)) return undefined;
+      ? Promise.resolve(undefined)
-  return {
+      : methods(method, message.params, signal);
-    jsonrpc: "2.0",
+  const replies = "id" in message;
-    id,
+  return outcome.then((settled): RpcReply | undefined =>
-    ...(outcome ?? {
+    replies
-      error: { code: METHOD_NOT_FOUND, message: `method not found: ${method}` },
+      ? {
-    }),
+          jsonrpc: "2.0",
+          id,
+          ...(settled ?? {
+            error: {
+              code: METHOD_NOT_FOUND,
+              message: `method not found: ${method}`,
+            },
+          }),
+        }
+      : undefined,
+  );
+};
+
+const overBudget = (replyBytes: number): RpcError => ({
+  code: SERVER_ERROR,
+  message: `the reply would take this response past its ${String(MAX_RPC_RESPONSE_BYTES)}-byte budget; the call ran, so repeat a read in another request`,
+  data: { max_response_bytes: MAX_RPC_RESPONSE_BYTES, reply_bytes: replyBytes },
+});
+
+/** Serialize a response's replies in order within its budget; one that would pass it becomes an error stating it, unless that error is no smaller. */
+const replyEncoder = (): ((reply: RpcReply) => string) => {
+  let used = 0;
+  return (reply) => {
+    const text = JSON.stringify(reply);
+    const bytes = Buffer.byteLength(text, "utf8");
+    if (used + bytes <= MAX_RPC_RESPONSE_BYTES) {
+      used += bytes;
+      return text;
+    }
+    const refused = JSON.stringify({
+      jsonrpc: "2.0",
+      id: reply.id,
+      error: overBudget(bytes),
+    });
+    return Buffer.byteLength(refused, "utf8") < bytes ? refused : text;
   };
 };
 
-/** Answer the data plane within the body bound; without methods every request is not found. */
+const jsonResponse = (text: string, status: number): Response =>
+  new Response(text, {
+    status,
+    headers: { "content-type": "application/json" },
+  });
+
+/** Answer a batch: every message is queued at arrival, and replies are encoded in order as each completes. */
+const answerBatch = async (
+  messages: readonly unknown[],
+  methods: RpcMethods | undefined,
+  signal: AbortSignal,
+): Promise<Response> => {
+  if (messages.length === 0)
+    return jsonRpcErrorResponse(400, INVALID_REQUEST, "invalid request");
+  if (messages.length > MAX_RPC_BATCH_MESSAGES)
+    return jsonRpcErrorResponse(
+      400,
+      INVALID_REQUEST,
+      `batch exceeds ${String(MAX_RPC_BATCH_MESSAGES)} messages`,
+    );
+  // Starting every message now keeps the batch consecutive in the arrival-order
+  // queue, so no other request's call lands between its messages.
+  const pending = messages.map((message) => startRpc(message, methods, signal));
+  // The loop below reports the first failure; later ones are not left unhandled.
+  for (const reply of pending) void reply.catch(() => undefined);
+  const encode = replyEncoder();
+  const texts: string[] = [];
+  for (const next of pending) {
+    const reply = await next;
+    if (reply !== undefined) texts.push(encode(reply));
+  }
+  return texts.length === 0
+    ? new Response(null, { status: 202 })
+    : jsonResponse(`[${texts.join(",")}]`, 200);
+};
+
+/** Answer the data plane within the body and response bounds; without methods every request is not found. */
 const answerRpc = async (
   request: Request,
   methods: RpcMethods | undefined,
@@ -184,33 +256,16 @@ const answerRpc = async (
   // Values parsing would alter become markers: such an id is refused rather than echoed as another number, and a state document carrying one is refused.
   const parsed = safeParseJson(read.text, exactJsonReviver);
   if (!parsed.ok) return jsonRpcErrorResponse(400, PARSE_ERROR, "parse error");
-  // A batch is answered message by message; notifications get no reply at all.
+  // The request's signal cancels its calls when the client leaves.
-  if (Array.isArray(parsed.value)) {
+  if (Array.isArray(parsed.value))
-    if (parsed.value.length === 0)
+    return answerBatch(parsed.value, methods, request.signal);
-      return jsonRpcErrorResponse(400, INVALID_REQUEST, "invalid request");
+  const reply = await startRpc(parsed.value, methods, request.signal);
-    if (parsed.value.length > MAX_RPC_BATCH_MESSAGES)
-      return jsonRpcErrorResponse(
-        400,
-        INVALID_REQUEST,
-        `batch exceeds ${String(MAX_RPC_BATCH_MESSAGES)} messages`,
-      );
-    // Messages run in order, so a batch's later reads see its earlier writes.
-    const replies: RpcReply[] = [];
-    for (const message of parsed.value) {
-      const reply = await rpcReply(message, methods);
-      if (reply !== undefined) replies.push(reply);
-    }
-    return replies.length === 0
-      ? new Response(null, { status: 202 })
-      : Response.json(replies);
-  }
-  const reply = await rpcReply(parsed.value, methods);
   return reply === undefined
     ? new Response(null, { status: 202 })
-    : Response.json(reply, {
+    : jsonResponse(
-        status:
+        replyEncoder()(reply),
-          "error" in reply && reply.error.code === INVALID_REQUEST ? 400 : 200,
+        "error" in reply && reply.error.code === INVALID_REQUEST ? 400 : 200,
-      });
+      );
 };
 
 /** The plain routes an agent environment serves beside `/mcp`; data-plane methods only when given. */
@@ -267,6 +322,11 @@ export const serveEnvironment = async (
   const { tools, dataPlane } = catalog.value;
   const card = environmentCard(name, tools, dataPlane);
   const answer = environmentToolAnswer(source, store);
+  // Stopping ends the running download and skips queued data-plane calls, so shutdown never waits on them.
+  const stopping = new AbortController();
+  const stop = (): void => {
+    stopping.abort(new Error("The agent environment is stopping"));
+  };
   const served = await serveStreamableHttp({
     host,
     port,
@@ -276,12 +336,16 @@ export const serveEnvironment = async (
       card,
       store === undefined
         ? undefined
-        : stateDataPlane(store, {
+        : stateDataPlane(
-            ...DEFAULT_STATE_INTAKE,
+            store,
-            ...(options.fetchState === undefined
+            {
-              ? {}
+              ...DEFAULT_STATE_INTAKE,
-              : { fetchState: options.fetchState }),
+              ...(options.fetchState === undefined
-          }),
+                ? {}
+                : { fetchState: options.fetchState }),
+            },
+            stopping.signal,
+          ),
     ),
     ...(logger === undefined ? {} : { logger }),
     onerror: (error) => {
@@ -292,5 +356,16 @@ export const serveEnvironment = async (
       options.onerror?.(error);
     },
   });
-  return ok({ ...served, card });
+  return ok({
+    ...served,
+    card,
+    prepareShutdown: () => {
+      stop();
+      return served.prepareShutdown();
+    },
+    close: () => {
+      stop();
+      return served.close();
+    },
+  });
 };

src/server/environment/serveEnvironmentState.test.ts

@@ -24,8 +24,10 @@ import {
   ENVIRONMENT_CARD_PATH,
   ENVIRONMENT_RPC_PATH,
   environmentRoutes,
+  MAX_RPC_RESPONSE_BYTES,
   serveEnvironment,
   type ServedEnvironment,
+  type ServeEnvironmentOptions,
 } from "./serveEnvironment.js";
 import {
   DEFAULT_STATE_INTAKE,
@@ -50,19 +52,23 @@ const renamedDocument = () => {
   };
 };
 
-/** A data plane over a fresh store, posting raw bodies to its route. */
+/** A data plane over a fresh store, posting raw bodies to its route; `stopped` is the server's stop signal. */
-const dataPlane = (intake: Partial<StateIntake> = {}) => {
+const dataPlane = (
+  intake: Partial<StateIntake> = {},
+  stopped?: AbortSignal,
+) => {
   const store = new EnvironmentStateStore(state);
   const route = environmentRoutes(
     environmentCard("grafana-example", [], "state"),
-    stateDataPlane(store, { ...DEFAULT_STATE_INTAKE, ...intake }),
+    stateDataPlane(store, { ...DEFAULT_STATE_INTAKE, ...intake }, stopped),
   ).find((entry) => entry.path === ENVIRONMENT_RPC_PATH);
   if (route === undefined) throw new Error("No data-plane route");
-  const post = async (body: unknown) => {
+  const post = async (body: unknown, signal?: AbortSignal) => {
     const response = await route.handle(
       new Request(`http://127.0.0.1${ENVIRONMENT_RPC_PATH}`, {
         method: "POST",
         body: typeof body === "string" ? body : JSON.stringify(body),
+        ...(signal === undefined ? {} : { signal }),
       }),
     );
     return {
@@ -86,6 +92,32 @@ const dataPlane = (intake: Partial<StateIntake> = {}) => {
 const loginOf = (store: EnvironmentStateStore) =>
   store.entity("/api/users", "/api/users", "2")?.fields.login;
 
+/** A `data/add` message loading the document an https uri names. */
+const addByUri = (
+  id: number,
+  uri = "https://state.example.test/state.json",
+) => ({
+  jsonrpc: "2.0",
+  id,
+  method: "data/add",
+  params: { parts: [{ kind: "file", file: { uri } }] },
+});
+
+/** A fetcher that reports each call and the signal it was given, then answers only when `answer` settles. */
+const heldFetcher = (answer: () => Promise<Response>) => {
+  const signals: AbortSignal[] = [];
+  let started: () => void = () => undefined;
+  const fetching = new Promise<void>((resolve) => {
+    started = resolve;
+  });
+  const fetchState = (_: URL, signal: AbortSignal) => {
+    signals.push(signal);
+    started();
+    return answer();
+  };
+  return { signals, fetching, fetchState };
+};
+
 describe("state data plane", () => {
   it("answers data/get, data/add and data/reset in the AgentEnv shapes", async () => {
     const plane = dataPlane();
@@ -271,6 +303,35 @@ describe("state data plane intake", () => {
     await expect(add()).resolves.toMatchObject({ error: { code: -32602 } });
   });
 
+  it("opens file uris only on this host, refusing another host or a network share before opening anything", async () => {
+    const plane = dataPlane();
+    for (const uri of [
+      "file://remote.example.test/share/state.json",
+      "file:////remote.example.test/share/state.json",
+    ]) {
+      const reply = await plane.add({ kind: "file", file: { uri } });
+      expect(reply).toEqual({
+        jsonrpc: "2.0",
+        id: 1,
+        error: {
+          code: -32602,
+          message:
+            "invalid params: a file:// uri names a path on this host, not a network share",
+        },
+      });
+    }
+    const root = await createTestTempDirectory("rea-environment-state-host-");
+    const path = join(root, "state.json");
+    await writeFile(path, JSON.stringify(renamedDocument()));
+    await expect(
+      plane.add({
+        kind: "file",
+        file: { uri: `file://localhost${pathToFileURL(path).pathname}` },
+      }),
+    ).resolves.toMatchObject({ result: {} });
+    expect(loginOf(plane.store)).toBe("renamed");
+  });
+
   it("runs batches in order and runs notifications without replying", async () => {
     const plane = dataPlane();
     const batch = await plane.post([
@@ -349,6 +410,162 @@ describe("state data plane timing and order", () => {
     // The reset arrived after the add, so it wins even though the add finished loading later.
     expect(loginOf(plane.store)).toBe("probe");
   });
+
+  it("queues every message of a batch at arrival, so a later request's reset never lands between them", async () => {
+    let release: () => void = () => undefined;
+    const gate = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    const held = heldFetcher(async () => {
+      await gate;
+      return new Response(JSON.stringify(renamedDocument()));
+    });
+    const plane = dataPlane({ fetchState: held.fetchState });
+    const batch = plane.post([
+      addByUri(1),
+      { jsonrpc: "2.0", id: 2, method: "data/get" },
+    ]);
+    await held.fetching;
+    const reset = plane.call("data/reset");
+    release();
+    const replies = (await batch).json;
+    expect(replies).toMatchObject([
+      { id: 1, result: {} },
+      { id: 2, result: { parts: [{ kind: "data" }] } },
+    ]);
+    // The get arrived with the add, before the reset, so it reports the loaded state.
+    expect(JSON.stringify(replies)).toContain('"login":"renamed"');
+    await expect(reset).resolves.toMatchObject({ result: {} });
+    expect(loginOf(plane.store)).toBe("probe");
+  });
+});
+
+describe("state data plane cancellation", () => {
+  /** A download that never answers on its own and outlives any test. */
+  const stalled = () =>
+    heldFetcher(() => new Promise<Response>(() => undefined));
+
+  it("ends the running download and skips queued calls once the server stops", async () => {
+    const stopping = new AbortController();
+    const held = stalled();
+    const plane = dataPlane(
+      { fetchState: held.fetchState, timeoutMs: 600_000 },
+      stopping.signal,
+    );
+    await plane.add({ kind: "data", data: renamedDocument() });
+    const batch = plane.post([
+      addByUri(1),
+      { jsonrpc: "2.0", id: 2, method: "data/reset" },
+    ]);
+    await held.fetching;
+    stopping.abort();
+    expect((await batch).json).toEqual([
+      {
+        jsonrpc: "2.0",
+        id: 1,
+        error: {
+          code: -32000,
+          message: "the state document could not be downloaded",
+          data: { reason: "cancelled" },
+        },
+      },
+      {
+        jsonrpc: "2.0",
+        id: 2,
+        error: {
+          code: -32000,
+          message: "the call was cancelled and left the state unchanged",
+          data: { reason: "cancelled" },
+        },
+      },
+    ]);
+    expect(held.signals.map((signal) => signal.aborted)).toEqual([true]);
+    // The queued reset never ran, and nothing runs after the stop.
+    expect(loginOf(plane.store)).toBe("renamed");
+    await expect(plane.call("data/reset")).resolves.toMatchObject({
+      error: { data: { reason: "cancelled" } },
+    });
+    expect(loginOf(plane.store)).toBe("renamed");
+  });
+
+  it("ends a download whose caller left, leaving the state unchanged", async () => {
+    const held = stalled();
+    const plane = dataPlane({
+      fetchState: held.fetchState,
+      timeoutMs: 600_000,
+    });
+    const leaving = new AbortController();
+    const add = plane.post(addByUri(1), leaving.signal);
+    await held.fetching;
+    leaving.abort();
+    expect((await add).json).toMatchObject({
+      error: { code: -32000, data: { reason: "cancelled" } },
+    });
+    expect(held.signals[0]?.aborted).toBe(true);
+    expect(loginOf(plane.store)).toBe("probe");
+    // Other callers' calls still run.
+    await expect(
+      plane.add({ kind: "data", data: renamedDocument() }),
+    ).resolves.toMatchObject({ result: {} });
+  });
+});
+
+describe("state data plane response budget", () => {
+  it("answers replies past the response budget with an error stating it instead of building the body", async () => {
+    const document = renamedDocument();
+    const large = {
+      rea_state: {
+        ...document.rea_state,
+        entities: document.rea_state.entities.map((entity) =>
+          entity.type === "/api/users" && entity.id === "2"
+            ? {
+                ...entity,
+                fields: {
+                  ...entity.fields,
+                  login: "x".repeat(Math.ceil(MAX_RPC_RESPONSE_BYTES / 3)),
+                },
+              }
+            : entity,
+        ),
+      },
+    };
+    const root = await createTestTempDirectory("rea-environment-state-budget-");
+    const path = join(root, "large.json");
+    await writeFile(path, JSON.stringify(large));
+    const plane = dataPlane();
+    await expect(
+      plane.add({ kind: "file", file: { uri: pathToFileURL(path).href } }),
+    ).resolves.toMatchObject({ result: {} });
+    const get = (id: number) => ({ jsonrpc: "2.0", id, method: "data/get" });
+    const batch = await plane.post([
+      get(1),
+      get(2),
+      get(3),
+      get(4),
+      { jsonrpc: "2.0", id: 5, method: "data/reset" },
+    ]);
+    // Two reads fit; the third and fourth would pass the budget, and the reset's small reply is kept.
+    expect(batch.json).toMatchObject([
+      { id: 1, result: { parts: [{ kind: "data" }] } },
+      { id: 2, result: { parts: [{ kind: "data" }] } },
+      {
+        id: 3,
+        error: {
+          code: -32000,
+          data: {
+            max_response_bytes: MAX_RPC_RESPONSE_BYTES,
+            reply_bytes: expect.any(Number),
+          },
+        },
+      },
+      { id: 4, error: { code: -32000 } },
+      { id: 5, result: {} },
+    ]);
+    expect(Buffer.byteLength(JSON.stringify(batch.json))).toBeLessThanOrEqual(
+      MAX_RPC_RESPONSE_BYTES,
+    );
+    expect(loginOf(plane.store)).toBe("probe");
+  });
 });
 
 const served: ServedEnvironment[] = [];
@@ -359,9 +576,11 @@ afterEach(async () => {
 /** Serve a state on a loopback port until the test ends. */
 const serveState = async (
   inputs: ReturnType<typeof environmentStateServeInputs>,
+  options: Pick<ServeEnvironmentOptions, "fetchState"> = {},
 ): Promise<ServedEnvironment> => {
   const entry = await serveEnvironment({
     ...inputs,
+    ...options,
     host: "127.0.0.1",
     port: 0,
   });
@@ -446,6 +665,29 @@ describe("serveEnvironment with a state", () => {
     });
   });
 
+  it("stops without waiting for a stalled download", async () => {
+    const held = heldFetcher(() => new Promise<Response>(() => undefined));
+    const environment = await serveState(
+      { source, state },
+      { fetchState: held.fetchState },
+    );
+    const answered = fetch(new URL(ENVIRONMENT_RPC_PATH, environment.url), {
+      method: "POST",
+      headers: { "content-type": "application/json" },
+      body: JSON.stringify(addByUri(1)),
+    }).then(
+      (response) => response.status,
+      () => "closed",
+    );
+    await held.fetching;
+    // The download's own limit is 30 s; stopping must not wait for it.
+    const started = performance.now();
+    await environment.close();
+    expect(performance.now() - started).toBeLessThan(10_000);
+    expect(held.signals[0]?.aborted).toBe(true);
+    await answered;
+  });
+
   it("serves query POSTs from their recordings and leaves state-changing writes out", async () => {
     const environment = await serveState(
       environmentStateServeInputs(

src/server/environment/stateIntake.ts

@@ -11,7 +11,7 @@ import {
 } from "../../filesystem/readStableFileBytes.js";
 import { openRegularFile } from "../../filesystem/RegularFile.js";
 
-/** Why a named file or download could not be read: a category naming neither the uri nor the content. */
+/** Why a named file or download could not be read, or a call was cancelled: a category naming neither the uri nor the content. */
 export type IntakeFailure =
   | "not-found"
   | "permission"
@@ -20,20 +20,29 @@ export type IntakeFailure =
   | "read-failed"
   | "timeout"
   | "http-status"
-  | "network";
+  | "network"
+  | "cancelled";
 
-/** A JSON-RPC error the data plane answers with; `data` locates a document problem or names an intake failure. */
+/** A JSON-RPC error the data plane answers with; `data` locates a document problem, names an intake failure or states the response budget. */
 export interface RpcError {
   readonly code: number;
   readonly message: string;
   readonly data?:
     | { readonly path: string }
-    | { readonly reason: IntakeFailure };
+    | { readonly reason: IntakeFailure }
+    | { readonly max_response_bytes: number; readonly reply_bytes: number };
 }
 
 const INVALID_PARAMS = -32602;
 /** The data plane could not complete an operation whose params were valid. */
-const SERVER_ERROR = -32000;
+export const SERVER_ERROR = -32000;
+
+/** A call the server's stop or its caller's departure ended before it could change the state. */
+export const CANCELLED: RpcError = {
+  code: SERVER_ERROR,
+  message: "the call was cancelled and left the state unchanged",
+  data: { reason: "cancelled" },
+};
 
 /** A loaded state document is at most this large, whether it arrives as bytes, a file or a download. */
 export const STATE_INTAKE_LIMIT_BYTES = 32 * 1024 * 1024;
@@ -178,14 +187,32 @@ const abortion = (signal: AbortSignal): Promise<never> => {
   return aborted;
 };
 
-/** Read a response body within the intake limit and the download's time, whether or not the fetcher ties the body to the signal. */
+/** What ends a download: its time limit or the caller's cancellation; `failure` names which, or a network failure when neither did. */
+interface DownloadBounds {
+  readonly signal: AbortSignal;
+  readonly failure: () => IntakeFailure;
+}
+
+const downloadBounds = (
+  timeoutMs: number,
+  cancel: AbortSignal,
+): DownloadBounds => {
+  const timeout = AbortSignal.timeout(timeoutMs);
+  return {
+    signal: AbortSignal.any([timeout, cancel]),
+    failure: () =>
+      cancel.aborted ? "cancelled" : timeout.aborted ? "timeout" : "network",
+  };
+};
+
+/** Read a response body within the intake limit and the download's bounds, whether or not the fetcher ties the body to the signal. */
 const readCapped = async (
   body: ReadableStream<Uint8Array> | null,
-  signal: AbortSignal,
+  bounds: DownloadBounds,
 ): Promise<Result<Buffer, RpcError>> => {
   if (body === null) return ok(Buffer.alloc(0));
   const reader = body.getReader();
-  const aborted = abortion(signal);
+  const aborted = abortion(bounds.signal);
   const chunks: Uint8Array[] = [];
   let total = 0;
   try {
@@ -201,29 +228,32 @@ const readCapped = async (
     }
   } catch {
     void reader.cancel().catch(() => undefined);
-    return err(fetchFailed(signal.aborted ? "timeout" : "network"));
+    return err(fetchFailed(bounds.failure()));
   }
 };
 
-/** Download one https document within the time and size limits. */
+/** Download one https document within the time and size limits, ending it when `cancel` aborts. */
 const downloadState = async (
   url: URL,
   intake: StateIntake,
+  cancel: AbortSignal,
 ): Promise<Result<Buffer, RpcError>> => {
-  const signal = AbortSignal.timeout(intake.timeoutMs);
+  const bounds = downloadBounds(intake.timeoutMs, cancel);
-  const fetching = Promise.resolve().then(() => intake.fetchState(url, signal));
+  const fetching = Promise.resolve().then(() =>
+    intake.fetchState(url, bounds.signal),
+  );
   let response: Response;
   try {
-    response = await Promise.race([fetching, abortion(signal)]);
+    response = await Promise.race([fetching, abortion(bounds.signal)]);
   } catch {
-    // A response that arrives after the timeout is released unread.
+    // A response that arrives after the download ended is released unread.
     void fetching
       .then(
         (late) => late.body?.cancel(),
         () => undefined,
       )
       .catch(() => undefined);
-    return err(fetchFailed(signal.aborted ? "timeout" : "network"));
+    return err(fetchFailed(bounds.failure()));
   }
   const declared = Number(response.headers.get("content-length") ?? "0");
   if (!response.ok || declared > STATE_INTAKE_LIMIT_BYTES) {
@@ -237,13 +267,33 @@ const downloadState = async (
           },
     );
   }
-  return readCapped(response.body, signal);
+  return readCapped(response.body, bounds);
 };
 
-/** The bytes a file part carries inline or names by a `file://` or `https://` uri. */
+const NETWORK_SHARE = invalidParams(
+  "a file:// uri names a path on this host, not a network share",
+);
+
+/** A path that names a host before its first segment, as `\\server\share` or `//server/share` does. */
+const UNC_PATH = /^[\\/]{2}/u;
+
+/** The local path a `file://` uri names; another host or a UNC path would open a network share, so neither is opened. */
+const localFilePath = (url: URL): Result<string, RpcError> => {
+  if (url.host !== "" && url.host !== "localhost") return err(NETWORK_SHARE);
+  let path: string;
+  try {
+    path = fileURLToPath(url);
+  } catch {
+    return err(invalidParams("a file:// uri names a local absolute path"));
+  }
+  return UNC_PATH.test(path) ? err(NETWORK_SHARE) : ok(path);
+};
+
+/** The bytes a file part carries inline or names by a `file://` or `https://` uri; `cancel` ends a download. */
 const fileBytes = async (
   file: Readonly<Record<string, unknown>>,
   intake: StateIntake,
+  cancel: AbortSignal,
 ): Promise<Result<Uint8Array, RpcError>> => {
   if (typeof file.bytes === "string" && file.uri === undefined) {
     if (!BASE64.test(file.bytes))
@@ -260,22 +310,18 @@ const fileBytes = async (
       invalidParams("a file part carries exactly one of bytes or a uri"),
     );
   const url = new URL(file.uri);
-  if (url.protocol === "https:") return downloadState(url, intake);
+  if (url.protocol === "https:") return downloadState(url, intake, cancel);
   if (url.protocol !== "file:")
     return err(invalidParams("a file uri is a file:// or https:// URL"));
-  let path: string;
+  const path = localFilePath(url);
-  try {
+  return path.ok ? readStateFile(path.value) : path;
-    path = fileURLToPath(url);
-  } catch {
-    return err(invalidParams("a file:// uri names a local absolute path"));
-  }
-  return readStateFile(path);
 };
 
-/** The state document one `data/add` part carries, parsed but not yet validated. */
+/** The state document one `data/add` part carries, parsed but not yet validated; `cancel` ends a download. */
 export const partDocument = async (
   part: unknown,
   intake: StateIntake,
+  cancel: AbortSignal,
 ): Promise<Result<unknown, RpcError>> => {
   if (isRecord(part) && part.kind === "data" && isRecord(part.data))
     return ok(part.data);
@@ -292,6 +338,6 @@ export const partDocument = async (
       mediaTypeEssence(mimeType) !== "application/json")
   )
     return err(invalidParams("a file part's mimeType is application/json"));
-  const bytes = await fileBytes(part.file, intake);
+  const bytes = await fileBytes(part.file, intake, cancel);
   return bytes.ok ? parseDocumentBytes(bytes.value) : bytes;
 };