starred/rea
1
0 Fork 0
mirror of https://github.com/morluto/rea.git synced 2026-10-21 21:32:26 -05:00
Code Issues Packages Projects Releases 10 Wiki Activity

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

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
Edgar ArakelyanandClaude Opus 5.5 committed 2026-10-20 00:11:25 -07:00
1 parent 6676582675
commit 83fcc739f7
13 files changed
+739 -232
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/agent-environments.md
+32 -9
View File
@@ -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,
so a reset sent while a download is loading applies after that load. `data/add`
Calls run one at a time in arrival order across requests. Every message of a
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
without following redirects and within 30 seconds, including reading the body;
`mimeType`, when given, is `application/json`. A loaded document is at most 32
either a `file://` URL of a local regular file, whose host is empty or
`localhost`, or an `https://` URL fetched without following redirects and
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
+7 -5
View File
@@ -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
timeout, coarse failure categories) and `environmentDataPlane.ts` for the
three methods, run one at a time in arrival order; let `serveEnvironment.ts`
serve the shared catalog and answer function and run data-plane methods,
batches in order and notifications.
6. Add `stateIntake.ts` for parts, files on this host and downloads
(injectable fetcher and timeout, cancellation, coarse failure categories)
and `environmentDataPlane.ts` for the three methods, run one at a time in
arrival order and cancelled by the server's stop or the caller's
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
+38 -15
View File
@@ -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
write or with other filters were never checked and answer from recordings.
otherwise the recorded members the state still holds. A member lacking one
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
empty `parts`; a 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, batches in order, and
[{kind: "data", data: rea_state}]}`). A `file://` uri whose host is neither
empty nor `localhost`, or whose path is a UNC path, is `-32602` before
anything is opened. Anything else is `-32602`, including empty `parts`; a
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
and malformed parts and invalid documents, inexact numbers, bytes, `file://`
and its failure categories, injected https downloads (status, declared and
streamed size, the timeout whether or not the fetcher honours its signal), the
default fetch's refused redirects, arrival-order application across requests,
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
the body limit, a view value that changes an entity's identity, and a list
member missing a required field. Adapters tests cover the data-plane shapes,
`-32602` for empty and malformed parts and invalid documents, inexact numbers,
bytes, `file://` and its failure categories, `file://` uris naming another
host or a UNC path, injected https downloads (status, declared and streamed
size, the timeout whether or not the fetcher honours its signal), the default
fetch's refused redirects, arrival-order application across requests, a batch
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
+5
View File
@@ -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
+48 -63
View File
@@ -1,78 +1,63 @@
# Verification
Verified base: e96484b10a1ae3dad8541d748599da32b447495e (main with #5319 and
#5349 merged into this branch), with the changes of this pass uncommitted.
Verified base: 6676582675e08574146a812ae58cff67e2e53282 (this branch's head),
with the review fixes of this pass uncommitted.
Executed on the implementation worktree with Node v22.23.1 (the repository pins
24.18.0 in `.nvmrc`, which was not installed). Every check below ran as written
after the last source change, in this order.
Executed on the implementation worktree on macOS with Node v22.23.1 (the
repository pins 24.18.0 in `.nvmrc`, which was not installed). Every check
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
report holds no diagnostic in any file this change touches.
- `npm run check:fast`: typecheck and lint passed. The lint report holds 0
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
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.
the 3.5% gate; no clone involves a file this pass changed.
- `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
process boundaries, the export CLI boundary, the CLI startup imports, the
agentenv verifier identity boundary, the bundle package pin conformance test,
the CI change scope boundary, the analysis-view CLI and MCP boundaries, and
the `serveEnvironment`, `serveEnvironmentState`, `environmentCapacity` and
`stateIntake` adapters tests: 14 files, 329 tests passed.
- After this file was written: `npm run format`, `npm run format:check` and
`npm run docs:check` passed again.
- `npm run build:cached` (the built output already included this pass's
sources), then `npm run test:focused --` the serve CLI, MCP and process
boundaries, the export CLI boundary, the CLI startup imports, the agentenv
verifier identity boundary, the bundle package pin conformance test, the CI
change scope boundary, the analysis-view CLI and MCP boundaries, and the
`serveEnvironment`, `serveEnvironmentState`, `environmentCapacity` and
`stateIntake` adapters tests: 14 files, 335 tests passed.
- 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
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.
The review fixes are covered by these tests, all passing in the runs above:
`--replay-only` against main: a script loaded main's (`3fca2e0c`) sources for
the Grafana-shaped fixture in-process and compared them with the built CLI
serving the same files with `--replay-only` and `ENVIRONMENT_NAME=renamed`. The
card was byte-identical (11,131 bytes; its SHA-256, computed from main's
sources, is the one the mcp-boundary test now pins), `tools/list` matched
main's 19 tool names, descriptions and input schemas, five tool calls
(`get_users_by_id` with `2` and `999`, `get_annotations`,
`get_folders_by_uid`, `post_folders`) returned main's replay results, and
`data/get` answered `-32601`.
- Response budget: a batch of four `data/get`s over a loaded state of about a
third of the budget keeps two results, answers the other two with `-32000`
and `max_response_bytes`, keeps the trailing reset's `{}`, and its body stays
within the budget.
- Batch order: a batch `[data/add (gated download), data/get]` with a
`data/reset` from another request sent while the download is held reports
the loaded state from its `data/get`, and the reset applies last.
- Cancellation: aborting the data plane's stop signal ends a stalled download
(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
announcement and reading `process.resourceUsage()` at exit, with a load average
of 6 to 7 on a shared host. Medians of six interleaved runs:
That each test fails without its fix is source-derived; the fixes were not
reverted to show it.
| Capture | Mode | Listen ms | CPU ms | Peak RSS MiB |
| --------------------------------- | ---------------------- | --------- | ------ | ------------ |
| Grafana fixture (25 KB) | `--replay-only` | 542 | 819 | 265 |
| Grafana fixture (25 KB) | default, derived state | 546 | 822 | 266 |
| Synthetic, 1,530 records (5.2 MB) | `--replay-only` | 610 | 934 | 277 |
| Synthetic, 1,530 records (5.2 MB) | default, derived state | 789 | 1234 | 303 |
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.
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
any path conversion on every platform), a real `https://` download (tests
inject the fetcher), the memory and time of a budgeted batch (not measured),
agent token usage, a real AgentEnv framework run, a real-world capture, and
write effects, which are a later increment.
src/application/environment/environmentState.test.ts
+23
View File
@@ -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
+23
View File
@@ -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
+35 -15
View File
@@ -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 = renderStateEntity(entity, view, volatile);
const missing = view.required_fields.filter(
(name) => !Object.hasOwn(shown, name),
const shown = shownEntity(
tool,
view,
entity,
new Set(store.type(view.type)?.volatile_fields),
);
if (missing.length > 0)
return err(
refused(
tool,
`The state's ${view.type} entity ${entity.id} lacks fields this tool's responses always carried: ${missing.join(", ")}.`,
),
);
if (!shown.ok) return shown;
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) =>
renderStateEntity(entity, view, volatile),
);
const items: JsonObject[] = [];
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
+34 -15
View File
@@ -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>();
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;
return viewFieldsProblem(state, type, entity, path);
};
/** The first problem with the documents: each names a document view once, at the initial document's path. */
src/server/environment/environmentDataPlane.ts
+49 -28
View File
@@ -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. */
const runMethod = (
store: EnvironmentStateStore,
intake: StateIntake,
method: string,
type Method = (
context: CallContext,
params: unknown,
): Promise<RpcOutcome> | undefined => {
switch (method) {
case "data/reset":
) => RpcOutcome | Promise<RpcOutcome>;
const METHODS: ReadonlyMap<string, Method> = new Map<string, Method>([
[
"data/reset",
({ store }) => {
store.reset();
return Promise.resolve({ result: {} });
case "data/add":
return addState(store, params, intake);
case "data/get":
return Promise.resolve({
result: { parts: [{ kind: "data", data: store.snapshot() }] },
});
default:
return undefined;
}
};
return { result: {} };
},
],
["data/add", addState],
[
"data/get",
({ store }) => ({
result: { parts: [{ kind: "data", data: store.snapshot() }] },
}),
],
]);
/**
* `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
* a later reset and at most one loaded document is parsed at once.
* at a time in the order they are made, across requests, so a slow load never
* 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) => {
const next = previous.then(() => runMethod(store, intake, method, params));
return (method, params, signal) => {
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
+126 -51
View File
@@ -40,14 +40,12 @@ import {
UNTRUSTED_CONTENT_INSTRUCTIONS,
} from "../toolResultEncoding.js";
import { RequestTrackingMcpServer } from "../RequestTrackingMcpServer.js";
import {
stateDataPlane,
type RpcMethods,
type RpcOutcome,
} from "./environmentDataPlane.js";
import { stateDataPlane, type RpcMethods } 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. */
const rpcReply = async (
/** 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. */
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 =
methods === undefined ? undefined : await methods(method, message.params);
if (!("id" in message)) return undefined;
return {
jsonrpc: "2.0",
id,
...(outcome ?? {
error: { code: METHOD_NOT_FOUND, message: `method not found: ${method}` },
}),
const outcome =
methods === undefined
? Promise.resolve(undefined)
: methods(method, message.params, signal);
const replies = "id" in message;
return outcome.then((settled): RpcReply | undefined =>
replies
? {
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.
if (Array.isArray(parsed.value)) {
if (parsed.value.length === 0)
return jsonRpcErrorResponse(400, INVALID_REQUEST, "invalid request");
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);
// The request's signal cancels its calls when the client leaves.
if (Array.isArray(parsed.value))
return answerBatch(parsed.value, methods, request.signal);
const reply = await startRpc(parsed.value, methods, request.signal);
return reply === undefined
? new Response(null, { status: 202 })
: Response.json(reply, {
status:
"error" in reply && reply.error.code === INVALID_REQUEST ? 400 : 200,
});
: jsonResponse(
replyEncoder()(reply),
"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, {
...DEFAULT_STATE_INTAKE,
...(options.fetchState === undefined
? {}
: { fetchState: options.fetchState }),
}),
: stateDataPlane(
store,
{
...DEFAULT_STATE_INTAKE,
...(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
+246 -4
View File
@@ -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. */
const dataPlane = (intake: Partial<StateIntake> = {}) => {
/** 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> = {},
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
+73 -27
View File
@@ -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 fetching = Promise.resolve().then(() => intake.fetchState(url, signal));
const bounds = downloadBounds(intake.timeoutMs, cancel);
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;
try {
path = fileURLToPath(url);
} catch {
return err(invalidParams("a file:// uri names a local absolute path"));
}
return readStateFile(path);
const path = localFilePath(url);
return path.ok ? readStateFile(path.value) : 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;
};