mirror of
https://github.com/vercel-labs/fx.git
synced 2026-10-28 10:15:17 +08:00
Keep tenants apart when libfx agents share a process
- A queue delivery runs only an input its session already stored. A delivery the session's log doesn't hold, such as one posted straight to the queue route, runs nothing, writes nothing, and emits `session.refused`. - Every durability, including `vercel()`, stores a prompt, steer, cancel or resume with its caller's context and settings before queueing it. The queued message carries only the input's id, session, agent name and store. - A steer that reuses a prompt's id is rejected with `FX_MESSAGE_ID_IN_USE` on every durability. On `vercel()` the worker used to drop it with a `session.error`. - Creating an agent under a name another agent already has on the same store, with different options, throws `FX_AGENT_NAME_IN_USE`. Agents created with the same options still share that name's sessions. - `wakeHandler()` delivers each message to the agent of its name on the store the message names, and answers `FX_NO_AGENT` when the process has none. It no longer hands the message to an agent of that name on another store. - `world()` takes a `store` option naming the store its sessions live in, defaulting to its name. `local()` treats each folder as its own store. - A prompt or steer whose input, context and settings come to more than 8 MiB of JSON is rejected with `FX_INPUT_TOO_LARGE` before anything is stored. - The Node native addon reads the environment from libc on every lookup, so an app that adds environment variables after the first engine starts no longer crashes the process on macOS. - Prompts queued by an earlier libfx and still undelivered when this version deploys are refused instead of run.
This commit is contained in:
36 files changed
+1367
-175
No files matched your search
@@ -389,8 +389,8 @@ jobs:
|
||||
bun tests/tla/drivers/drive.mjs cutoff --walks 20 --seed 5
|
||||
bun tests/tla/drivers/drive.mjs leased --walks 40 --seed 12
|
||||
bun tests/tla/drivers/drive.mjs queue --walks 40 --seed 4
|
||||
bun tests/tla/drivers/drive-agents.mjs send --walks 60 --seed 11
|
||||
bun tests/tla/drivers/drive-agents.mjs consume --walks 60 --seed 13
|
||||
bun tests/tla/drivers/drive-agents.mjs --walks 60 --seed 11
|
||||
bun tests/tla/drivers/drive-tenants.mjs --walks 60 --seed 11
|
||||
bun tests/tla/drivers/drive-steer.mjs ideal --walks 20 --seed 2
|
||||
bun tests/tla/drivers/drive-steer.mjs faults --walks 20 --seed 4
|
||||
bun tests/tla/drivers/drive-steer.mjs faults --walks 10 --seed 4 --hooks
|
||||
|
||||
+1
-1
@@ -59,7 +59,7 @@ Do not treat the demos or this file as the implementation contract. When prose a
|
||||
| `encodeXtermKeyEvent()` or `xtermAdapter()` only | `sdk/node/test-xterm-adapter.mjs` |
|
||||
| Core debugger query behavior or automation state | `sdk/tests/test-core-browser.mjs` |
|
||||
| Terminal demo asset references, integrity, or cache policy | Package into a fresh temporary directory and inspect the generated HTML, manifest, and headers |
|
||||
| Durable sessions: leases, fencing, takeovers, tool reruns, the UI stream | `sdk/tests/test-durable-core.mjs` on memory, local, world, and queue (the core with a scripted harness; world is an app-built World passed to `world()`, queue one whose queue delivers as Vercel Queues does, with `queueLeases`), `sdk/tests/test-durable.mjs` on memory, local, and world, then `tests/tla/sdk/check.sh` and `bun tests/tla/drivers/drive.mjs send`, `lookup`, `write`, `cutoff` (deadline cut-offs up to their bound), `leased` (a World without liveness, where each worker renews a short lease), and `queue` (a World whose queue shows whether a worker lives, so leases are held without renewals), which walk the real code along the TLC state graphs of `DurableSession.tla`; change the spec with the code. For agent names and routing, also `bun tests/tla/drivers/drive-agents.mjs send` and `consume`, which walk `AgentRouting.tla`'s graphs with each process in a worker thread of its own. With a Vercel access token, also run both against Vercel's World: `test-durable.mjs vercel native` and `drive.mjs --durability vercel`, configured as `sdk/tests/vercel-storage.mjs` describes |
|
||||
| Durable sessions: leases, fencing, takeovers, tool reruns, the UI stream | `sdk/tests/test-durable-core.mjs` on memory, local, world, and queue (the core with a scripted harness; world is an app-built World passed to `world()`, queue one whose queue delivers as Vercel Queues does, with `queueLeases`), `sdk/tests/test-durable.mjs` on memory, local, and world, then `tests/tla/sdk/check.sh` and `bun tests/tla/drivers/drive.mjs send`, `lookup`, `write`, `cutoff` (deadline cut-offs up to their bound), `leased` (a World without liveness, where each worker renews a short lease), and `queue` (a World whose queue shows whether a worker lives, so leases are held without renewals), which walk the real code along the TLC state graphs of `DurableSession.tla`; change the spec with the code. For agent names and routing, also `bun tests/tla/drivers/drive-agents.mjs`, which walks `AgentRouting.tla`'s graph with each process in a worker thread of its own. For what a delivery may run, on whose agent and store, and input limits, also `bun tests/tla/drivers/drive-tenants.mjs`, which walks `TenantIsolation.tla`'s graph with four tenants and a forger sharing one route. With a Vercel access token, also run both against Vercel's World: `test-durable.mjs vercel native` and `drive.mjs --durability vercel`, configured as `sdk/tests/vercel-storage.mjs` describes |
|
||||
| Steering: interrupts, waits during tools, joins, hand-backs on cancel, image handoffs, steer ids | The libfx steering tests in `src/acp/libfx_steering.zig`, `sdk/tests/test-agent-steering.mjs` and `sdk/tests/test-durable.mjs` on native and wasm, then `tests/tla/sdk/check.sh` and `bun tests/tla/drivers/drive-steer.mjs ideal` and `faults`, with `--backend wasm` too, which walk the real code along the TLC state graphs of `DurableSteer.tla`; change the spec with the code |
|
||||
| Supported public behavior or setup | Update `sdk/README.md` in the same change |
|
||||
|
||||
|
||||
+9
-3
@@ -56,7 +56,8 @@ agent.session(sessionId).prompt("Start with the schema.");
|
||||
- **One turn at a time.** A prompt sent while a turn runs waits behind it, even when it reaches a different server.
|
||||
- **Turns finish on their own.** A turn runs to the end whether or not anyone reads its events, and a turn nobody reads holds no connection open: its view starts reading only once you use its events, `readable` or `result`. Stop it with `session.cancel()`, or with the `signal` you passed to `prompt()`.
|
||||
- **Steering.** `session.steer(input)` steers the running turn the way the fx CLI does. While the model streams a reply, the steer cuts it short and the turn continues with it. While a tool runs, the steer waits for the tool. A steer sent while a turn waits to start joins that turn, and when no turn runs or waits, the steer becomes the next turn. A steer with images can't join a turn, so the running turn ends at its next safe point and the steer runs next.
|
||||
- **Retries are safe.** Give a prompt or a steer a `messageId`, and a repeated request with the same id runs nothing new. A repeated prompt follows the original turn, and its result has `repeated: true`. One id names one input, so a steer that reuses a prompt's id is rejected with `FX_MESSAGE_ID_IN_USE`. On a queue that keeps messages past the process, such as `vercel()`'s, the sender doesn't read the session first, so the worker drops that steer and emits a `session.error` with the same code. A steer's `user_message` event carries its id as `id` where it joined a turn.
|
||||
- **Retries are safe.** Give a prompt or a steer a `messageId`, and a repeated request with the same id runs nothing new. A repeated prompt follows the original turn, and its result has `repeated: true`. One id names one input, so a steer that reuses a prompt's id is rejected with `FX_MESSAGE_ID_IN_USE`. A steer's `user_message` event carries its id as `id` where it joined a turn.
|
||||
- **Inputs have a size limit.** A prompt or steer whose input, with the session's `context` and settings, comes to more than 8 MiB of JSON is rejected with `FX_INPUT_TOO_LARGE` before anything is stored. The model can't take a larger one.
|
||||
|
||||
Each conversation can bring its own settings, which apply to the turns that session object starts:
|
||||
|
||||
@@ -126,7 +127,9 @@ createFxAgent({ durability: memory() }); // nothing outlives the process
|
||||
|
||||
On `local()`, `vercel()` and `world()`, a session id is a Workflow run id, `wrun_` followed by letters and digits. Take it from `turn.accepted` or `await agent.newSessionId()` instead of making one up. On `memory()`, any 1 to 128 letters, digits, `.`, `_` or `-` work.
|
||||
|
||||
A session stores the conversation and nothing else. Every process passes its own model, credentials, tools, MCP clients and skills when it creates the agent. If a session's saved checkpoint came from a different libfx version, tool set or model, the session still loads, and `onEvent` gets a `checkpoint.mismatch` event naming what changed.
|
||||
A session stores the conversation, and with each prompt and steer the caller's `context` and the session object's settings. Every process passes its own model, credentials, tools, MCP clients and skills when it creates the agent.
|
||||
|
||||
libfx doesn't know who is calling. Anyone with a session's id can prompt, steer, cancel, stream or checkpoint it, so your app must check that the caller may use a session before it opens it. A queue message only names an input libfx already stored: a delivery that names one the session's log doesn't hold runs nothing, and the agent's `onEvent` gets a `session.refused` event. If a session's saved checkpoint came from a different libfx version, tool set or model, the session still loads, and `onEvent` gets a `checkpoint.mismatch` event naming what changed.
|
||||
|
||||
### Several agents
|
||||
|
||||
@@ -139,10 +142,11 @@ export const coding = createFxAgent({ name: "coding", model: "anthropic/claude-o
|
||||
|
||||
A session belongs to the agent that created it, and its turns run only on an agent with that name. A prompt sent through another agent, such as `coding.session(supportSessionId).prompt()`, still runs on `support`, and the sending agent's `onEvent` gets a `session.forwarded` event. An agent created without a name is named `default`.
|
||||
|
||||
- Agents with the same name are copies of one agent, such as one on each server, and take turns running its sessions.
|
||||
- Agents with the same name are copies of one agent, such as one on each server, and take turns running its sessions. In one process, every agent with a name on a store must have the same configuration: creating one with different options throws `FX_AGENT_NAME_IN_USE`. Agents created without a name or a durability share the name `default` and the default store, so an app that serves several customers gives each customer's agent its own name. Create an agent once, at module scope, or close it when you're done with it.
|
||||
- Keep names stable. A session keeps the name it was created with, so renaming an agent leaves its sessions with no agent to run them.
|
||||
- A process that gets a turn for a name it has no agent for refuses it with an `FX_NO_AGENT` error naming the agent, and the queue delivers it again.
|
||||
- Agents share sessions through their durability. Give agents that share sessions the same durability, or none: agents created without one share the default.
|
||||
- One route takes the deliveries for every World in the process and hands each to the agent of its name on the store it names. A World's store is its `store` option, or its name. An app that serves several customers gives each customer a store of its own, with `world()`'s `store`, and never shares one store between customers who must not reach each other's sessions.
|
||||
|
||||
### On Vercel
|
||||
|
||||
@@ -170,6 +174,8 @@ Then subscribe it to libfx's topic in `vercel.json`:
|
||||
}
|
||||
```
|
||||
|
||||
The trigger makes the route private: it has no public URL, and only Vercel Queues can call it. Keep `wakeHandler()` on that route only. An app whose own World queues to a route the internet can reach must check that each request comes from its queue before it calls `wakeHandler()`. libfx runs only inputs its sessions already stored, but a forged request can still wake a session.
|
||||
|
||||
That's all. When no other invocation holds the session, `prompt()` runs the turn in the request that called it, so the first model request goes out without a queue hop, and the function stays alive for the turn after the route responds. Vercel Queues keeps the prompt until after that request's time limit, and if the request dies mid-turn, that delivery continues the turn. Other turns go through Vercel Queues to this route:
|
||||
|
||||
- When a turn gets within 30 seconds of the function's time limit, libfx stops it, and the next invocation continues it. `vercel({ reserveMs })` from `libfx/durable-vercel` changes that margin.
|
||||
|
||||
+200
-59
@@ -338,7 +338,14 @@ export function foldSessionLog(entries, { now = 0, alive = () => null } = {}) {
|
||||
// landed ended. Its id is kept like a turn's, so a late delivery of the
|
||||
// same message is not taken as a new cancel.
|
||||
for (const input of inputs) {
|
||||
if (input.type !== "cancel" || consumed.has(input.messageId)) continue;
|
||||
if (consumed.has(input.messageId)) continue;
|
||||
// A resume only asks that its session be woken: it is spent once
|
||||
// stored, and its id is kept so its deliveries are known.
|
||||
if (input.type === "resume") {
|
||||
consume(input.messageId);
|
||||
continue;
|
||||
}
|
||||
if (input.type !== "cancel") continue;
|
||||
const live = typeof input.target === "string"
|
||||
? !consumed.has(input.target) || openTurn?.id === input.target
|
||||
: openTurn !== null && input.cursor > openTurn.at;
|
||||
@@ -659,6 +666,10 @@ function noAgentNamed(name) {
|
||||
// copy of this module in the process.
|
||||
const agentsByName = (globalThis[Symbol.for("libfx.durableAgents")] ??= new Map());
|
||||
|
||||
// The backend a member opens, or null when it cannot open, as an agent
|
||||
// libfx refused to create cannot.
|
||||
const openMember = (member) => member.open().catch(() => null);
|
||||
|
||||
// The World backend of an agent named `name` in this process whose
|
||||
// durability `matches`, the agent's copies in turn, or null.
|
||||
async function agentNamed(name, matches) {
|
||||
@@ -669,8 +680,8 @@ async function agentNamed(name, matches) {
|
||||
for (let index = 0; index < members.length; index += 1) {
|
||||
const member = members[(first + index) % members.length];
|
||||
if (!matches(await member.durability())) continue;
|
||||
const created = await member.open();
|
||||
if (typeof created.deliverHere === "function") return created;
|
||||
const created = await openMember(member);
|
||||
if (typeof created?.deliverHere === "function") return created;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
@@ -679,30 +690,82 @@ async function agentNamed(name, matches) {
|
||||
// one on another durability can hold other sessions.
|
||||
const agentSharing = (name, durability) => agentNamed(name, (other) => other === durability);
|
||||
|
||||
// A route takes its deployment's deliveries, which every route durability
|
||||
// of the same kind in the process shares, such as each `vercel()` an app
|
||||
// creates: the agent a message names on this durability, or on another of
|
||||
// that kind.
|
||||
const agentOnRoute = async (name, durability) => (await agentSharing(name, durability)) ??
|
||||
agentNamed(name, (other) => other !== durability && other.name === durability.name && typeof other.world === "function" && other.create === undefined);
|
||||
// The store a World keeps its sessions in, as its queue messages name it:
|
||||
// its `store`, which is its name unless the app named the store.
|
||||
const worldStore = (durability) => durability?.store ?? durability?.name;
|
||||
|
||||
// The store a durability keeps its sessions in, as libfx tells stores apart:
|
||||
// a memory() is its own, and a World's is its worldStore(). Two agents of one
|
||||
// name on one store are one agent.
|
||||
const storeOf = (durability) => (typeof durability?.create === "function" ? durability : `world:${worldStore(durability)}`);
|
||||
|
||||
// A route takes its deployment's deliveries for every World in the process,
|
||||
// so a delivery goes to the agent its message names on the store it names,
|
||||
// and never to one of that name on another store
|
||||
// (`tests/tla/sdk/TenantIsolation.tla` RunsInSenderStore).
|
||||
const agentOnStore = (name, store) => agentNamed(name, (other) => typeof other.world === "function" && other.create === undefined && worldStore(other) === store);
|
||||
|
||||
/**
|
||||
* The request handler a deployment mounts for queue deliveries, for every
|
||||
* durable agent in the process: each delivery goes to the agent its message
|
||||
* names. One route and one queue trigger serve any number of agents.
|
||||
* names, on the store it names. One route and one queue trigger serve any
|
||||
* number of agents.
|
||||
*/
|
||||
export function wakeHandler() {
|
||||
return async (request) => {
|
||||
for (const group of agentsByName.values()) {
|
||||
for (const member of group.members) {
|
||||
const created = await member.open();
|
||||
if (created.viaRoute === true) return created.queueHandler(request);
|
||||
const created = await openMember(member);
|
||||
if (created?.viaRoute === true) return created.queueHandler(request);
|
||||
}
|
||||
}
|
||||
return new Response("no libfx agent in this process takes queue deliveries from a route", { status: 503 });
|
||||
};
|
||||
}
|
||||
|
||||
const nameInUse = (name) => Object.assign(
|
||||
new Error(`an agent named ${JSON.stringify(name)} with another configuration already runs on this store in this process; give each configuration its own name, or each store its own \`store\``),
|
||||
{ code: "FX_AGENT_NAME_IN_USE", agent: name },
|
||||
);
|
||||
|
||||
// Whether two agents' options are one configuration: the same values, and
|
||||
// the same functions and objects other than plain ones. `onEvent` only
|
||||
// watches the agent, so it is left out.
|
||||
function definitionOf(options) {
|
||||
const { onEvent: _onEvent, ...definition } = options;
|
||||
return definition;
|
||||
}
|
||||
function sameDefinition(a, b, depth = 0) {
|
||||
if (Object.is(a, b)) return true;
|
||||
if (depth > 32 || a === null || b === null || typeof a !== "object" || typeof b !== "object") return false;
|
||||
if (ArrayBuffer.isView(a) || ArrayBuffer.isView(b)) {
|
||||
if (!ArrayBuffer.isView(a) || !ArrayBuffer.isView(b) || a.byteLength !== b.byteLength) return false;
|
||||
const left = new Uint8Array(a.buffer, a.byteOffset, a.byteLength);
|
||||
const right = new Uint8Array(b.buffer, b.byteOffset, b.byteLength);
|
||||
for (let index = 0; index < left.length; index += 1) if (left[index] !== right[index]) return false;
|
||||
return true;
|
||||
}
|
||||
const plain = (value) => Array.isArray(value) || Object.getPrototypeOf(value) === Object.prototype || Object.getPrototypeOf(value) === null;
|
||||
if (!plain(a) || !plain(b) || Array.isArray(a) !== Array.isArray(b)) return false;
|
||||
const keys = Object.keys(a);
|
||||
if (keys.length !== Object.keys(b).length) return false;
|
||||
return keys.every((key) => Object.hasOwn(b, key) && sameDefinition(a[key], b[key], depth + 1));
|
||||
}
|
||||
|
||||
// Refuses `member` when another agent of `name` on its store has another
|
||||
// configuration: deliveries for a name on a store go to any of its agents,
|
||||
// so they must be one agent (`tests/tla/sdk/TenantIsolation.tla`
|
||||
// RunsOnSenderAgent). An agent created with the `replicaOption` symbol set
|
||||
// stands for another server's copy, as a test that runs several servers in
|
||||
// one process creates it, and is never refused.
|
||||
const replicaOption = Symbol.for("libfx.replica");
|
||||
function claimName(name, member) {
|
||||
if (member.replica) return;
|
||||
for (const other of agentsByName.get(name)?.members ?? []) {
|
||||
if (other !== member && !other.replica && other.store === member.store && !sameDefinition(other.definition, member.definition)) throw nameInUse(name);
|
||||
}
|
||||
}
|
||||
|
||||
// libfx starts and closes a World it `owned`, having created it; a World the
|
||||
// app passed in stays the app's.
|
||||
function createWorldBackend(world, durability) {
|
||||
@@ -738,16 +801,19 @@ function createWorldBackend(world, durability) {
|
||||
if (result?.redeliver === true) throw new Error("libfx asks the queue to deliver this message again");
|
||||
return Number.isFinite(result?.timeoutSeconds) ? { timeoutSeconds: result.timeoutSeconds } : undefined;
|
||||
};
|
||||
const store = worldStore(durability);
|
||||
// The route a deployment mounts delivers to the agent in this process the
|
||||
// message names, whichever agent's route it is.
|
||||
// message names, on the store it names, whichever agent's route it is. A
|
||||
// message from before stores were named is for this World's store.
|
||||
const queueHandler = world.createQueueHandler(queuePrefix, async (message, meta) => {
|
||||
const named = agentOf(message?.input);
|
||||
const agent = await agentOnRoute(named, durability);
|
||||
const agent = await agentOnStore(named, typeof message?.input?.store === "string" ? message.input.store : store);
|
||||
if (!agent) throw noAgentNamed(named);
|
||||
return agent.deliverHere(message, meta);
|
||||
});
|
||||
const backend = {
|
||||
name,
|
||||
store,
|
||||
livenessKnown,
|
||||
queueLeases,
|
||||
// A prompt runs in the request that took it, held by its backstop's
|
||||
@@ -1129,10 +1195,53 @@ const errorSummary = (error) => ({ name: error?.name ?? "Error", message: String
|
||||
// The harness contract: a fenced write rejects with this code.
|
||||
const isFenced = (error) => error?.code === "FX_FENCED";
|
||||
|
||||
// What a queue message adds to its session's log: a restored agent's first
|
||||
// The kinds of message a session's queue carries.
|
||||
const messageTypes = new Set(["prompt", "steer", "cancel", "resume"]);
|
||||
|
||||
// What the queue carries for a message: only which input of which session
|
||||
// it names, for which agent, on which store. A delivery can arrive at a
|
||||
// public route, so its body may come from anyone: a worker acts only on a
|
||||
// message its session's log already holds, and runs only what that log
|
||||
// says (`tests/tla/sdk/TenantIsolation.tla` OnlySentRuns).
|
||||
function queuedMessage(message, store) {
|
||||
return {
|
||||
type: message.type,
|
||||
messageId: message.messageId,
|
||||
sessionId: message.sessionId,
|
||||
agent: typeof message.agent === "string" ? message.agent : defaultAgentName,
|
||||
...(typeof store === "string" ? { store } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
// The message a delivery names, as queuedMessage() has it, or null for one
|
||||
// no sender could have queued on `backend`.
|
||||
function deliveredMessage(backend, delivered) {
|
||||
if (!delivered || typeof delivered !== "object" || !messageTypes.has(delivered.type)) return null;
|
||||
if (!validId(delivered.messageId) || !backend.validSessionId(delivered.sessionId)) return null;
|
||||
if (delivered.agent !== undefined && typeof delivered.agent !== "string") return null;
|
||||
return queuedMessage(delivered);
|
||||
}
|
||||
|
||||
// Whether the session's log holds the input `message` names, or names it
|
||||
// among those a turn already took.
|
||||
const knownMessage = (state, message) => state.inputKeys.has(`${message.type}:${message.messageId}`) || state.consumed.has(message.messageId);
|
||||
|
||||
// The most a durable input may hold: its content blocks with its caller's
|
||||
// context and its session object's settings, as JSON. The engine refuses a
|
||||
// prompt frame over 8 MiB (`maxPromptFrameBytes` in fx-sdk.js), so a larger
|
||||
// input could never run; it is refused before anything is stored.
|
||||
const maxInputBytes = 8 * 1024 * 1024;
|
||||
const jsonBytes = (value) => (value === undefined ? 0 : encoder.encode(JSON.stringify(value)).byteLength);
|
||||
const inputTooLarge = (bytes) => Object.assign(
|
||||
new RangeError(`a durable input with its context and settings is ${bytes} bytes; the most is ${maxInputBytes}`),
|
||||
{ code: "FX_INPUT_TOO_LARGE", bytes, limit: maxInputBytes },
|
||||
);
|
||||
|
||||
// What a message adds to its session's log: a restored agent's first
|
||||
// checkpoint and the message's input, a prompt with its caller's context and
|
||||
// its session object's settings.
|
||||
// Nothing the log already holds. Pure.
|
||||
// its session object's settings. Only the sender calls it, before it queues
|
||||
// the message, so what a turn runs and as whom comes from the app's own
|
||||
// call, never from a queue delivery. Nothing the log already holds. Pure.
|
||||
function messageEntries(state, message) {
|
||||
const entries = [];
|
||||
const named = typeof message.agent === "string" ? { agent: message.agent } : {};
|
||||
@@ -1141,7 +1250,7 @@ function messageEntries(state, message) {
|
||||
}
|
||||
const key = `${message.type}:${message.messageId}`;
|
||||
const known = state.inputKeys.has(key) || state.consumed.has(message.messageId) || idInUse(state, message);
|
||||
if ((message.type === "prompt" || message.type === "steer" || message.type === "cancel") && !known) {
|
||||
if (messageTypes.has(message.type) && !known) {
|
||||
entries.push({
|
||||
k: "input",
|
||||
key,
|
||||
@@ -1174,9 +1283,9 @@ const messageIdInUse = (message) => Object.assign(
|
||||
{ code: "FX_MESSAGE_ID_IN_USE" },
|
||||
);
|
||||
|
||||
// Writes what a queue message adds to its session, in order, and returns how
|
||||
// many entries it wrote. The sender and the consumer can both call it. A
|
||||
// message that reuses another input's id writes nothing and throws.
|
||||
// Writes what a message adds to its session, in order, and returns how many
|
||||
// entries it wrote. Only the sender calls it, before it queues the message.
|
||||
// A message that reuses another input's id writes nothing and throws.
|
||||
async function recordMessage(log, state, message) {
|
||||
if (idInUse(state, message)) throw messageIdInUse(message);
|
||||
const entries = messageEntries(state, message);
|
||||
@@ -1259,23 +1368,30 @@ class SessionWorker {
|
||||
}
|
||||
|
||||
// One queue delivery, `delivery` its message's id and attempt when the
|
||||
// queue reports them. Its input reaches the log at once, even while this
|
||||
// process runs the session, so a steer or cancel reaches the running turn.
|
||||
// Then it runs the session when no live worker holds it. Returns
|
||||
// `{ timeoutSeconds }` to have the same message delivered again later.
|
||||
// queue reports them. Its sender stored its input before queueing it, so a
|
||||
// steer or cancel reaches the running turn even before this runs. A
|
||||
// delivery the session's log does not hold is not from a sender and is
|
||||
// dropped before anything is written, whether or not its session exists
|
||||
// (`tests/tla/sdk/TenantIsolation.tla` OnlySentRuns). Then it runs the
|
||||
// session when no live worker holds it. Returns `{ timeoutSeconds }` to
|
||||
// have the same message delivered again later.
|
||||
async consume(message, delivery = null) {
|
||||
const log = await this.backend.session(this.sessionId);
|
||||
let entries = await log.read();
|
||||
let state = this.fold(entries, delivery);
|
||||
if (!knownMessage(state, message)) {
|
||||
this.agent.emit("session.refused", { sessionId: this.sessionId, messageId: message.messageId, reason: "unknown message" });
|
||||
return undefined;
|
||||
}
|
||||
await this.backend.refreshDeadline?.();
|
||||
// When this delivery's function stops, if it does.
|
||||
const deadline = this.backend.deadline() ?? (this.backend.maxDurationMs === null ? null : Date.now() + this.backend.maxDurationMs);
|
||||
let entries = await log.read();
|
||||
let state = this.fold(entries, delivery);
|
||||
// A session runs only on the agent it belongs to, which a new one is the
|
||||
// agent that sent its first message. A message that reached another
|
||||
// agent, as one sent to another agent's session does, goes on to it.
|
||||
// A session runs only on the agent it belongs to, the one its first
|
||||
// stored input names. A message that reached another agent, as one sent
|
||||
// to another agent's session does, goes on to it.
|
||||
const owner = state.agent ?? agentOf(message);
|
||||
if (owner !== this.agent.name) {
|
||||
await this.backend.queue.send({ ...message, agent: owner }, {
|
||||
await this.backend.queue.send(queuedMessage({ ...message, agent: owner }, this.backend.store), {
|
||||
idempotencyKey: `${message.sessionId}:${message.type}:${message.messageId}:to:${owner}`,
|
||||
});
|
||||
this.agent.emit("session.forwarded", { sessionId: this.sessionId, agent: owner });
|
||||
@@ -1283,26 +1399,13 @@ class SessionWorker {
|
||||
}
|
||||
for (let claims = 0; claims < 8; claims += 1) {
|
||||
// A session with no open turn starts its next prompt at once: the
|
||||
// first model request goes out while the input and the lease are
|
||||
// written, and nothing else is written until both land.
|
||||
// first model request goes out while the lease is written, and
|
||||
// nothing else is written until it lands.
|
||||
const early = this.running || state.lease ? null : earlyStart(entries, state, message, (items) => this.fold(items, delivery));
|
||||
let outcome;
|
||||
if (early) {
|
||||
outcome = await this.claim(log, state, deadline, early, delivery);
|
||||
} else {
|
||||
let wrote;
|
||||
try {
|
||||
wrote = await recordMessage(log, state, message);
|
||||
} catch (error) {
|
||||
if (error?.code !== "FX_MESSAGE_ID_IN_USE") throw error;
|
||||
// Its sender could not hear it: a durable queue took it first.
|
||||
this.agent.emit("session.error", { sessionId: this.sessionId, error: errorSummary(error) });
|
||||
return undefined;
|
||||
}
|
||||
if (wrote > 0) {
|
||||
entries = await log.read();
|
||||
state = this.fold(entries, delivery);
|
||||
}
|
||||
if (this.running) return { timeoutSeconds: runningBackstopSeconds };
|
||||
if (!state.hasWork) {
|
||||
if (message.type === "resume") await this.answerIdle(log, message);
|
||||
@@ -2184,10 +2287,29 @@ export function createDurableAgentFactory({ harness, defaultDurability, name = "
|
||||
if (closed) return Promise.reject(new Error(`${label} is closed`));
|
||||
backendPromise ??= (async () => {
|
||||
const chosen = await durabilityOf();
|
||||
// An agent created without a durability learns its store only now.
|
||||
if (member.store !== storeOf(chosen)) {
|
||||
member.store = storeOf(chosen);
|
||||
try {
|
||||
claimName(agentName, member);
|
||||
} catch (error) {
|
||||
leaveGroup(agentsByName, agentName, registered, member);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
chosenDurability = chosen;
|
||||
const created = await createBackend(chosen);
|
||||
await created.start?.();
|
||||
unlisten = created.listen((message, meta) => receive(created, message, deliveryOf(meta)), agentName);
|
||||
unlisten = created.listen((delivered, meta) => {
|
||||
// Only which input a delivery names is read from it: the rest of
|
||||
// its body, which a forged delivery would choose, is ignored.
|
||||
const message = deliveredMessage(created, delivered);
|
||||
if (!message) {
|
||||
emit("session.refused", { reason: "malformed message" });
|
||||
return undefined;
|
||||
}
|
||||
return receive(created, message, deliveryOf(meta));
|
||||
}, agentName);
|
||||
return created;
|
||||
})();
|
||||
return backendPromise;
|
||||
@@ -2195,8 +2317,10 @@ export function createDurableAgentFactory({ harness, defaultDurability, name = "
|
||||
// The durability the agent's sessions live in, chosen once.
|
||||
const durabilityOf = () => (durabilityPromise ??= durability ? Promise.resolve(durability) : defaultDurability());
|
||||
// In the process's agents from now, so a route that receives a message
|
||||
// for this agent finds it before anything uses it.
|
||||
const member = { open: backend, durability: durabilityOf };
|
||||
// for this agent finds it before anything uses it. One created without a
|
||||
// durability claims the default's store until it knows which that is.
|
||||
const member = { open: backend, durability: durabilityOf, definition: definitionOf(harnessOptions), store: durability ? storeOf(durability) : defaultDurability, replica: options[replicaOption] === true };
|
||||
claimName(agentName, member);
|
||||
const registered = joinGroup(agentsByName, agentName, member);
|
||||
// One delivery of `message` to the session's worker here: from the
|
||||
// queue, with `delivery` naming the queue's message and attempt, or with
|
||||
@@ -2243,6 +2367,14 @@ export function createDurableAgentFactory({ harness, defaultDurability, name = "
|
||||
try { JSON.stringify(context); } catch { throw new TypeError("session context must be JSON"); }
|
||||
}
|
||||
const settings = plugged.settings?.(sessionOptions ?? {});
|
||||
// What every input this session object sends carries besides itself.
|
||||
const carried = jsonBytes(context) + jsonBytes(settings);
|
||||
// Refuses an input too large to store before anything is stored
|
||||
// (`tests/tla/sdk/TenantIsolation.tla` StoredWithinLimit).
|
||||
const admit = (input) => {
|
||||
const bytes = jsonBytes(input) + carried;
|
||||
if (bytes > maxInputBytes) throw inputTooLarge(bytes);
|
||||
};
|
||||
let id = requestedId ?? null;
|
||||
let seeded = seed === undefined || requestedId !== undefined;
|
||||
const resolved = backend().then((created) => {
|
||||
@@ -2261,12 +2393,15 @@ export function createDurableAgentFactory({ harness, defaultDurability, name = "
|
||||
...(settings === undefined ? {} : { settings }),
|
||||
};
|
||||
const full = { ...message, ...extra, agent: agentName, sessionId };
|
||||
// A queue that ends with the process could lose the message, so its
|
||||
// input reaches the log before the caller hears it was accepted.
|
||||
if (!created.queueDurable) {
|
||||
const log = await created.session(sessionId);
|
||||
await recordMessage(log, foldSessionLog(await log.read()), full);
|
||||
}
|
||||
// The input and its caller's context reach the log here, on the
|
||||
// app's own call, before the caller hears it was accepted. The queue
|
||||
// carries only which input to run, so a delivery cannot choose what
|
||||
// a turn runs or as whom (`tests/tla/sdk/TenantIsolation.tla`
|
||||
// OnlySentRuns), and a queue that ends with the process loses no
|
||||
// input.
|
||||
const log = await created.session(sessionId);
|
||||
await recordMessage(log, foldSessionLog(await log.read()), full);
|
||||
const queued = queuedMessage(full, created.store);
|
||||
const idempotencyKey = `${sessionId}:${message.type}:${message.messageId}`;
|
||||
if (inRequest && created.runInRequest) {
|
||||
await created.refreshDeadline?.();
|
||||
@@ -2278,16 +2413,16 @@ export function createDurableAgentFactory({ harness, defaultDurability, name = "
|
||||
// if the request died, the backstop takes the session over and
|
||||
// continues the turn.
|
||||
const delaySeconds = Math.ceil(Math.max(0, deadline - Date.now()) / 1000) + inlineBackstopMarginSeconds;
|
||||
const { messageId: backstop } = await created.queue.send(full, { idempotencyKey, delaySeconds }) ?? {};
|
||||
const { messageId: backstop } = await created.queue.send(queued, { idempotencyKey, delaySeconds }) ?? {};
|
||||
seeded = true;
|
||||
const delivery = backstop ? { messageId: backstop, attempt: 0 } : null;
|
||||
// receive() reports a failure to the host; the backstop retries it.
|
||||
const work = receive(created, full, delivery, { inRequest: true }).then(() => {}, () => {});
|
||||
const work = receive(created, queued, delivery, { inRequest: true }).then(() => {}, () => {});
|
||||
created.waitUntil?.(work);
|
||||
return;
|
||||
}
|
||||
}
|
||||
await created.queue.send(full, { idempotencyKey });
|
||||
await created.queue.send(queued, { idempotencyKey });
|
||||
seeded = true;
|
||||
};
|
||||
const startTurn = (type, messageId, input, retried = false, aborted = false) => {
|
||||
@@ -2372,6 +2507,7 @@ export function createDurableAgentFactory({ harness, defaultDurability, name = "
|
||||
if (Array.isArray(input) && input.some((block) => block?.type === "image" && block.data !== undefined && typeof block.data !== "string")) {
|
||||
throw new TypeError("a durable prompt carries image data as a base64 string");
|
||||
}
|
||||
admit(input);
|
||||
const signal = promptOptions.signal;
|
||||
if (signal !== undefined && typeof signal?.addEventListener !== "function") throw new TypeError("prompt signal must be an AbortSignal");
|
||||
const turn = startTurn("prompt", messageId, input, promptOptions.messageId !== undefined, signal?.aborted === true);
|
||||
@@ -2479,6 +2615,7 @@ export function createDurableAgentFactory({ harness, defaultDurability, name = "
|
||||
if (steerOptions === null || typeof steerOptions !== "object") throw new TypeError("steer options must be an object");
|
||||
const messageId = steerOptions.messageId ?? newId("str");
|
||||
if (!validId(messageId)) throw new TypeError(`messageId must be ${idRule}`);
|
||||
admit(input);
|
||||
await send({ type: "steer", messageId, input });
|
||||
return { messageId };
|
||||
},
|
||||
@@ -2519,7 +2656,11 @@ export function createDurableAgentFactory({ harness, defaultDurability, name = "
|
||||
// Waits for what a session's running turn has written so far.
|
||||
settled: async (sessionId) => { await workers.get(sessionId)?.settled?.(); },
|
||||
// Delivers a queue message to this agent as its queue would.
|
||||
deliver: async (message, delivery) => receive(await backend(), message, delivery),
|
||||
deliver: async (delivered, delivery) => {
|
||||
const created = await backend();
|
||||
const message = deliveredMessage(created, delivered);
|
||||
return message ? receive(created, message, delivery) : undefined;
|
||||
},
|
||||
liveWorkers: () => workers.size,
|
||||
// A session's log as one worker holds it, with its own view of the
|
||||
// chain.
|
||||
|
||||
@@ -2,6 +2,7 @@
|
||||
// libfx bundles that World into this module when it is packaged, so an app
|
||||
// installs nothing more.
|
||||
import { hostname } from "node:os";
|
||||
import { resolve } from "node:path";
|
||||
import { createWorld } from "@workflow/world-local";
|
||||
import { world } from "./world.mjs";
|
||||
|
||||
@@ -17,6 +18,8 @@ export function local({ dir, maxDurationMs, reserveMs } = {}) {
|
||||
if (dir !== undefined && (typeof dir !== "string" || dir.length === 0)) throw new TypeError("local() dir must be a path");
|
||||
return world(() => createWorld({ ...(dir === undefined ? {} : { dataDir: dir }), recoverActiveRuns: false }), {
|
||||
name: "local",
|
||||
// Each folder is a store of its own.
|
||||
store: `local:${dir === undefined ? "" : resolve(dir)}`,
|
||||
livenessKnown: true,
|
||||
// world-local keeps its queue in memory, so a message dies with the process.
|
||||
queueDurable: false,
|
||||
|
||||
@@ -32,18 +32,23 @@ const durabilityTag = Symbol.for("libfx.durability");
|
||||
* - `leaseMs`: how long a lease lasts after it was taken or last renewed,
|
||||
* when liveness is unknown. Default 15000.
|
||||
* - `queueDurable`: the queue keeps a message until a delivery acknowledges
|
||||
* it, across crashes. Otherwise `prompt()` stores the prompt first.
|
||||
* it, across crashes. `prompt()` stores the prompt first either way.
|
||||
* - `pollMs`: how often a waiting worker reads the session again.
|
||||
* - `maxDurationMs` and `reserveMs`: each delivery stops `reserveMs` before
|
||||
* `maxDurationMs`, or before the World's `getRuntimeDeadline()`.
|
||||
* - `gatewayKey()`: the AI Gateway credential when the app passes none.
|
||||
* - `store`: names the store this World keeps sessions in; default its name.
|
||||
* Queue messages name it, so a route delivers each to the agents on its
|
||||
* store, and on one store an agent name belongs to one configuration. Give
|
||||
* each customer of an app a store of its own.
|
||||
*/
|
||||
export function world(source, options = {}) {
|
||||
const owned = typeof source === "function";
|
||||
if (!owned && !(source && typeof source === "object")) throw new TypeError("world() takes a World or a function that creates one");
|
||||
if (!options || typeof options !== "object" || Array.isArray(options)) throw new TypeError("world() options must be an object");
|
||||
const { name = "world", livenessKnown = false, queueLeases = false, runInRequest = queueLeases, waitUntil, queueDurable = false, pollMs, leaseMs, maxDurationMs, reserveMs, alive, holderInfo, gatewayKey } = options;
|
||||
const { name = "world", store = name, livenessKnown = false, queueLeases = false, runInRequest = queueLeases, waitUntil, queueDurable = false, pollMs, leaseMs, maxDurationMs, reserveMs, alive, holderInfo, gatewayKey } = options;
|
||||
if (typeof name !== "string" || name.length === 0) throw new TypeError("world() name must be a non-empty string");
|
||||
if (typeof store !== "string" || store.length === 0) throw new TypeError(`${name}() store must be a non-empty string`);
|
||||
for (const [key, value] of Object.entries({ livenessKnown, queueLeases, runInRequest, queueDurable })) {
|
||||
if (typeof value !== "boolean") throw new TypeError(`${name}() ${key} must be a boolean`);
|
||||
}
|
||||
@@ -63,6 +68,7 @@ export function world(source, options = {}) {
|
||||
return {
|
||||
[durabilityTag]: true,
|
||||
name,
|
||||
store,
|
||||
world: owned ? source : () => source,
|
||||
owned,
|
||||
livenessKnown,
|
||||
|
||||
@@ -1427,6 +1427,197 @@ if (durabilityKind === "queue") {
|
||||
});
|
||||
}
|
||||
|
||||
if (durabilityKind === "memory") {
|
||||
test("one name on one store is one agent: another configuration under it is refused", async () => {
|
||||
const durability = memory();
|
||||
const support = createTaggedAgent({ durability, name: "support", tag: "acme" });
|
||||
assert.throws(() => createTaggedAgent({ durability, name: "support", tag: "evil" }), (error) => error?.code === "FX_AGENT_NAME_IN_USE" && error.agent === "support");
|
||||
// A copy with the same configuration is the same agent, as one on each
|
||||
// server is, and only watching it differently does not change it.
|
||||
const copy = createTaggedAgent({ durability, name: "support", tag: "acme", onEvent: () => {} });
|
||||
// Another store, or another name, holds other sessions.
|
||||
const elsewhere = createTaggedAgent({ durability: memory(), name: "support", tag: "evil" });
|
||||
const named = createTaggedAgent({ durability, name: "billing", tag: "evil" });
|
||||
await copy.close();
|
||||
await support.close();
|
||||
// Once the first configuration is gone, the name is free again.
|
||||
const later = createTaggedAgent({ durability, name: "support", tag: "evil" });
|
||||
for (const agent of [elsewhere, named, later]) await agent.close();
|
||||
});
|
||||
|
||||
test("agents created without a name or a durability are one agent only when configured alike", async () => {
|
||||
const tagged = createDurableAgentFactory({ harness: ({ tag }) => scriptedHarness()(), defaultDurability: async () => memory(), name: "createDefaultAgent" });
|
||||
const first = tagged({ tag: "key-a" });
|
||||
assert.throws(() => tagged({ tag: "key-b" }), (error) => error?.code === "FX_AGENT_NAME_IN_USE" && error.agent === "default");
|
||||
const same = tagged({ tag: "key-a" });
|
||||
await same.close();
|
||||
await first.close();
|
||||
});
|
||||
|
||||
test("an input over the size limit is refused before anything is stored", async () => {
|
||||
const agent = createAgent({ durability: memory() });
|
||||
const session = agent.session(undefined, { context: { user: "a" } });
|
||||
assert.equal((await session.prompt("first").result).stopReason, "end_turn");
|
||||
const inputs = async () => (await (await agent[Symbol.for("libfx.durableInternals")].openLog(session.id)).read()).filter(({ entry }) => entry.k === "input").length;
|
||||
const before = await inputs();
|
||||
const big = "x".repeat(8 * 1024 * 1024);
|
||||
assert.throws(() => session.prompt(big), (error) => error?.code === "FX_INPUT_TOO_LARGE" && error.limit === 8 * 1024 * 1024);
|
||||
await assert.rejects(session.steer(big), (error) => error?.code === "FX_INPUT_TOO_LARGE");
|
||||
assert.equal(await inputs(), before, "no input was stored");
|
||||
await agent.close();
|
||||
});
|
||||
}
|
||||
|
||||
if (durabilityKind === "queue") {
|
||||
// A World that delivers only over HTTP, to the route the test posts to.
|
||||
const routeWorld = async (store) => {
|
||||
const { createWorld } = await import(new URL("../durable/node_modules/@workflow/world-local/dist/index.js", import.meta.url).href);
|
||||
const dir = await mkdtemp(join(tmpdir(), "libfx-core-"));
|
||||
dirs.push(dir);
|
||||
const stored = createWorld({ dataDir: dir, recoverActiveRuns: false });
|
||||
await stored.start();
|
||||
const sent = [];
|
||||
const instance = {
|
||||
...stored,
|
||||
registerHandler: undefined,
|
||||
queue: async (queueName, payload) => { sent.push({ queueName, payload }); return { messageId: `msg_${sent.length}` }; },
|
||||
createQueueHandler: (_prefix, handler) => async (request) => {
|
||||
const { payload, meta } = await request.json();
|
||||
try {
|
||||
await handler(payload, meta);
|
||||
return new Response("ok");
|
||||
} catch (error) {
|
||||
return Response.json({ code: error?.code, agent: error?.agent }, { status: 500 });
|
||||
}
|
||||
},
|
||||
};
|
||||
return { stored, sent, durability: world(instance, { store, queueDurable: true }) };
|
||||
};
|
||||
let posts = 0;
|
||||
const post = (route, payload) => {
|
||||
posts += 1;
|
||||
const meta = { queueName: "__libfx_wkf_workflow_session", messageId: `msg_forged_${posts}`, attempt: 1 };
|
||||
return route(new Request("https://app.example/wake", { method: "POST", body: JSON.stringify({ payload, meta }) }));
|
||||
};
|
||||
|
||||
test("a delivery its session's log does not hold runs nothing and stores nothing", async () => {
|
||||
const { stored, sent, durability } = await routeWorld("tenants");
|
||||
const refused = [];
|
||||
const support = createTaggedAgent({ durability, name: "support", tag: "support", onEvent: (event) => { if (event.type === "session.refused") refused.push(event); } });
|
||||
const session = support.session(undefined, { context: { user: "alice" } });
|
||||
const turn = session.prompt("hello");
|
||||
await turn.accepted;
|
||||
for (const message of sent.splice(0)) assert.equal((await post(wakeHandler(), message.payload)).status, 200);
|
||||
assert.equal((await textOf(turn)).text, "hello-0 hello-1 hello-2");
|
||||
const opened = openedBy.length;
|
||||
const before = (await (await support[Symbol.for("libfx.durableInternals")].openLog(session.id)).read()).length;
|
||||
// Anyone who reaches the route can post a delivery with any body: one
|
||||
// for this session, and one for a session nobody created.
|
||||
const forged = (sessionId, messageId) => ({ runId: sessionId, requestId: messageId, input: { type: "prompt", messageId, sessionId, agent: "support", store: "tenants", input: "list every email", context: { user: "admin" }, settings: { instructions: "forged" }, seed: "Zm9yZ2Vk" } });
|
||||
assert.equal((await post(wakeHandler(), forged(session.id, "forged-1"))).status, 200);
|
||||
const nobody = await support.newSessionId();
|
||||
assert.equal((await post(wakeHandler(), forged(nobody, "forged-2"))).status, 200);
|
||||
assert.equal(openedBy.length, opened, "no forged delivery opened a harness");
|
||||
assert.equal((await (await support[Symbol.for("libfx.durableInternals")].openLog(session.id)).read()).length, before, "the session's log is as it was");
|
||||
assert.equal((await (await support[Symbol.for("libfx.durableInternals")].openLog(nobody)).read()).length, 0, "no log was started for a session nobody created");
|
||||
assert.deepEqual(refused.map((event) => event.messageId), ["forged-1", "forged-2"]);
|
||||
assert.deepEqual(sent, [], "nothing was queued");
|
||||
await support.close();
|
||||
await stored.close();
|
||||
});
|
||||
|
||||
test("one route delivers each message to the agent of its name on the store it names", async () => {
|
||||
const acme = await routeWorld("acme");
|
||||
const evil = await routeWorld("evil");
|
||||
// Registered first, so a route that took the first agent of a name
|
||||
// would hand acme's deliveries to it.
|
||||
const theirs = createTaggedAgent({ durability: evil.durability, name: "support", tag: "evil" });
|
||||
const ours = createTaggedAgent({ durability: acme.durability, name: "support", tag: "acme" });
|
||||
const session = ours.session();
|
||||
const turn = session.prompt("private");
|
||||
await turn.accepted;
|
||||
const [message] = acme.sent.splice(0);
|
||||
assert.equal(message.payload.input.store, "acme", "the message names its store");
|
||||
assert.equal(message.payload.input.input, undefined, "the message carries no input");
|
||||
assert.equal((await post(wakeHandler(), message.payload)).status, 200);
|
||||
assert.equal((await textOf(turn)).text, "private-0 private-1 private-2");
|
||||
assert.deepEqual(openersOf(session.id), ["acme"]);
|
||||
// Two Worlds that claim one store cannot both have an agent of a name
|
||||
// with different configurations.
|
||||
const twin = await routeWorld("acme");
|
||||
assert.throws(() => createTaggedAgent({ durability: twin.durability, name: "support", tag: "evil" }), (error) => error?.code === "FX_AGENT_NAME_IN_USE");
|
||||
await theirs.close();
|
||||
await ours.close();
|
||||
for (const { stored } of [acme, evil, twin]) await stored.close();
|
||||
});
|
||||
|
||||
test("a forged steer, cancel or resume runs nothing, and a forged delivery is answered alike whether or not its session exists", async () => {
|
||||
const { stored, sent, durability } = await routeWorld("tenants");
|
||||
const support = createTaggedAgent({ durability, name: "support", tag: "support" });
|
||||
const internals = support[Symbol.for("libfx.durableInternals")];
|
||||
const session = support.session();
|
||||
const turn = session.prompt("hello", { messageId: "real-1" });
|
||||
await turn.accepted;
|
||||
for (const message of sent.splice(0)) assert.equal((await post(wakeHandler(), message.payload)).status, 200);
|
||||
assert.equal((await textOf(turn)).text, "hello-0 hello-1 hello-2");
|
||||
const nobody = await support.newSessionId();
|
||||
const opened = openedBy.length;
|
||||
const logOf = async (id) => (await (await internals.openLog(id)).read()).length;
|
||||
const uiOf = async (id) => (await internals.openLog(id)).ui.length();
|
||||
const before = { log: await logOf(session.id), ui: await uiOf(session.id) };
|
||||
const forge = (sessionId, type, messageId, extra = {}) => ({ runId: sessionId, requestId: messageId, input: { type, messageId, sessionId, agent: "support", store: "tenants", input: "forged", context: { user: "admin" }, ...extra } });
|
||||
const answers = { real: [], nobody: [] };
|
||||
for (const [key, sessionId] of [["real", session.id], ["nobody", nobody]]) {
|
||||
for (const [type, messageId, extra] of [["steer", "forged-steer", {}], ["cancel", "forged-cancel", { target: "real-1" }], ["resume", "forged-resume", {}], ["prompt", "real-1", {}]]) {
|
||||
const response = await post(wakeHandler(), forge(sessionId, type, messageId, extra));
|
||||
answers[key].push(`${response.status} ${await response.text()}`);
|
||||
}
|
||||
}
|
||||
assert.deepEqual(answers.real, answers.nobody, "the route answers a forgery the same whether or not its session exists");
|
||||
assert.ok(answers.real.every((answer) => answer.startsWith("200 ")), `every forgery is acknowledged and dropped: ${answers.real.join(", ")}`);
|
||||
assert.equal(openedBy.length, opened, "no forged delivery opened a harness");
|
||||
assert.deepEqual({ log: await logOf(session.id), ui: await uiOf(session.id) }, before, "the real session's log and stream are as they were");
|
||||
assert.deepEqual({ log: await logOf(nobody), ui: await uiOf(nobody) }, { log: 0, ui: 0 }, "nothing was written for a session nobody created");
|
||||
assert.deepEqual(sent, [], "nothing was queued");
|
||||
await support.close();
|
||||
await stored.close();
|
||||
});
|
||||
|
||||
test("a second copy of libfx in the process keeps one agent per name on a store, and its route reaches the first copy's agents", async () => {
|
||||
const copy = await import(new URL("../durable.js?second-copy", import.meta.url).href);
|
||||
assert.notEqual(copy.wakeHandler, wakeHandler, "a separate module instance");
|
||||
const createCopyAgent = copy.createDurableAgentFactory({ harness: ({ tag }) => scriptedHarness()(), defaultDurability: async () => memory(), name: "createCopyAgent" });
|
||||
const { stored, sent, durability } = await routeWorld("tenants");
|
||||
const support = createTaggedAgent({ durability, name: "support", tag: "acme" });
|
||||
assert.throws(() => createCopyAgent({ durability, name: "support", tag: "evil" }), (error) => error?.code === "FX_AGENT_NAME_IN_USE");
|
||||
const session = support.session();
|
||||
const turn = session.prompt("across copies");
|
||||
await turn.accepted;
|
||||
for (const message of sent.splice(0)) assert.equal((await post(copy.wakeHandler(), message.payload)).status, 200);
|
||||
assert.equal((await textOf(turn)).text, "across copies-0 across copies-1 across copies-2");
|
||||
assert.deepEqual(openersOf(session.id), ["acme"]);
|
||||
await support.close();
|
||||
await stored.close();
|
||||
});
|
||||
|
||||
test("a delivery for an agent that has closed fails, and no agent of its name on another store takes it", async () => {
|
||||
const acme = await routeWorld("acme");
|
||||
const carl = await routeWorld("carl");
|
||||
const theirs = createTaggedAgent({ durability: carl.durability, name: "support", tag: "carl" });
|
||||
const ours = createTaggedAgent({ durability: acme.durability, name: "support", tag: "acme" });
|
||||
const session = ours.session();
|
||||
await session.prompt("after close").accepted;
|
||||
const [message] = acme.sent.splice(0);
|
||||
await ours.close();
|
||||
const response = await post(wakeHandler(), message.payload);
|
||||
assert.equal(response.status, 500, "the queue keeps the message to deliver again");
|
||||
assert.deepEqual(await response.json(), { code: "FX_NO_AGENT", agent: "support" });
|
||||
assert.deepEqual(openersOf(session.id), [], "no agent ran the closed agent's session");
|
||||
await theirs.close();
|
||||
for (const { stored } of [acme, carl]) await stored.close();
|
||||
});
|
||||
}
|
||||
|
||||
let failed = 0;
|
||||
for (const [name, fn] of tests) {
|
||||
const started = performance.now();
|
||||
|
||||
+24
-37
@@ -673,18 +673,13 @@ if (durabilityKind === "memory") {
|
||||
await second.close();
|
||||
});
|
||||
|
||||
// A sender whose queue outlives the process does not read the log first,
|
||||
// so the worker reports a steer that reuses a prompt's id.
|
||||
test("on a queue that outlives the process, a steer reusing a prompt's id is reported and runs nothing", async () => {
|
||||
const errors = [];
|
||||
const agent = createFxAgent(agentOptions(queueOutlivesProcess(await durabilityFor()), {
|
||||
onEvent: (event) => { if (event.type === "session.error") errors.push(event); },
|
||||
}));
|
||||
// Every sender stores its input before it queues the message, whatever
|
||||
// the queue, so it refuses a steer that reuses a prompt's id itself.
|
||||
test("on a queue that outlives the process, a steer reusing a prompt's id is refused and runs nothing", async () => {
|
||||
const agent = createFxAgent(agentOptions(queueOutlivesProcess(await durabilityFor())));
|
||||
const session = agent.session();
|
||||
assert.equal((await session.prompt("hello there", { messageId: "m-dup" }).result).stopReason, "end_turn");
|
||||
await session.steer("again", { messageId: "m-dup" });
|
||||
await until(() => errors.length > 0, "the collision's session.error");
|
||||
assert.equal(errors[0].error?.code, "FX_MESSAGE_ID_IN_USE");
|
||||
await assert.rejects(session.steer("again", { messageId: "m-dup" }), (error) => error?.code === "FX_MESSAGE_ID_IN_USE");
|
||||
const lines = await readLines(session.stream(0), 64);
|
||||
assert.equal(lines.filter((line) => line.type === "turn_start").length, 1, "the steer ran no turn");
|
||||
await agent.close();
|
||||
@@ -748,17 +743,16 @@ if (durabilityKind === "memory") {
|
||||
await agent.close();
|
||||
});
|
||||
|
||||
// The redelivery reads the log before the first delivery's input lands, so
|
||||
// it writes the input again.
|
||||
// A redelivery names the input its sender stored and carries none, so the
|
||||
// steer is in the log once however often it is delivered.
|
||||
test("on a queue that outlives the process, a steer that starts a turn and is delivered again runs once", async () => {
|
||||
const [held, writes] = holdWrites(queueOutlivesProcess(await durabilityFor()), (entry) => entry.k === "input");
|
||||
const [held, writes] = holdWrites(queueOutlivesProcess(await durabilityFor()), (entry) => entry.k === "lease");
|
||||
const agent = createFxAgent(agentOptions(held));
|
||||
const session = agent.session();
|
||||
const before = requests.length;
|
||||
await session.steer("guide me again", { messageId: "m-redelivered" });
|
||||
await writes.next();
|
||||
void agent[Symbol.for("libfx.durableInternals")].deliver({ type: "steer", messageId: "m-redelivered", input: "guide me again", agent: "default", sessionId: session.id });
|
||||
await until(() => writes.held.length === 2, "the redelivery's copy of the input");
|
||||
void agent[Symbol.for("libfx.durableInternals")].deliver({ type: "steer", messageId: "m-redelivered", input: "a body the worker ignores", agent: "default", sessionId: session.id });
|
||||
await writes.release();
|
||||
const lines = await linesUntil(session, (line) => line.type === "turn_end" && line.messageId === "m-redelivered", "the steer's turn");
|
||||
assert.equal(lines.find((line) => line.type === "turn_end" && line.messageId === "m-redelivered").stopReason, "end_turn");
|
||||
@@ -804,33 +798,26 @@ if (durabilityKind === "memory") {
|
||||
await agent.close();
|
||||
});
|
||||
|
||||
// With a queue that outlives the process, as on Vercel, a message's input
|
||||
// is written with its claim's lease, so the lease can land first.
|
||||
test("a prompt whose input landed after another worker took the session over still runs", async () => {
|
||||
const base = queueOutlivesProcess(await durabilityFor());
|
||||
const [held, writes] = holdWrites(base, (entry) => entry.k === "input");
|
||||
const first = createFxAgent(agentOptions(held));
|
||||
const turn = first.session().prompt("one");
|
||||
const { sessionId } = await turn.accepted;
|
||||
// With a queue that outlives the process, as on Vercel, a prompt's input
|
||||
// still reaches the log before its message is queued, so no delivery can
|
||||
// run ahead of it, and its caller hears it was accepted only once stored.
|
||||
test("on a queue that outlives the process, a prompt is accepted and queued only once its input is stored", async () => {
|
||||
const [held, writes] = holdWrites(queueOutlivesProcess(await durabilityFor()), (entry) => entry.k === "input");
|
||||
const agent = createFxAgent(agentOptions(held));
|
||||
const before = requests.length;
|
||||
const turn = agent.session().prompt("one");
|
||||
let accepted = false;
|
||||
void turn.accepted.then(() => { accepted = true; });
|
||||
await writes.next();
|
||||
await until(async () => (await first[Symbol.for("libfx.durableInternals")].lastLease(sessionId)) !== null, "the first worker's lease");
|
||||
// The platform gives up on the first worker, and another takes the
|
||||
// session over for a prompt of its own and finishes it.
|
||||
const holders = globalThis[Symbol.for("libfx.liveHolders")];
|
||||
for (const holder of [...holders]) holders.delete(holder);
|
||||
const second = createFxAgent(agentOptions(base));
|
||||
assert.equal((await collect(second.session(sessionId).prompt("two"))).result.stopReason, "end_turn");
|
||||
// Closing waits for its delivery to finish, so it cannot take the first
|
||||
// prompt as well.
|
||||
await second.close();
|
||||
// The first prompt lands only now; its worker reads the log again and
|
||||
// runs it.
|
||||
await new Promise((resolve) => setTimeout(resolve, 200 * timeScale));
|
||||
assert.equal(accepted, false, "not accepted while its input is held");
|
||||
assert.equal(requests.length, before, "nothing ran while its input is held");
|
||||
writes.release();
|
||||
let timer;
|
||||
const stranded = new Promise((_, fail) => { timer = setTimeout(() => fail(new Error("the first prompt never ran")), 10_000 * timeScale); });
|
||||
const stranded = new Promise((_, fail) => { timer = setTimeout(() => fail(new Error("the prompt never ran")), 10_000 * timeScale); });
|
||||
const ran = await Promise.race([collect(turn), stranded]).finally(() => clearTimeout(timer));
|
||||
assert.equal(ran.text, "echo: one");
|
||||
await first.close();
|
||||
await agent.close();
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
#!/usr/bin/env node
|
||||
// The native addon reads the environment from libc at each use. Node.js owns
|
||||
// the environment, and an app that adds variables after the first native
|
||||
// runtime started makes libc move its array and free the old one; an addon
|
||||
// that kept the old array would read freed memory on its next lookup, which
|
||||
// crashes the process on macOS.
|
||||
//
|
||||
// node sdk/tests/test-native-core-env-growth.mjs [libfx.node]
|
||||
import { strict as assert } from "node:assert";
|
||||
import { spawnSync } from "node:child_process";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
const scriptDir = fileURLToPath(new URL(".", import.meta.url));
|
||||
const addon = resolve(process.argv[2] && process.argv[2] !== "child" ? process.argv[2] : resolve(scriptDir, "../../zig-out/lib/libfx.node"));
|
||||
|
||||
if (process.argv.includes("child")) {
|
||||
const names = Number(process.argv.at(-1));
|
||||
const { createFxEngine } = await import("../node.js");
|
||||
// A variable written before the addon loads, as an app's dotenv does,
|
||||
// leaves libc's array on the heap, where the next growth can free it.
|
||||
process.env.LIBFX_ENV_GROWTH_BEFORE = "1";
|
||||
const home = await mkdtemp(join(tmpdir(), "libfx-env-home-"));
|
||||
const workspaceRoot = await mkdtemp(join(tmpdir(), "libfx-env-workspace-"));
|
||||
const stream = [
|
||||
'data: {"type":"text-delta","delta":"ok"}\n\n',
|
||||
'data: {"type":"finish","finishReason":{"unified":"stop","raw":"stop"},"usage":{"inputTokens":{"total":1},"outputTokens":{"total":1}},"providerMetadata":{"gateway":{"generationId":"gen_01ARZ3NDEKTSV4RRFFQ69G5FAV"}}}\n\n',
|
||||
"data: [DONE]\n\n",
|
||||
].join("");
|
||||
const engine = await createFxEngine({
|
||||
nativeAddon: addon,
|
||||
backend: "native",
|
||||
fetch: async (_input, init) => (init.method === "GET"
|
||||
? Response.json({ object: "list", data: [{ id: "native/test-model", type: "language" }] })
|
||||
: new Response(stream, { headers: { "content-type": "text/event-stream" } })),
|
||||
home,
|
||||
workspaceRoot,
|
||||
apiKey: "env-growth-key",
|
||||
gatewayChatUrl: "http://127.0.0.1:9/chat",
|
||||
model: "native/test-model",
|
||||
});
|
||||
for (let index = 0; index < names; index += 1) process.env[`LIBFX_ENV_GROWTH_${index}`] = "x".repeat(32);
|
||||
for (let round = 0; round < 3; round += 1) {
|
||||
const turn = engine.prompt(`round ${round}`);
|
||||
for await (const _ of turn) {}
|
||||
assert.equal((await turn.result).stopReason, "end_turn");
|
||||
}
|
||||
await engine.close();
|
||||
await rm(home, { recursive: true, force: true });
|
||||
await rm(workspaceRoot, { recursive: true, force: true });
|
||||
process.exit(0);
|
||||
}
|
||||
|
||||
// Each run is a process of its own: a crash shows as its exit, not ours.
|
||||
for (const names of [0, 10, 200]) {
|
||||
for (let run = 0; run < 3; run += 1) {
|
||||
const child = spawnSync(process.execPath, [fileURLToPath(import.meta.url), addon, "child", String(names)], {
|
||||
encoding: "utf8",
|
||||
env: { PATH: process.env.PATH, HOME: process.env.HOME ?? tmpdir() },
|
||||
timeout: 30_000,
|
||||
});
|
||||
assert.equal(child.signal, null, `${names} new variables, run ${run}: the process died of ${child.signal}\n${child.stderr}`);
|
||||
assert.equal(child.status, 0, `${names} new variables, run ${run}: exit ${child.status}\n${child.stderr}`);
|
||||
}
|
||||
}
|
||||
console.log("native environment growth passed: 0, 10 and 200 new variables after the first runtime, 3 runs each");
|
||||
@@ -14,6 +14,7 @@ const scripts = [
|
||||
"test-native-core-exit-code.mjs",
|
||||
"test-native-core.mjs",
|
||||
"test-native-core-config-isolation.mjs",
|
||||
"test-native-core-env-growth.mjs",
|
||||
"test-agent-request-context.mjs",
|
||||
"test-agent-transport-retry.mjs",
|
||||
"test-agent-codex.mjs",
|
||||
|
||||
Vendored
+7
-1
@@ -41,7 +41,13 @@ export type WorldSource = World | (() => World | Promise<World>);
|
||||
export type WorldOptions = {
|
||||
/** The durability's name in errors. Default `"world"`. */
|
||||
name?: string;
|
||||
/** The queue keeps a message until a delivery acknowledges it, across crashes. When false, `prompt()` stores the prompt first. Default false. */
|
||||
/**
|
||||
* The store this World keeps sessions in. Queue messages name it, so one route hands each delivery to the agents on
|
||||
* its store, and on one store an agent name belongs to one configuration. Give each customer of an app a store of its
|
||||
* own. Default the durability's `name`.
|
||||
*/
|
||||
store?: string;
|
||||
/** The queue keeps a message until a delivery acknowledges it, across crashes. `prompt()` stores the prompt first either way. Default false. */
|
||||
queueDurable?: boolean;
|
||||
/** How often a waiting worker reads the session again. Default 1000. */
|
||||
pollMs?: number;
|
||||
|
||||
@@ -320,6 +320,8 @@ export type FxSessionEvent = FxEventBase & (
|
||||
| { type: "session.fenced"; sessionId: string; epoch: number }
|
||||
/** A message for a session of the agent `agent` reached this one, which sent it on. */
|
||||
| { type: "session.forwarded"; sessionId: string; agent: string }
|
||||
/** A queue delivery named an input the session's log does not hold, or none at all, so it ran nothing. */
|
||||
| { type: "session.refused"; sessionId?: string; messageId?: string; reason: "unknown message" | "malformed message" }
|
||||
/** `cutoffs`: the times in a row the deadline cut off the same step, this one included. */
|
||||
| { type: "session.deadline"; sessionId: string; epoch: number; cutoffs: number }
|
||||
| { type: "ui.error"; sessionId: string; error: string }
|
||||
@@ -772,8 +774,7 @@ export interface FxPromptOptions {
|
||||
export interface FxSteerOptions {
|
||||
/**
|
||||
* The steer's id, so a retried request adds nothing. Follows the session id rule. Reusing a prompt's id rejects
|
||||
* with `FX_MESSAGE_ID_IN_USE`; on a queue that outlives the process, the worker drops the steer and emits a
|
||||
* `session.error` with that code instead.
|
||||
* with `FX_MESSAGE_ID_IN_USE`.
|
||||
*/
|
||||
messageId?: string;
|
||||
}
|
||||
|
||||
+86
-9
@@ -14,6 +14,10 @@ var fallback_threaded: std.Io.Threaded = .init_single_threaded;
|
||||
var global_environ: ?*const std.process.Environ.Map = null;
|
||||
var global_environ_block: ?std.process.Environ.Block = null;
|
||||
var global_raw_environ: ?RawEnviron = null;
|
||||
// Set when another program owns the process environment and may change it,
|
||||
// as Node.js does for the N-API addon: the environment is read from libc at
|
||||
// each use and no pointer to it is kept.
|
||||
var global_libc_environ: bool = false;
|
||||
|
||||
pub fn setIo(zio: std.Io) void {
|
||||
real_io = process_io_for(builtin.target.os.tag, zio);
|
||||
@@ -398,27 +402,56 @@ pub fn setEnvironMap(m: *const std.process.Environ.Map) void {
|
||||
global_environ = m;
|
||||
global_environ_block = null;
|
||||
global_raw_environ = null;
|
||||
global_libc_environ = false;
|
||||
}
|
||||
|
||||
pub fn setEnvironBlock(block: std.process.Environ.Block) void {
|
||||
global_environ = null;
|
||||
global_environ_block = block;
|
||||
global_raw_environ = null;
|
||||
global_libc_environ = false;
|
||||
}
|
||||
|
||||
/// Installs the environment `main` received. libc never frees that array,
|
||||
/// so it stays readable even after the process changes its environment.
|
||||
pub fn setRawEnviron(raw: RawEnviron) void {
|
||||
global_environ = null;
|
||||
global_environ_block = null;
|
||||
global_raw_environ = raw;
|
||||
global_libc_environ = false;
|
||||
}
|
||||
|
||||
/// Reads the environment from libc at each use, for a library loaded into a
|
||||
/// process whose host can change the environment at any time. Growing the
|
||||
/// environment lets libc move its array and free the old one, so no pointer
|
||||
/// to it is kept between uses.
|
||||
pub fn setLibcEnviron() void {
|
||||
global_environ = null;
|
||||
global_environ_block = null;
|
||||
global_raw_environ = null;
|
||||
global_libc_environ = true;
|
||||
}
|
||||
|
||||
pub fn getenv(key: []const u8) ?[]const u8 {
|
||||
if (global_environ) |m| return m.get(key);
|
||||
if (global_environ_block) |block| return getenvFromBlock(block, key);
|
||||
if (global_raw_environ) |raw| return getenvFromLibc(key) orelse getenvFromRaw(raw, key);
|
||||
if (global_libc_environ) {
|
||||
if (getenvFromLibc(key)) |value| return value;
|
||||
// getenvFromLibc copies the key into a 128-byte buffer.
|
||||
if (key.len < 128) return null;
|
||||
return getenvFromRaw(liveLibcEnviron() orelse return null, key);
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
/// The environment array libc holds now, valid only until the process next
|
||||
/// changes its environment.
|
||||
fn liveLibcEnviron() ?RawEnviron {
|
||||
if (comptime !builtin.link_libc) return null;
|
||||
return @ptrCast(std.c.environ);
|
||||
}
|
||||
|
||||
/// The home directory fx keeps its own files under: `HOME` when it is an
|
||||
/// absolute path, otherwise null. An empty or relative `HOME` counts as
|
||||
/// unset, so fx never builds profile paths relative to the current folder.
|
||||
@@ -450,19 +483,23 @@ pub fn cloneEnvironMap(
|
||||
if (global_environ_block) |block| {
|
||||
return std.process.Environ.createMap(.{ .block = block }, alloc);
|
||||
}
|
||||
if (global_raw_environ) |raw| {
|
||||
var len: usize = 0;
|
||||
while (raw[len] != null) : (len += 1) {}
|
||||
const entries: []const [*:0]const u8 = @ptrCast(raw[0..len]);
|
||||
var map = std.process.Environ.Map.init(alloc);
|
||||
errdefer map.deinit();
|
||||
try map.putPosixBlock(.{ .slice = entries });
|
||||
return map;
|
||||
}
|
||||
if (global_raw_environ) |raw| return mapFromRaw(alloc, raw);
|
||||
if (global_libc_environ) return mapFromRaw(alloc, liveLibcEnviron() orelse return error.EnvironmentUnavailable);
|
||||
if (comptime builtin.is_test) return std.process.Environ.createMap(std.testing.environ, alloc);
|
||||
return error.EnvironmentUnavailable;
|
||||
}
|
||||
|
||||
/// A copy of `raw`, owned by the caller.
|
||||
fn mapFromRaw(alloc: std.mem.Allocator, raw: RawEnviron) CloneEnvironMapError!std.process.Environ.Map {
|
||||
var len: usize = 0;
|
||||
while (raw[len] != null) : (len += 1) {}
|
||||
const entries: []const [*:0]const u8 = @ptrCast(raw[0..len]);
|
||||
var map = std.process.Environ.Map.init(alloc);
|
||||
errdefer map.deinit();
|
||||
try map.putPosixBlock(.{ .slice = entries });
|
||||
return map;
|
||||
}
|
||||
|
||||
fn getenvFromBlock(block: std.process.Environ.Block, key: []const u8) ?[]const u8 {
|
||||
if (comptime builtin.target.os.tag == .windows or builtin.target.os.tag == .freestanding or builtin.target.os.tag == .other) {
|
||||
return null;
|
||||
@@ -1273,6 +1310,46 @@ test "cloneEnvironMap copies installed raw environment state" {
|
||||
try std.testing.expectEqualStrings("/raw/home", cloned.get("HOME").?);
|
||||
}
|
||||
|
||||
test "libc environment lookups follow a host that grows the environment" {
|
||||
if (comptime !builtin.link_libc) return error.SkipZigTest;
|
||||
const libc = struct {
|
||||
extern "c" fn setenv(name: [*:0]const u8, value: [*:0]const u8, overwrite: c_int) c_int;
|
||||
extern "c" fn unsetenv(name: [*:0]const u8) c_int;
|
||||
};
|
||||
const previous_map = global_environ;
|
||||
const previous_block = global_environ_block;
|
||||
const previous_raw = global_raw_environ;
|
||||
const previous_libc = global_libc_environ;
|
||||
defer {
|
||||
global_environ = previous_map;
|
||||
global_environ_block = previous_block;
|
||||
global_raw_environ = previous_raw;
|
||||
global_libc_environ = previous_libc;
|
||||
}
|
||||
setLibcEnviron();
|
||||
|
||||
const count = 200;
|
||||
var names: [count][32:0]u8 = undefined;
|
||||
var set: usize = 0;
|
||||
defer for (names[0..set]) |*name| {
|
||||
_ = libc.unsetenv(name);
|
||||
};
|
||||
// Enough new names that libc moves its array, as a host's writes do.
|
||||
while (set < count) : (set += 1) {
|
||||
const written = try std.mem.print(names[set][0..31], "FX_IO_LIVE_ENV_{d}", .{set});
|
||||
names[set][written.len] = 0;
|
||||
try std.testing.expectEqual(@as(c_int, 0), libc.setenv(&names[set], "live", 1));
|
||||
}
|
||||
|
||||
try std.testing.expectEqualStrings("live", getenv("FX_IO_LIVE_ENV_0").?);
|
||||
try std.testing.expectEqualStrings("live", getenv("FX_IO_LIVE_ENV_199").?);
|
||||
try std.testing.expect(getenv("FX_IO_LIVE_ENV_MISSING") == null);
|
||||
|
||||
var cloned = try cloneEnvironMap(std.testing.allocator);
|
||||
defer cloned.deinit();
|
||||
try std.testing.expectEqualStrings("live", cloned.get("FX_IO_LIVE_ENV_150").?);
|
||||
}
|
||||
|
||||
test "readFileToEnd: file under cap returns full content" {
|
||||
const alloc = std.testing.allocator;
|
||||
var tmp = std.testing.tmpDir(.{});
|
||||
|
||||
@@ -839,8 +839,9 @@ fn ensureThreadedIo() void {
|
||||
if (threaded_io_state.cmpxchgStrong(0, 1, .acq_rel, .acquire) == null) {
|
||||
threaded_io = std.Io.Threaded.init(std.heap.c_allocator, .{});
|
||||
io_mod.setIo(threaded_io.?.io());
|
||||
const raw_environ: io_mod.RawEnviron = @ptrCast(std.c.environ);
|
||||
io_mod.setRawEnviron(raw_environ);
|
||||
// Node.js owns the environment and may grow it at any time, which
|
||||
// can move and free libc's array, so it is read at each use.
|
||||
io_mod.setLibcEnviron();
|
||||
threaded_io_state.store(2, .release);
|
||||
return;
|
||||
}
|
||||
|
||||
+4
-3
@@ -10,12 +10,12 @@ TLA+ specifications for fx, checked with TLC, together with the scripts and Bun
|
||||
| `indexer/` | When a scanned root's listing may be reused, merging parallel grep results, the file picker's loading row, and listing snapshots (`src/core/indexer/`). | `indexer/check.sh` |
|
||||
| `mcp/` | The MCP-v2 engine (`src/mcp/`): wire format, requests, stdio, version detection, the tool catalog, both HTTP transports, input-required rounds, authorization, subscriptions, elicitation, and the engine. `rules.md` lists the MCP specification rules the models cite by ID. | `mcp/check.sh`, then the drivers in `drivers/mcp/` |
|
||||
| `mcp_host/` | fx's MCP host (`src/core/mcp_host/`): host lifecycle, saved credentials, logins shared between processes, sign-in, and elicitation routing. | `mcp_host/check.sh`, `bun test drivers/mcp_host/` |
|
||||
| `sdk/` | libfx durable sessions (`DurableSession.tla`), agent names and routing (`AgentRouting.tla`), steering (`DurableSteer.tla`), and the provider credential edge (`ProviderAuth.tla`). `check.sh` also writes the TLC state graphs the drivers walk. | `sdk/check.sh`, then the drivers |
|
||||
| `sdk/` | libfx durable sessions (`DurableSession.tla`), agent names and routing (`AgentRouting.tla`), tenants whose agents share one process (`TenantIsolation.tla`), steering (`DurableSteer.tla`), and the provider credential edge (`ProviderAuth.tla`). `check.sh` also writes the TLC state graphs the drivers walk. | `sdk/check.sh`, then the drivers |
|
||||
| `session_manager/` | The session log, turn lifecycle, writer lock, snapshots, fork, subagents, catalog, host wiring, the side-file move, and v1 conversion. `rollout/` holds the models of the changes that moved fx onto the session manager. `system-design.md` is the design record the specs check and defines the `D1` to `D62` decision numbers that code comments cite. | `session_manager/check.sh`, `session_manager/rollout/run.py`, `bun test drivers/session_manager/` |
|
||||
| `shell_command/` | How an agent git command is admitted and run (`src/core/shell_command/safe_git.zig`), and one model turn through the shell tool, for the current design and a redesign. | `shell_command/check.sh`, `bun test drivers/shell_command/` |
|
||||
| `terminal/` | The in-memory shell snapshot owner (`src/core/terminal/shell_snapshot.zig`) and the lifecycle of hosted terminals. | `terminal/check.sh` |
|
||||
| `usage/` | `/usage` refreshes and the shared usage cache (`src/usage/`): `LeanUsage.tla` and `UsageModel.tla`. | `usage/check.sh`, then the drivers in `drivers/usage/` |
|
||||
| `drivers/` | The Bun drivers. They run the real fx binary or SDK code, record what happens, and have TLC check each run against its model. `drive.mjs`, `drive-agents.mjs`, and `drive-steer.mjs` walk the SDK code along the `sdk/` state graphs. | See each folder's `check.sh` |
|
||||
| `drivers/` | The Bun drivers. They run the real fx binary or SDK code, record what happens, and have TLC check each run against its model. `drive.mjs`, `drive-agents.mjs`, `drive-tenants.mjs`, and `drive-steer.mjs` walk the SDK code along the `sdk/` state graphs. | See each folder's `check.sh` |
|
||||
| `lib/` | `tlc.sh`, which finds TLC and Java and runs a table of configs with their expected verdicts, and `run-tlc.sh`, which runs TLC once for the drivers. | |
|
||||
|
||||
## Requirements
|
||||
@@ -74,7 +74,8 @@ tests/tla/mcp/check.sh # every MCP-v2 model and
|
||||
tests/tla/mcp_host/check.sh # every MCP host model, planted bug, and finding
|
||||
tests/tla/sdk/check.sh # the SDK specs; writes the state graphs the drivers walk
|
||||
bun tests/tla/drivers/drive.mjs send --walks 40 # one durable-session graph; see .github/workflows/ci.yml for the rest
|
||||
bun tests/tla/drivers/drive-agents.mjs send --walks 60 # agent routing
|
||||
bun tests/tla/drivers/drive-agents.mjs --walks 60 # agent routing
|
||||
bun tests/tla/drivers/drive-tenants.mjs --walks 60 # tenants in one process
|
||||
bun tests/tla/drivers/drive-steer.mjs ideal --walks 20 # steering
|
||||
bun test ./tests/tla/drivers/shell_command/ # a driver written as bun tests; see each check.sh for its own
|
||||
```
|
||||
|
||||
@@ -9,7 +9,7 @@ import { parentPort, workerData } from "node:worker_threads";
|
||||
import { createDurableAgentFactory, foldSessionLog } from "../../../sdk/durable.js";
|
||||
import { world } from "../../../sdk/durable/world.mjs";
|
||||
|
||||
const { names, mode } = workerData;
|
||||
const { names } = workerData;
|
||||
const { createWorld } = await import(new URL("../../../sdk/durable/node_modules/@workflow/world-local/dist/index.js", import.meta.url).href);
|
||||
const encoder = new TextEncoder();
|
||||
const inner = Symbol("handler");
|
||||
@@ -80,7 +80,7 @@ const methods = {
|
||||
},
|
||||
};
|
||||
};
|
||||
const durability = world(createInstance, { name: "routing", pollMs: 250, queueDurable: mode === "consume", runInRequest: false });
|
||||
const durability = world(createInstance, { name: "routing", pollMs: 250, runInRequest: false });
|
||||
const agents = Object.fromEntries(names.map((name) => [name, createAgent({ durability, name, tag: name })]));
|
||||
walk = { handlers, agents };
|
||||
ran = [];
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
// AgentRouting.tla, and checks after every step that the code is where the
|
||||
// model allows.
|
||||
//
|
||||
// bun tests/tla/drivers/drive-agents.mjs [send|consume] [--walks N] [--depth N] [--seed S]
|
||||
// bun tests/tla/drivers/drive-agents.mjs [--walks N] [--depth N] [--seed S]
|
||||
//
|
||||
// Two processes, p1 and p2, are two worker threads (agent-process.mjs), each
|
||||
// with a module instance and a globalThis of its own: p1 has the agents S
|
||||
@@ -10,9 +10,8 @@
|
||||
// World of its own, as agents created without a durability have. The
|
||||
// queue they share is the driver's: it holds every message until the walk
|
||||
// delivers it to a process, to one of its Worlds picked at random, as the
|
||||
// graph's Send and Deliver steps say. With "send" a prompt's input is stored
|
||||
// before it is queued, as on local(); with "consume" the worker that takes
|
||||
// it stores it, as on a World whose queue keeps its messages (Vercel).
|
||||
// graph's Send and Deliver steps say. A prompt's input is stored before it
|
||||
// is queued, as on every durability.
|
||||
//
|
||||
// After each step the driver reads back each session's owner from its log,
|
||||
// the turns each agent ran, the names a process had no agent for, and the
|
||||
@@ -32,7 +31,6 @@ const option = (name, fallback) => {
|
||||
const index = args.indexOf(`--${name}`);
|
||||
return index >= 0 ? args[index + 1] : fallback;
|
||||
};
|
||||
const mode = args.find((arg) => arg === "send" || arg === "consume") ?? "send";
|
||||
const walks = Number(option("walks", 40));
|
||||
const depth = Number(option("depth", 24));
|
||||
let seed = Number(option("seed", Date.now() % 100000));
|
||||
@@ -45,7 +43,7 @@ const random = () => {
|
||||
const pick = (items) => items[Math.floor(random() * items.length)];
|
||||
|
||||
const here = fileURLToPath(new URL(".", import.meta.url));
|
||||
const graph = await loadGraph(resolve(here, "../sdk/graphs", `agents-${mode}.dot`));
|
||||
const graph = await loadGraph(resolve(here, "../sdk/graphs/agents-send.dot"));
|
||||
const hosts = { p1: ["S", "C"], p2: ["S"] };
|
||||
|
||||
// The queue the processes share: every message, until a step delivers it.
|
||||
@@ -63,7 +61,7 @@ function enqueue({ queueName, payload, idempotencyKey }) {
|
||||
|
||||
// A process: calls to its worker, which also sends to the queue.
|
||||
function startProcess(name) {
|
||||
const worker = new Worker(new URL("./agent-process.mjs", import.meta.url), { workerData: { names: hosts[name], mode } });
|
||||
const worker = new Worker(new URL("./agent-process.mjs", import.meta.url), { workerData: { names: hosts[name] } });
|
||||
const calls = new Map();
|
||||
let called = 0;
|
||||
worker.on("message", (message) => {
|
||||
@@ -197,5 +195,5 @@ try {
|
||||
for (const process of Object.values(processes)) await process.stop();
|
||||
}
|
||||
const edges = [...graph.next.values()].reduce((total, successors) => total + successors.size, 0);
|
||||
console.log(`${walks} walks (${mode}, seed ${firstSeed}), ${steps} steps (${Object.entries(outcomes).map(([name, count]) => `${count} ${name}`).join(", ")}): ${seenStates.size} of ${graph.states.size} states and ${seenEdges.size} of ${edges} edges, every step one the model allows`);
|
||||
console.log(`${walks} walks (seed ${firstSeed}), ${steps} steps (${Object.entries(outcomes).map(([name, count]) => `${count} ${name}`).join(", ")}): ${seenStates.size} of ${graph.states.size} states and ${seenEdges.size} of ${edges} edges, every step one the model allows`);
|
||||
assert.ok(outcomes.forward > 0 && outcomes.miss > 0, "the walks forward a message and miss an agent at least once");
|
||||
@@ -0,0 +1,383 @@
|
||||
#!/usr/bin/env bun
|
||||
// Drives real libfx durable agents along the TLC state graph of
|
||||
// TenantIsolation.tla, and checks after every step that the code is in a
|
||||
// state the model allows.
|
||||
//
|
||||
// bun tests/tla/drivers/drive-tenants.mjs [--walks N] [--depth N] [--seed S]
|
||||
//
|
||||
// `tests/tla/sdk/check.sh` model-checks the spec and writes
|
||||
// graphs/tenants-route.dot, the "four" topology with every rule on.
|
||||
//
|
||||
// This process is the shared function. Four tenants each create one agent
|
||||
// in it, as the model's Register steps say: acme and evil both name theirs
|
||||
// "support" on store key k1 but keep their sessions in different Worlds, bob
|
||||
// names his "billing" on acme's World, and carl names his "support" on store
|
||||
// key k2. Every World delivers only over HTTP, to one wakeHandler() route, as
|
||||
// vercel() does, and its queue is the driver's: it holds each message until a
|
||||
// Deliver step posts it to the route. A Forge step posts a delivery no queue
|
||||
// sent, as anyone who can reach the route can. The harness is scripted: its
|
||||
// turns record which tenant's agent ran them, in which World, with which
|
||||
// context, which is what the model checks.
|
||||
//
|
||||
// After each step the driver reads every session's log in every World, the
|
||||
// turns that ran, and what is queued, and keeps the model states that match.
|
||||
// A step that leaves none is a conformance failure.
|
||||
import { strict as assert } from "node:assert";
|
||||
import { mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join, resolve } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
import { createDurableAgentFactory, foldSessionLog, wakeHandler } from "../../../sdk/durable.js";
|
||||
import { world } from "../../../sdk/durable/world.mjs";
|
||||
import { loadGraph } from "./graph.mjs";
|
||||
|
||||
const args = process.argv.slice(2);
|
||||
const option = (name, fallback) => {
|
||||
const index = args.indexOf(`--${name}`);
|
||||
return index >= 0 ? args[index + 1] : fallback;
|
||||
};
|
||||
const walks = Number(option("walks", 40));
|
||||
const depth = Number(option("depth", 14));
|
||||
let seed = Number(option("seed", Date.now() % 100000));
|
||||
const firstSeed = seed;
|
||||
// A small deterministic generator, so a failing walk can be run again.
|
||||
const random = () => {
|
||||
seed = (seed * 1103515245 + 12345) % 2147483648;
|
||||
return seed / 2147483648;
|
||||
};
|
||||
const pick = (items) => items[Math.floor(random() * items.length)];
|
||||
|
||||
const here = fileURLToPath(new URL(".", import.meta.url));
|
||||
const { createWorld } = await import(new URL("../../../sdk/durable/node_modules/@workflow/world-local/dist/index.js", import.meta.url).href);
|
||||
const graph = await loadGraph(resolve(here, "../sdk/graphs/tenants-route.dot"));
|
||||
const durableInternals = Symbol.for("libfx.durableInternals");
|
||||
const encoder = new TextEncoder();
|
||||
|
||||
// The "four" topology, as TenantIsolation.tla defines it.
|
||||
const tenants = ["acme", "evil", "bob", "carl"];
|
||||
const nameOf = { acme: "support", evil: "support", bob: "billing", carl: "support" };
|
||||
const keyOf = { acme: "k1", evil: "k1", bob: "k1", carl: "k2" };
|
||||
const storeOf = { acme: "A", evil: "B", bob: "A", carl: "C" };
|
||||
const stores = ["A", "B", "C"];
|
||||
const keyOfStore = { A: "k1", B: "k1", C: "k2" };
|
||||
// Comfortably over the limit libfx admits for one input (sdk/durable.js
|
||||
// maxInputBytes).
|
||||
const oversizeBytes = 9 * 1024 * 1024;
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// The code under test.
|
||||
|
||||
// Every turn any agent ran, as the harness saw it.
|
||||
const turns = [];
|
||||
const createTenantAgent = createDurableAgentFactory({
|
||||
name: "createTenantAgent",
|
||||
defaultDurability: async () => { throw new Error("the driver passes a durability"); },
|
||||
harness: ({ tenant, storeOfDurability }) => ({
|
||||
async open({ sessionId, store, context, durability }) {
|
||||
await store.load();
|
||||
return {
|
||||
prompt(_input, { turnId }) {
|
||||
const result = (async () => {
|
||||
await store.append({ idempotencyKey: `${turnId}:start`, data: encoder.encode(JSON.stringify({ turnId, start: true })), marks: [{ start: turnId }] });
|
||||
turns.push({ tenant, store: storeOfDurability(durability), sessionId, turnId, context });
|
||||
await store.append({ idempotencyKey: `${turnId}:end`, data: encoder.encode(JSON.stringify({ turnId, end: true })), marks: [{ end: true }] });
|
||||
return { stopReason: "end_turn" };
|
||||
})();
|
||||
return { result, steer: async () => {}, cancel() {}, async *[Symbol.asyncIterator]() { await result; } };
|
||||
},
|
||||
resume() {
|
||||
throw new Error("no turn of this harness is ever left open");
|
||||
},
|
||||
get openTurn() {
|
||||
return null;
|
||||
},
|
||||
settled: async () => {},
|
||||
saveCheckpoint: async () => {},
|
||||
exportCheckpoint: async () => new Uint8Array(),
|
||||
close: async () => {},
|
||||
};
|
||||
},
|
||||
}),
|
||||
});
|
||||
|
||||
// The queue every World shares: each message, until a step delivers it.
|
||||
const held = [];
|
||||
const keys = new Set();
|
||||
let queued = 0;
|
||||
function enqueue(queueName, payload, idempotencyKey) {
|
||||
queued += 1;
|
||||
const messageId = `msg_${String(queued).padStart(6, "0")}`;
|
||||
if (idempotencyKey !== undefined && keys.has(idempotencyKey)) return { messageId };
|
||||
if (idempotencyKey !== undefined) keys.add(idempotencyKey);
|
||||
held.push({ messageId, queueName, payload, attempt: 0 });
|
||||
return { messageId };
|
||||
}
|
||||
|
||||
// A World over `dir` that delivers only to the route the driver posts to.
|
||||
function routeWorld(dir) {
|
||||
const stored = createWorld({ dataDir: dir, recoverActiveRuns: false });
|
||||
return {
|
||||
stored,
|
||||
instance: {
|
||||
...stored,
|
||||
registerHandler: undefined,
|
||||
queue: async (queueName, payload, { idempotencyKey } = {}) => enqueue(queueName, payload, idempotencyKey),
|
||||
createQueueHandler: (_prefix, handler) => async (request) => {
|
||||
const { payload, meta } = await request.json();
|
||||
try {
|
||||
await handler(payload, meta);
|
||||
return Response.json({ ok: true });
|
||||
} catch (error) {
|
||||
return Response.json({ code: error?.code ?? null, message: String(error?.message ?? error) }, { status: 500 });
|
||||
}
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
let posted = 0;
|
||||
async function post(payload) {
|
||||
posted += 1;
|
||||
const meta = { queueName: "__libfx_wkf_workflow_session", messageId: `msg_post_${posted}`, attempt: 1 };
|
||||
const response = await wakeHandler()(new Request("https://app.example/api/libfx", { method: "POST", body: JSON.stringify({ payload, meta }) }));
|
||||
return { status: response.status, body: await response.json().catch(() => null) };
|
||||
}
|
||||
|
||||
async function openWalk() {
|
||||
const base = await mkdtemp(join(tmpdir(), "libfx-tenants-"));
|
||||
const worlds = {};
|
||||
const durabilities = {};
|
||||
const labelOf = new Map();
|
||||
for (const st of stores) {
|
||||
worlds[st] = routeWorld(join(base, st));
|
||||
await worlds[st].stored.start?.();
|
||||
durabilities[st] = world(worlds[st].instance, { store: keyOfStore[st], queueDurable: true, runInRequest: false, pollMs: 250 });
|
||||
labelOf.set(durabilities[st], st);
|
||||
}
|
||||
const storeOfDurability = (durability) => labelOf.get(durability) ?? "?";
|
||||
// One reader per World, named apart from every tenant's agent.
|
||||
const observers = {};
|
||||
for (const st of stores) {
|
||||
observers[st] = createTenantAgent({ name: `observer-${st}`, durability: durabilities[st], tenant: `observer-${st}`, storeOfDurability });
|
||||
}
|
||||
held.length = 0;
|
||||
keys.clear();
|
||||
turns.length = 0;
|
||||
// Each tenant's session, a run id a World accepts.
|
||||
const sessionOf = {};
|
||||
for (const t of tenants) sessionOf[t] = await observers.A.newSessionId();
|
||||
const tenantOfSession = new Map(tenants.map((t) => [sessionOf[t], t]));
|
||||
const walk = { base, worlds, durabilities, storeOfDurability, observers, sessionOf, tenantOfSession, agents: {}, refused: new Set(), sent: [] };
|
||||
walk.close = async () => {
|
||||
for (const agent of [...Object.values(walk.agents), ...Object.values(observers)]) await agent.close().catch(() => {});
|
||||
for (const st of stores) await Promise.resolve(worlds[st].stored.close?.()).catch(() => {});
|
||||
await rm(base, { recursive: true, force: true });
|
||||
};
|
||||
return walk;
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// The code's state, in the model's terms.
|
||||
|
||||
const idOf = (messageId) => (messageId === "f0" ? 0 : /^m(\d+)$/.test(messageId ?? "") ? Number(messageId.slice(1)) : `?${messageId}`);
|
||||
const contextOf = (context) => (context === undefined || context === null ? "none" : typeof context.tenant === "string" ? context.tenant : "?");
|
||||
const sizeOf = (input) => (JSON.stringify(input ?? "").length > oversizeBytes - 1024 ? "big" : "ok");
|
||||
|
||||
async function observe(walk) {
|
||||
const { sessionOf, tenantOfSession } = walk;
|
||||
const log = {};
|
||||
const owner = {};
|
||||
const done = {};
|
||||
for (const st of stores) {
|
||||
log[st] = [];
|
||||
owner[st] = {};
|
||||
done[st] = [];
|
||||
const internals = walk.observers[st][durableInternals];
|
||||
for (const t of tenants) {
|
||||
const entries = await (await internals.openLog(sessionOf[t])).read();
|
||||
const folded = foldSessionLog(entries);
|
||||
owner[st][t] = folded.agent ?? "none";
|
||||
for (const { entry } of entries) {
|
||||
if (entry?.k !== "input" || (entry.type !== "prompt" && entry.type !== "steer")) continue;
|
||||
log[st].push({ id: idOf(entry.messageId), s: tenantOfSession.get(entry.sessionId ?? sessionOf[t]) ?? t, n: entry.agent ?? "none", ctx: contextOf(entry.context), size: sizeOf(entry.input) });
|
||||
}
|
||||
for (const id of folded.consumed) done[st].push(idOf(id));
|
||||
}
|
||||
}
|
||||
const queue = held.map(({ payload }) => ({ id: idOf(payload.input?.messageId), s: tenantOfSession.get(payload.input?.sessionId) ?? "?", n: payload.input?.agent ?? "none" }));
|
||||
const ran = turns.map((turn) => ({ id: idOf(turn.turnId), def: turn.tenant, store: turn.store, ctx: contextOf(turn.context) }));
|
||||
return { agents: Object.keys(walk.agents), refused: [...walk.refused], log, owner, done, queue, ran };
|
||||
}
|
||||
|
||||
// One canonical form for model and code alike: every collection in this
|
||||
// spec is a set, so arrays compare sorted.
|
||||
const canonical = (value) => {
|
||||
if (Array.isArray(value)) return `{${[...new Set(value.map(canonical))].sort().join(",")}}`;
|
||||
if (value && typeof value === "object") return `[${Object.keys(value).sort().map((key) => `${key}:${canonical(value[key])}`).join(",")}]`;
|
||||
return JSON.stringify(value);
|
||||
};
|
||||
const viewOf = (state) => canonical({
|
||||
agents: state.agents,
|
||||
refused: state.refused,
|
||||
log: state.log,
|
||||
owner: state.owner,
|
||||
done: state.done,
|
||||
queue: state.queue.map(({ id, s, n }) => ({ id, s, n })),
|
||||
ran: state.ran,
|
||||
});
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// The steps.
|
||||
|
||||
async function perform(walk, act) {
|
||||
const { sessionOf } = walk;
|
||||
switch (act.name) {
|
||||
case "Register": {
|
||||
const t = act.t;
|
||||
try {
|
||||
const agent = createTenantAgent({ name: nameOf[t], durability: walk.durabilities[storeOf[t]], tenant: t, storeOfDurability: walk.storeOfDurability });
|
||||
// Opens its backend, as its first request does.
|
||||
await agent.newSessionId();
|
||||
walk.agents[t] = agent;
|
||||
} catch (error) {
|
||||
if (error?.code !== "FX_AGENT_NAME_IN_USE") throw error;
|
||||
walk.refused.add(t);
|
||||
return "register refused";
|
||||
}
|
||||
return "register";
|
||||
}
|
||||
case "Prompt": {
|
||||
const session = walk.agents[act.t].session(sessionOf[act.t], { context: { tenant: act.t } });
|
||||
await session.prompt(`prompt ${act.id}`, { messageId: `m${act.id}` }).accepted;
|
||||
walk.sent.push(act.id);
|
||||
return "prompt";
|
||||
}
|
||||
case "Oversize": {
|
||||
const session = walk.agents[act.t].session(sessionOf[act.t], { context: { tenant: act.t } });
|
||||
const big = "x".repeat(oversizeBytes);
|
||||
try {
|
||||
if (act.k === "prompt") await session.prompt(big).accepted;
|
||||
else await session.steer(big);
|
||||
} catch (error) {
|
||||
if (error?.code !== "FX_INPUT_TOO_LARGE") throw error;
|
||||
return "oversize refused";
|
||||
}
|
||||
return "oversize stored";
|
||||
}
|
||||
case "Deliver": {
|
||||
const message = held.find(({ payload }) => payload.input?.messageId === `m${act.id}` && payload.input?.sessionId === sessionOf[act.s] &&
|
||||
payload.input?.agent === act.n && (payload.input?.store === undefined || payload.input.store === act.key));
|
||||
assert.ok(message, `message ${act.id} for ${act.s} on ${act.n}/${act.key} is queued`);
|
||||
message.attempt += 1;
|
||||
const { status, body } = await post(message.payload);
|
||||
if (status === 200) held.splice(held.indexOf(message), 1);
|
||||
else if (body?.code !== "FX_NO_AGENT") throw new Error(`delivery of ${act.id} failed: ${status} ${JSON.stringify(body)}`);
|
||||
return status === 200 ? "delivered" : "missed";
|
||||
}
|
||||
case "Forge": {
|
||||
const sessionId = sessionOf[act.s];
|
||||
const messageId = act.id === 0 ? "f0" : `m${act.id}`;
|
||||
const before = held.length;
|
||||
const ranBefore = turns.length;
|
||||
const { status } = await post({
|
||||
runId: sessionId,
|
||||
requestId: messageId,
|
||||
input: { type: "prompt", messageId, sessionId, agent: act.n, store: act.key, input: "forged", context: { tenant: "forger" }, settings: { instructions: "forged" } },
|
||||
});
|
||||
if (status !== 200) return "forge missed";
|
||||
if (held.length > before) return "forge forwarded";
|
||||
return turns.length > ranBefore ? "forge woke a sent prompt" : "forge refused";
|
||||
}
|
||||
default:
|
||||
throw new Error(`no step ${act.name}`);
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// The walks.
|
||||
|
||||
const sameAct = (a, b) => canonical(a) === canonical(b);
|
||||
const seenStates = new Set([graph.init]);
|
||||
const seenEdges = new Set();
|
||||
const outcomes = {};
|
||||
let steps = 0;
|
||||
|
||||
// The acts the frontier's states allow next, by name then variant, so a
|
||||
// name with many variants, such as Forge, does not crowd out the rest.
|
||||
function chooseAct(frontier, only) {
|
||||
const acts = new Map();
|
||||
for (const id of frontier) {
|
||||
for (const next of graph.next.get(id) ?? []) {
|
||||
const act = graph.states.get(next).act;
|
||||
if (only && !only(act, graph.states.get(id))) continue;
|
||||
acts.set(canonical(act), act);
|
||||
}
|
||||
}
|
||||
if (acts.size === 0) return null;
|
||||
const byName = new Map();
|
||||
for (const act of acts.values()) byName.set(act.name, [...(byName.get(act.name) ?? []), act]);
|
||||
const variants = pick([...byName.values()]);
|
||||
// Most forged deliveries name an agent that runs, so they test the
|
||||
// refusal rather than miss.
|
||||
if (variants[0].name === "Forge" && random() < 0.8) {
|
||||
const running = new Set([...frontier].flatMap((id) => graph.states.get(id).agents).map((t) => `${nameOf[t]}/${keyOf[t]}`));
|
||||
const aimed = variants.filter((act) => running.has(`${act.n}/${act.key}`));
|
||||
if (aimed.length > 0) return pick(aimed);
|
||||
}
|
||||
return pick(variants);
|
||||
}
|
||||
|
||||
async function walkOnce(number) {
|
||||
const walk = await openWalk();
|
||||
let frontier = new Set([graph.init]);
|
||||
const path = [];
|
||||
const step = async (act) => {
|
||||
path.push(canonical(act));
|
||||
const outcome = await perform(walk, act);
|
||||
outcomes[outcome] = (outcomes[outcome] ?? 0) + 1;
|
||||
const seen = await observe(walk);
|
||||
const view = viewOf(seen);
|
||||
const next = new Set();
|
||||
for (const from of frontier) {
|
||||
for (const to of graph.next.get(from) ?? []) {
|
||||
const state = graph.states.get(to);
|
||||
if (!sameAct(state.act, act) || viewOf(state) !== view) continue;
|
||||
next.add(to);
|
||||
seenEdges.add(`${from}>${to}`);
|
||||
seenStates.add(to);
|
||||
}
|
||||
}
|
||||
if (next.size === 0) {
|
||||
const expected = [...frontier].flatMap((from) => [...(graph.next.get(from) ?? [])]).filter((to) => sameAct(graph.states.get(to).act, act)).map((to) => viewOf(graph.states.get(to)));
|
||||
throw new Error(`walk ${number} (seed ${firstSeed}) left the model after\n ${path.join("\n ")}\n code: ${view}\n model: ${[...new Set(expected)].join("\n ")}`);
|
||||
}
|
||||
frontier = next;
|
||||
steps += 1;
|
||||
};
|
||||
try {
|
||||
for (let taken = 0; taken < depth; taken += 1) {
|
||||
const act = chooseAct(frontier);
|
||||
if (!act) break;
|
||||
await step(act);
|
||||
}
|
||||
// The queue's fairness: every message an agent can take reaches it.
|
||||
const deliverable = (act, state) => act.name === "Deliver" && state.agents.some((t) => nameOf[t] === act.n && keyOf[t] === act.key);
|
||||
for (let left = 16; left > 0; left -= 1) {
|
||||
const act = chooseAct(frontier, deliverable);
|
||||
if (!act) break;
|
||||
await step(act);
|
||||
}
|
||||
// Every prompt an app sent ran, on its own agent, as SentRuns says.
|
||||
const ranIds = new Set(turns.map((turn) => idOf(turn.turnId)));
|
||||
for (const id of walk.sent) assert.ok(ranIds.has(id), `walk ${number}: prompt ${id} ran`);
|
||||
} finally {
|
||||
await walk.close();
|
||||
}
|
||||
}
|
||||
|
||||
for (let number = 1; number <= walks; number += 1) await walkOnce(number);
|
||||
const edges = [...graph.next.values()].reduce((total, successors) => total + successors.size, 0);
|
||||
console.log(`${walks} walks (seed ${firstSeed}), ${steps} steps (${Object.entries(outcomes).sort().map(([name, count]) => `${count} ${name}`).join(", ")}): ${seenStates.size} of ${graph.states.size} states and ${seenEdges.size} of ${edges} edges, every step one the model allows`);
|
||||
for (const outcome of ["register", "register refused", "prompt", "delivered", "forge refused", "forge missed", "oversize refused"]) {
|
||||
assert.ok((outcomes[outcome] ?? 0) > 0, `the walks reach "${outcome}" at least once`);
|
||||
}
|
||||
@@ -234,6 +234,8 @@ async function harness() {
|
||||
writes[w] = held;
|
||||
agents[w] = createFxAgent({
|
||||
...(testHooks ? { hooks: testHooks } : {}),
|
||||
// Each worker stands for another server's copy of one agent.
|
||||
[Symbol.for("libfx.replica")]: true,
|
||||
backend,
|
||||
nativeAddon: addon,
|
||||
...(wasm ? { wasm } : {}),
|
||||
|
||||
@@ -1,15 +0,0 @@
|
||||
CONSTANTS
|
||||
Names = {"S", "C"}
|
||||
Sessions = {"s1", "s2"}
|
||||
Procs = {"p1", "p2"}
|
||||
P2Lacks = "C"
|
||||
MaxSends = 3
|
||||
RecordAtSend = FALSE
|
||||
ByName = TRUE
|
||||
CheckOwner = TRUE
|
||||
SPECIFICATION Spec
|
||||
INVARIANTS
|
||||
TypeOK
|
||||
RunsOnOwner
|
||||
MissOnlyWhereLacking
|
||||
PROPERTY OwnerKept
|
||||
@@ -1,11 +0,0 @@
|
||||
CONSTANTS
|
||||
Names = {"S", "C"}
|
||||
Sessions = {"s1", "s2"}
|
||||
Procs = {"p1", "p2"}
|
||||
P2Lacks = "C"
|
||||
MaxSends = 3
|
||||
RecordAtSend = FALSE
|
||||
ByName = TRUE
|
||||
CheckOwner = TRUE
|
||||
SPECIFICATION FairSpec
|
||||
PROPERTY Drains
|
||||
@@ -4,7 +4,6 @@ CONSTANTS
|
||||
Procs = {"p1", "p2"}
|
||||
P2Lacks = "C"
|
||||
MaxSends = 3
|
||||
RecordAtSend = TRUE
|
||||
ByName = TRUE
|
||||
CheckOwner = TRUE
|
||||
SPECIFICATION FairSpec
|
||||
|
||||
@@ -4,7 +4,6 @@ CONSTANTS
|
||||
Procs = {"p1", "p2"}
|
||||
P2Lacks = "C"
|
||||
MaxSends = 3
|
||||
RecordAtSend = TRUE
|
||||
ByName = TRUE
|
||||
CheckOwner = TRUE
|
||||
SPECIFICATION Spec
|
||||
|
||||
@@ -4,7 +4,6 @@ CONSTANTS
|
||||
Procs = {"p1", "p2"}
|
||||
P2Lacks = "C"
|
||||
MaxSends = 6
|
||||
RecordAtSend = FALSE
|
||||
ByName = TRUE
|
||||
CheckOwner = TRUE
|
||||
SPECIFICATION Spec
|
||||
|
||||
@@ -11,9 +11,9 @@
|
||||
(* session belongs to the agent its first stored input names, so a prompt *)
|
||||
(* one agent sends to another agent's session is sent on to that agent. *)
|
||||
(* *)
|
||||
(* An input is stored by the sender before it queues the message on a *)
|
||||
(* queue that ends with the process (memory, local: RecordAtSend), and by *)
|
||||
(* the worker that takes it on one that keeps messages (Vercel). *)
|
||||
(* The sender stores an input before it queues the message, on every *)
|
||||
(* queue, so a session has its owner from its first send on. A message *)
|
||||
(* only names the input (TenantIsolation.tla LogFirst). *)
|
||||
(* *)
|
||||
(* With ByName = FALSE a process hands each delivery to any of its agents, *)
|
||||
(* as libfx did before names (Witness-NoNames). With CheckOwner = FALSE a *)
|
||||
@@ -31,7 +31,6 @@ CONSTANTS
|
||||
Procs, \* processes the queue delivers to; each has an agent of every name but P2Lacks
|
||||
P2Lacks, \* a name process "p2" has no agent for, or "none"
|
||||
MaxSends, \* prompts sent in all
|
||||
RecordAtSend, \* the sender stores a prompt's input before it queues it
|
||||
ByName, \* a delivery goes to the agent the message names
|
||||
CheckOwner \* a worker runs only its own agent's sessions
|
||||
|
||||
@@ -73,12 +72,12 @@ Send(n, s) ==
|
||||
/\ sends < MaxSends
|
||||
/\ queue' = queue \cup {[id |-> sends, s |-> s, n |-> n]}
|
||||
/\ sends' = sends + 1
|
||||
/\ owner' = IF RecordAtSend THEN [owner EXCEPT ![s] = Stored(@, n)] ELSE owner
|
||||
/\ owner' = [owner EXCEPT ![s] = Stored(@, n)]
|
||||
/\ act' = [name |-> "Send", n |-> n, s |-> s, id |-> sends]
|
||||
/\ UNCHANGED <<ran, missed>>
|
||||
|
||||
\* The queue delivers m to process p. The agent a that takes it stores the
|
||||
\* input, then runs the turn or sends the message on to the session's owner.
|
||||
\* The queue delivers m to process p. The agent a that takes it runs the
|
||||
\* turn, or sends the message on to the session's owner.
|
||||
Deliver(p, m) ==
|
||||
/\ m \in queue
|
||||
/\ act' = [name |-> "Deliver", p |-> p, id |-> m.id]
|
||||
|
||||
@@ -0,0 +1,10 @@
|
||||
CONSTANTS
|
||||
Topology = "four"
|
||||
MaxPrompts = 2
|
||||
MaxForges = 1
|
||||
LogFirst = TRUE
|
||||
RouteByStore = TRUE
|
||||
OneDefinition = TRUE
|
||||
Admission = TRUE
|
||||
SPECIFICATION FairSpec
|
||||
PROPERTY SentRuns
|
||||
@@ -0,0 +1,16 @@
|
||||
CONSTANTS
|
||||
Topology = "four"
|
||||
MaxPrompts = 2
|
||||
MaxForges = 1
|
||||
LogFirst = TRUE
|
||||
RouteByStore = TRUE
|
||||
OneDefinition = TRUE
|
||||
Admission = TRUE
|
||||
SPECIFICATION Spec
|
||||
INVARIANTS
|
||||
TypeOK
|
||||
OnlySentRuns
|
||||
RunsOnSenderAgent
|
||||
RunsInSenderStore
|
||||
RunsWithSenderContext
|
||||
StoredWithinLimit
|
||||
@@ -0,0 +1,242 @@
|
||||
--------------------------- MODULE TenantIsolation ---------------------------
|
||||
(***************************************************************************)
|
||||
(* Tenants whose durable agents share one process, such as one Vercel *)
|
||||
(* Fluid function: what a turn may run, on whose agent, in which store, *)
|
||||
(* and with whose context. *)
|
||||
(* *)
|
||||
(* Each tenant's app creates one agent, with a name, on a durability. The *)
|
||||
(* durability has a store key, which libfx reads off it (the `store` of *)
|
||||
(* world(), or its name), and a store, where its sessions are written. *)
|
||||
(* Two tenants can collide on a name and a key while their stores differ, *)
|
||||
(* as two world() Worlds both named "world" do. The app prompts only its *)
|
||||
(* own tenant's session. Every queued message reaches one route, *)
|
||||
(* wakeHandler(), which hands it to an agent in the process. An attacker *)
|
||||
(* can post any delivery to that route: a message id no app sent, or one *)
|
||||
(* it did, with any session, agent name, store key and context. *)
|
||||
(* *)
|
||||
(* Four rules keep tenants apart. Each constant turns one on: *)
|
||||
(* *)
|
||||
(* - LogFirst: a prompt's input, with its caller's context, reaches the *)
|
||||
(* session's log before its message is queued, and the message only *)
|
||||
(* names it. A delivery runs only what the log holds, so one the log *)
|
||||
(* does not hold is refused (Witness-TrustBody). *)
|
||||
(* - RouteByStore: a delivery goes only to an agent on the store key its *)
|
||||
(* message names, never to one on another store that has the same name *)
|
||||
(* (Witness-RouteByName). *)
|
||||
(* - OneDefinition: libfx refuses to create a second agent with a name a *)
|
||||
(* different agent already has on the same store key, so one name on *)
|
||||
(* one store is one agent (Witness-ShareName). *)
|
||||
(* - Admission: an input over the size limit is refused before anything *)
|
||||
(* is stored (Witness-NoAdmission). *)
|
||||
(* *)
|
||||
(* With all four off, the spec is libfx before them (Witness-Today). *)
|
||||
(* *)
|
||||
(* A worker that takes a message runs every input of the session its *)
|
||||
(* store holds and no turn ran yet, as one turn each. A session belongs to *)
|
||||
(* the agent its first stored input names; a message that reached another *)
|
||||
(* agent is sent on to that one, as AgentRouting.tla checks. *)
|
||||
(***************************************************************************)
|
||||
EXTENDS Naturals, FiniteSets
|
||||
|
||||
CONSTANTS
|
||||
Topology, \* which tenants share the process: "four", "shared" or "split"
|
||||
MaxPrompts, \* prompts the apps send in all
|
||||
MaxForges, \* deliveries the attacker posts in all
|
||||
LogFirst,
|
||||
RouteByStore,
|
||||
OneDefinition,
|
||||
Admission
|
||||
|
||||
None == "none"
|
||||
Forger == "forger"
|
||||
|
||||
\* "four": acme and evil collide on a name and a store key with different
|
||||
\* stores; bob shares acme's store under another name; carl has acme's name
|
||||
\* on another store key. "shared": two tenants on one store, one name, as
|
||||
\* two agents created without a name or a durability are. "split": one name
|
||||
\* on two store keys.
|
||||
Tenants ==
|
||||
CASE Topology = "four" -> {"acme", "evil", "bob", "carl"}
|
||||
[] Topology = "shared" -> {"acme", "evil"}
|
||||
[] Topology = "split" -> {"acme", "carl"}
|
||||
NameOf == [t \in Tenants |-> IF t = "bob" THEN "billing" ELSE "support"]
|
||||
KeyOf == [t \in Tenants |-> IF t = "carl" THEN "k2" ELSE "k1"]
|
||||
StoreOf == [t \in Tenants |->
|
||||
CASE t = "acme" -> "A"
|
||||
[] t = "evil" -> IF Topology = "shared" THEN "A" ELSE "B"
|
||||
[] t = "bob" -> "A"
|
||||
[] t = "carl" -> "C"]
|
||||
|
||||
Names == {NameOf[t] : t \in Tenants}
|
||||
Keys == {KeyOf[t] : t \in Tenants}
|
||||
Stores == {StoreOf[t] : t \in Tenants}
|
||||
\* Id 0 is one no app ever sends.
|
||||
Ids == 0..MaxPrompts
|
||||
Contexts == Tenants \cup {Forger, None}
|
||||
Sizes == {"ok", "big", None}
|
||||
|
||||
Msg == [id: Ids, s: Tenants, n: Names, key: Keys, ctx: Contexts, size: Sizes]
|
||||
Entry == [id: Ids, s: Tenants, n: Names, ctx: Contexts, size: Sizes]
|
||||
Turn == [id: Ids, def: Tenants, store: Stores, ctx: Contexts]
|
||||
|
||||
VARIABLES
|
||||
agents, \* tenants whose agent runs in the process
|
||||
refused, \* tenants whose agent libfx refused to create
|
||||
log, \* the inputs each store holds
|
||||
owner, \* the agent each session belongs to, in each store
|
||||
done, \* the inputs whose turn ran, in each store
|
||||
queue, \* messages queued and not yet taken
|
||||
ran, \* turns run: which input, on whose agent, in which store, with which context
|
||||
sent, \* the prompts the apps sent, by id and tenant
|
||||
forges, \* deliveries the attacker posted so far
|
||||
act \* the step just taken, for the driver
|
||||
|
||||
vars == <<agents, refused, log, owner, done, queue, ran, sent, forges, act>>
|
||||
|
||||
TypeOK ==
|
||||
/\ agents \subseteq Tenants
|
||||
/\ refused \subseteq Tenants
|
||||
/\ log \in [Stores -> SUBSET Entry]
|
||||
/\ owner \in [Stores -> [Tenants -> Names \cup {None}]]
|
||||
/\ done \in [Stores -> SUBSET Ids]
|
||||
/\ queue \subseteq Msg
|
||||
/\ ran \subseteq Turn
|
||||
/\ sent \subseteq [id: Ids, t: Tenants]
|
||||
/\ forges \in 0..MaxForges
|
||||
|
||||
Init ==
|
||||
/\ agents = {}
|
||||
/\ refused = {}
|
||||
/\ log = [st \in Stores |-> {}]
|
||||
/\ owner = [st \in Stores |-> [s \in Tenants |-> None]]
|
||||
/\ done = [st \in Stores |-> {}]
|
||||
/\ queue = {}
|
||||
/\ ran = {}
|
||||
/\ sent = {}
|
||||
/\ forges = 0
|
||||
/\ act = [name |-> "Init"]
|
||||
|
||||
\* What the queue carries for m: with LogFirst, only which input it names.
|
||||
Ref(m) == IF LogFirst THEN [m EXCEPT !.ctx = None, !.size = None] ELSE m
|
||||
|
||||
\* Tenant t's app creates its agent.
|
||||
Register(t) ==
|
||||
/\ t \notin agents \cup refused
|
||||
/\ IF OneDefinition /\ \E u \in agents: NameOf[u] = NameOf[t] /\ KeyOf[u] = KeyOf[t]
|
||||
THEN /\ refused' = refused \cup {t}
|
||||
/\ UNCHANGED agents
|
||||
ELSE /\ agents' = agents \cup {t}
|
||||
/\ UNCHANGED refused
|
||||
/\ act' = [name |-> "Register", t |-> t]
|
||||
/\ UNCHANGED <<log, owner, done, queue, ran, sent, forges>>
|
||||
|
||||
\* Tenant t's app sends an input of `size` to its own session.
|
||||
Send(t, size) ==
|
||||
LET id == Cardinality(sent) + 1
|
||||
st == StoreOf[t]
|
||||
m == [id |-> id, s |-> t, n |-> NameOf[t], key |-> KeyOf[t], ctx |-> t, size |-> size]
|
||||
IN
|
||||
/\ t \in agents
|
||||
/\ Cardinality(sent) < MaxPrompts
|
||||
/\ sent' = sent \cup {[id |-> id, t |-> t]}
|
||||
/\ queue' = queue \cup {Ref(m)}
|
||||
/\ IF LogFirst
|
||||
THEN /\ log' = [log EXCEPT ![st] = @ \cup {[id |-> id, s |-> t, n |-> NameOf[t], ctx |-> t, size |-> size]}]
|
||||
/\ owner' = [owner EXCEPT ![st][t] = IF @ = None THEN NameOf[t] ELSE @]
|
||||
ELSE UNCHANGED <<log, owner>>
|
||||
/\ UNCHANGED <<agents, refused, done, ran, forges>>
|
||||
|
||||
Prompt(t) ==
|
||||
/\ Send(t, "ok")
|
||||
/\ act' = [name |-> "Prompt", t |-> t, id |-> Cardinality(sent) + 1]
|
||||
|
||||
\* The app sends an input over the limit, as a prompt or a steer.
|
||||
Oversize(t, k) ==
|
||||
/\ t \in agents
|
||||
/\ IF Admission
|
||||
THEN UNCHANGED <<agents, refused, log, owner, done, queue, ran, sent, forges>>
|
||||
ELSE Send(t, "big")
|
||||
/\ act' = [name |-> "Oversize", t |-> t, k |-> k]
|
||||
|
||||
\* The agents in the process a delivery of m can reach.
|
||||
Candidates(m) == {a \in agents: NameOf[a] = m.n /\ (RouteByStore => KeyOf[a] = m.key)}
|
||||
|
||||
\* Agent a's worker takes m; `rest` is the queue without it.
|
||||
Take(m, a, rest) ==
|
||||
LET st == StoreOf[a]
|
||||
held == {x \in log[st]: x.id = m.id /\ x.s = m.s}
|
||||
own == IF owner[st][m.s] = None THEN m.n ELSE owner[st][m.s]
|
||||
body == [id |-> m.id, s |-> m.s, n |-> m.n, ctx |-> m.ctx, size |-> m.size]
|
||||
logged == IF held = {} THEN log[st] \cup {body} ELSE log[st]
|
||||
pend == {x \in logged: x.s = m.s /\ x.id \notin done[st]}
|
||||
IN
|
||||
IF LogFirst /\ held = {}
|
||||
THEN \* The log does not hold it: refused, and nothing is written.
|
||||
/\ queue' = rest
|
||||
/\ UNCHANGED <<log, owner, done, ran>>
|
||||
ELSE IF own # NameOf[a]
|
||||
THEN \* Another agent's session: sent on to it.
|
||||
/\ queue' = rest \cup {Ref([m EXCEPT !.n = own])}
|
||||
/\ UNCHANGED <<log, owner, done, ran>>
|
||||
ELSE /\ log' = [log EXCEPT ![st] = logged]
|
||||
/\ owner' = [owner EXCEPT ![st][m.s] = own]
|
||||
/\ ran' = ran \cup {[id |-> x.id, def |-> a, store |-> st, ctx |-> x.ctx] : x \in pend}
|
||||
/\ done' = [done EXCEPT ![st] = @ \cup {x.id : x \in pend}]
|
||||
/\ queue' = rest
|
||||
|
||||
\* A delivery of m reaches the route. With no agent to take it, the queue
|
||||
\* keeps `missed` as it was.
|
||||
Handle(m, rest, missed) ==
|
||||
IF Candidates(m) = {}
|
||||
THEN /\ queue' = missed
|
||||
/\ UNCHANGED <<log, owner, done, ran>>
|
||||
ELSE \E a \in Candidates(m): Take(m, a, rest)
|
||||
|
||||
\* The queue delivers a message to the route.
|
||||
Deliver(m) ==
|
||||
/\ m \in queue
|
||||
/\ Handle(m, queue \ {m}, queue)
|
||||
/\ act' = [name |-> "Deliver", id |-> m.id, s |-> m.s, n |-> m.n, key |-> m.key]
|
||||
/\ UNCHANGED <<agents, refused, sent, forges>>
|
||||
|
||||
\* The attacker posts a delivery straight to the route.
|
||||
Forge(id, s, n, key) ==
|
||||
/\ forges < MaxForges
|
||||
/\ id \in {0} \cup {x.id : x \in sent}
|
||||
/\ Handle([id |-> id, s |-> s, n |-> n, key |-> key, ctx |-> Forger, size |-> "ok"], queue, queue)
|
||||
/\ forges' = forges + 1
|
||||
/\ act' = [name |-> "Forge", id |-> id, s |-> s, n |-> n, key |-> key]
|
||||
/\ UNCHANGED <<agents, refused, sent>>
|
||||
|
||||
Next ==
|
||||
\/ \E t \in Tenants: Register(t)
|
||||
\/ \E t \in Tenants: Prompt(t)
|
||||
\/ \E t \in Tenants, k \in {"prompt", "steer"}: Oversize(t, k)
|
||||
\/ \E m \in queue: Deliver(m)
|
||||
\/ \E id \in Ids, s \in Tenants, n \in Names, key \in Keys: Forge(id, s, n, key)
|
||||
|
||||
Spec == Init /\ [][Next]_vars
|
||||
|
||||
\* The queue keeps delivering each message some agent can take.
|
||||
FairSpec == Spec /\ WF_vars(\E m \in queue: Candidates(m) # {} /\ Deliver(m))
|
||||
|
||||
-----------------------------------------------------------------------------
|
||||
(* Properties *)
|
||||
|
||||
Sender(r) == {x \in sent: x.id = r.id}
|
||||
|
||||
\* Only an input an app sent runs.
|
||||
OnlySentRuns == \A r \in ran: Sender(r) # {}
|
||||
|
||||
\* It runs on its own tenant's agent, in its tenant's store, with the
|
||||
\* context its app gave.
|
||||
RunsOnSenderAgent == \A r \in ran: \A x \in Sender(r): r.def = x.t
|
||||
RunsInSenderStore == \A r \in ran: \A x \in Sender(r): r.store = StoreOf[x.t]
|
||||
RunsWithSenderContext == \A r \in ran: \A x \in Sender(r): r.ctx = x.t
|
||||
|
||||
\* No store holds an input over the limit.
|
||||
StoredWithinLimit == \A st \in Stores: \A e \in log[st]: e.size # "big"
|
||||
|
||||
\* Every prompt an app sent runs.
|
||||
SentRuns == \A i \in 1..MaxPrompts: []((\E x \in sent: x.id = i) => <>(\E r \in ran: r.id = i))
|
||||
=============================================================================
|
||||
@@ -0,0 +1,10 @@
|
||||
CONSTANTS
|
||||
Topology = "split"
|
||||
MaxPrompts = 2
|
||||
MaxForges = 1
|
||||
LogFirst = TRUE
|
||||
RouteByStore = TRUE
|
||||
OneDefinition = TRUE
|
||||
Admission = FALSE
|
||||
SPECIFICATION Spec
|
||||
INVARIANT StoredWithinLimit
|
||||
@@ -0,0 +1,10 @@
|
||||
CONSTANTS
|
||||
Topology = "split"
|
||||
MaxPrompts = 2
|
||||
MaxForges = 1
|
||||
LogFirst = TRUE
|
||||
RouteByStore = FALSE
|
||||
OneDefinition = TRUE
|
||||
Admission = TRUE
|
||||
SPECIFICATION FairSpec
|
||||
PROPERTY SentRuns
|
||||
@@ -0,0 +1,10 @@
|
||||
CONSTANTS
|
||||
Topology = "shared"
|
||||
MaxPrompts = 2
|
||||
MaxForges = 1
|
||||
LogFirst = TRUE
|
||||
RouteByStore = TRUE
|
||||
OneDefinition = FALSE
|
||||
Admission = TRUE
|
||||
SPECIFICATION Spec
|
||||
INVARIANT RunsOnSenderAgent
|
||||
@@ -0,0 +1,10 @@
|
||||
CONSTANTS
|
||||
Topology = "split"
|
||||
MaxPrompts = 2
|
||||
MaxForges = 1
|
||||
LogFirst = FALSE
|
||||
RouteByStore = FALSE
|
||||
OneDefinition = FALSE
|
||||
Admission = FALSE
|
||||
SPECIFICATION Spec
|
||||
INVARIANT RunsInSenderStore
|
||||
@@ -0,0 +1,10 @@
|
||||
CONSTANTS
|
||||
Topology = "split"
|
||||
MaxPrompts = 2
|
||||
MaxForges = 1
|
||||
LogFirst = FALSE
|
||||
RouteByStore = TRUE
|
||||
OneDefinition = TRUE
|
||||
Admission = TRUE
|
||||
SPECIFICATION Spec
|
||||
INVARIANT OnlySentRuns
|
||||
@@ -4,7 +4,6 @@ CONSTANTS
|
||||
Procs = {"p1", "p2"}
|
||||
P2Lacks = "C"
|
||||
MaxSends = 3
|
||||
RecordAtSend = TRUE
|
||||
ByName = FALSE
|
||||
CheckOwner = FALSE
|
||||
SPECIFICATION Spec
|
||||
|
||||
@@ -4,7 +4,6 @@ CONSTANTS
|
||||
Procs = {"p1", "p2"}
|
||||
P2Lacks = "C"
|
||||
MaxSends = 3
|
||||
RecordAtSend = FALSE
|
||||
ByName = TRUE
|
||||
CheckOwner = FALSE
|
||||
SPECIFICATION Spec
|
||||
|
||||
+40
-6
@@ -14,8 +14,9 @@
|
||||
# bun ../drivers/drive.mjs write && bun ../drivers/drive.mjs cutoff
|
||||
# bun ../drivers/drive.mjs leased # the Heartbeat graph, on a World without liveness
|
||||
# bun ../drivers/drive.mjs queue # the QueueHeld graph, on a World whose queue shows it
|
||||
# bun ../drivers/drive-agents.mjs send && bun ../drivers/drive-agents.mjs consume # AgentRouting
|
||||
# bun ../drivers/drive-agents.mjs # AgentRouting
|
||||
# bun ../drivers/drive-steer.mjs ideal && bun ../drivers/drive-steer.mjs faults # DurableSteer
|
||||
# bun ../drivers/drive-tenants.mjs # TenantIsolation
|
||||
set -eu
|
||||
cd "$(dirname "$0")"
|
||||
|
||||
@@ -98,13 +99,12 @@ for pair in NoInterrupt:NoSteerWaitsOnReply NoHandBack:NothingLost \
|
||||
done
|
||||
|
||||
# AgentRouting.tla: which agent runs a session's turns when an app has
|
||||
# several. Its graphs are what ../drivers/drive-agents.mjs walks: inputs stored when
|
||||
# sent (memory, local) and when delivered (Vercel), with process p2 lacking
|
||||
# one agent. Drains is checked under fairness, a simulation runs a larger
|
||||
# several. Its graph is what ../drivers/drive-agents.mjs walks, with process p2
|
||||
# lacking one agent; every sender stores its input before it queues it. Drains is checked under fairness, a simulation runs a larger
|
||||
# config, and each witness breaks RunsOnOwner: NoNames is routing without
|
||||
# names, as before them, and NoOwnerCheck a worker that runs any session it
|
||||
# is handed.
|
||||
for config in send consume; do
|
||||
for config in send; do
|
||||
echo "== AgentRouting ($config)"
|
||||
if ! "$java" -XX:+UseParallelGC -cp "$jar" tlc2.TLC -deadlock -workers 1 -cleanup \
|
||||
-metadir "$work/agents-$config" -dump dot,actionlabels "graphs/agents-$config" \
|
||||
@@ -113,7 +113,7 @@ for config in send consume; do
|
||||
fi
|
||||
grep -E 'distinct states|depth of' "$work/agents-$config.log"
|
||||
done
|
||||
for config in drains-send drains-consume; do
|
||||
for config in drains-send; do
|
||||
if "$java" -cp "$jar" tlc2.TLC -deadlock -workers auto -cleanup -metadir "$work/agents-$config" \
|
||||
-config "AgentRouting-$config.cfg" AgentRouting.tla > "$work/agents-$config.log" 2>&1; then
|
||||
echo "== AgentRouting ($config): Drains holds"
|
||||
@@ -135,4 +135,38 @@ for witness in NoNames NoOwnerCheck; do
|
||||
echo "== witness $witness NOT reached" >&2; status=1
|
||||
fi
|
||||
done
|
||||
|
||||
# TenantIsolation.tla: tenants whose agents share one process, and an
|
||||
# attacker who can post to its route. Its graph, with every rule on, is what
|
||||
# ../drivers/drive-tenants.mjs walks; SentRuns is checked under fairness.
|
||||
# Each witness turns one rule off and must break what it protects:
|
||||
# TrustBody runs a forged delivery, ShareName runs one tenant's prompt on
|
||||
# another's agent, RouteByName strands a prompt on another store,
|
||||
# NoAdmission stores an oversized input, and Today, every rule off as libfx
|
||||
# was before them, runs a prompt in another tenant's store.
|
||||
echo "== TenantIsolation (route)"
|
||||
if ! "$java" -XX:+UseParallelGC -cp "$jar" tlc2.TLC -deadlock -workers 1 -cleanup \
|
||||
-metadir "$work/tenants-route" -dump dot,actionlabels graphs/tenants-route \
|
||||
-config TenantIsolation-route.cfg TenantIsolation.tla > "$work/tenants-route.log" 2>&1; then
|
||||
cat "$work/tenants-route.log"; status=1
|
||||
else
|
||||
grep -E 'distinct states|depth of' "$work/tenants-route.log"
|
||||
fi
|
||||
if "$java" -cp "$jar" tlc2.TLC -deadlock -workers auto -cleanup -metadir "$work/tenants-drains" \
|
||||
-config TenantIsolation-drains.cfg TenantIsolation.tla > "$work/tenants-drains.log" 2>&1; then
|
||||
echo "== TenantIsolation (drains): SentRuns holds"
|
||||
else
|
||||
cat "$work/tenants-drains.log"; status=1
|
||||
fi
|
||||
for pair in "TrustBody:Invariant OnlySentRuns is violated" "ShareName:Invariant RunsOnSenderAgent is violated" \
|
||||
"RouteByName:Temporal properties were violated" "NoAdmission:Invariant StoredWithinLimit is violated" \
|
||||
"Today:Invariant RunsInSenderStore is violated"; do
|
||||
witness="${pair%%:*}" verdict="${pair#*:}"
|
||||
if "$java" -cp "$jar" tlc2.TLC -deadlock -workers auto -cleanup -metadir "$work/tw-$witness" \
|
||||
-config "TenantWitness-$witness.cfg" TenantIsolation.tla 2>&1 | grep -q -F "$verdict"; then
|
||||
echo "== tenant witness $witness reached"
|
||||
else
|
||||
echo "== tenant witness $witness NOT reached" >&2; status=1
|
||||
fi
|
||||
done
|
||||
exit "$status"
|
||||
Reference in new issue
Block a user