From 37707bdd0ca0b103095596e8250d86ca6f837035 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Thu, 22 Oct 2026 19:51:12 +0300 Subject: [PATCH 1/4] feat: createPolicyDecider and getDecisionRequests --- .changeset/policy-decider-helpers.md | 7 + docs/decisions.md | 21 +++ docs/environment-and-policy.md | 2 +- docs/evals.md | 21 +++ skills/add-evals/SKILL.md | 10 +- skills/improve-machine/SKILL.md | 4 +- skills/loop-to-machine/policy.md | 23 +-- src/decision-requests.test.ts | 90 +++++++++++ src/decision-requests.ts | 53 +++++++ src/fold-traces.ts | 23 +-- src/index.ts | 2 + src/internal/trace-states.ts | 28 ++++ src/policy-decider.test.ts | 223 +++++++++++++++++++++++++++ src/policy-decider.ts | 79 ++++++++++ src/testing/index.ts | 2 + 15 files changed, 544 insertions(+), 44 deletions(-) create mode 100644 .changeset/policy-decider-helpers.md create mode 100644 src/decision-requests.test.ts create mode 100644 src/decision-requests.ts create mode 100644 src/internal/trace-states.ts create mode 100644 src/policy-decider.test.ts create mode 100644 src/policy-decider.ts diff --git a/.changeset/policy-decider-helpers.md b/.changeset/policy-decider-helpers.md new file mode 100644 index 00000000..82ec480b --- /dev/null +++ b/.changeset/policy-decider-helpers.md @@ -0,0 +1,7 @@ +--- +"@statelyai/agent": minor +--- + +**`createPolicyDecider` runs a policy machine as a decide executor.** Each decision request starts one policy run, whose output event is the choice. The input defaults to the request's `metadata.facts`, offered events, `system`, and `prompt`; the policy's token usage is reported on the result. A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. + +**`getDecisionRequests` reads the decisions a run asked for from its trace.** From `@statelyai/agent/testing`: `getDecisionRequests(trace, { name? })` returns each decision's first-attempt request with the states it was asked in, from a live or serialized trace, for eval rows and policy probes. diff --git a/docs/decisions.md b/docs/decisions.md index 5dd12c7c..e4dc198b 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -256,6 +256,27 @@ const { decide } = createAiSdkExecutors({ See [Decision models](models-and-providers.md#decision-models). +## A policy as the decider + +A decision can be answered by another machine: a policy that applies a rule when the facts are enough and asks a model only for the rest. `createPolicyDecider` turns a policy machine into a decide executor. + +```ts no-check +import { createPolicyDecider, runAgent } from "@statelyai/agent"; + +const decide = createPolicyDecider(policyMachine, { executors: modelExecutors }); + +await runAgent(deskMachine, { input, executors: { ...modelExecutors, decide } }); +``` + +- Each decision request starts one run of the policy. Its output, an event, is the choice. The environment still judges it: an event it does not accept is retried like any decider's. +- The policy's input defaults to `{ facts, options, system, prompt }`: the request's `metadata.facts`, the offered events, and its prompt. The policy's input schema validates it. Pass `input: (request) => ...` to build it another way. +- The policy's own `agent.decide` invokes are answered by `executors`. Pass the environment's `metadata` on to them, so the model sees what it would without the policy. +- The result carries the policy's token usage, and none when it decided with a rule. +- A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. Give each of the policy's decisions an `onError` that falls back to an event the environment always accepts. +- Every other `runAgent` option passes through: `store`, `threadId`, and `metadata` to journal the policy's runs, `onTrace` to trace them. + +To route by request name, wrap it: `decide: (request, info) => (request.name === "chooseResolution" ? policy(request, info) : modelExecutors.decide!(request, info))`. + ## Decisions at the boundary of a plain machine A machine does not have to know about AI. `decideNext` asks a model to pick one event that a plain XState machine accepts right now, built with `createMachine` or `setup`, with no `agent.*` invoke anywhere. diff --git a/docs/environment-and-policy.md b/docs/environment-and-policy.md index 2b85dee7..ee2d2669 100644 --- a/docs/environment-and-policy.md +++ b/docs/environment-and-policy.md @@ -30,7 +30,7 @@ environment (TypeScript, fixed) policy (TS or JSON, improvable) | Policy | A model, reviewed by a person | TypeScript `setupAgent` machine, or JSON `AgentWorkflowConfig` | Between runs, on measured gains | TS: `tsc`, `lintAgentMachine`, replay, objective; JSON: `verifyAgentConfig`, replay, objective | - The environment exposes decision points: states where it asks for one event. Its guards decide whether to accept whatever the policy proposes, so a policy cannot take an action the environment forbids. -- The policy answers at those decision points. It can be a single model call, a ladder of `choice` states over the facts, a judgment, or a request to ask the customer. Its final state outputs the event it proposes. The host runs the policy as the environment's `decide` executor, as the [worked example](../examples/self-improving-refunds) does. +- The policy answers at those decision points. It can be a single model call, a ladder of `choice` states over the facts, a judgment, or a request to ask the customer. Its final state outputs the event it proposes. The host runs the policy as the environment's `decide` executor: `createPolicyDecider(policyMachine, { executors })` starts one policy run per decision and returns its output event. See [A policy as the decider](decisions.md#a-policy-as-the-decider). - An invariant sentinel, a state that is entered only when a rule is broken, stays unreachable for every policy. `canReach` proves it once, against the environment, for every event sequence and payload variant you list. See [Payloads and invoke outputs](verify.md#payloads-and-invoke-outputs). The environment can be a plain XState machine. [`decideNext`](decisions.md#decisions-at-the-boundary-of-a-plain-machine) asks a model to pick one event that a plain XState machine accepts right now, and [closed choices](decisions.md#closed-choices) list the concrete events a decider may pick. diff --git a/docs/evals.md b/docs/evals.md index 5e135c0a..17c4ae53 100644 --- a/docs/evals.md +++ b/docs/evals.md @@ -220,6 +220,27 @@ const stats = foldTraces(traces, { - `decideNext` called by the host emits no trace. Push its request onto the trace before sending the event: `trace.push({ type: "request.start", request: result.request })`. - Rows come in the order states were first seen. Sort them yourself, for example by `failures / visits`. +## Decision requests from traces + +`getDecisionRequests` (from `@statelyai/agent/testing`) lists the decisions a run asked for, read from its trace: each decision's first `request.start`, and the states the run was in. + +```ts no-check +import { getDecisionRequests } from "@statelyai/agent/testing"; + +for (const { states, request } of getDecisionRequests(trace, { name: "chooseResolution" })) { + const { event } = await decide(request); // a candidate decider + console.log( + states, + request.options?.map((o) => o.event), + event, + ); +} +``` + +- A trace is one run's events, live or from `serializeTraceEvent`. Retries of a decision are left out, so each decision appears once. +- A serialized request keeps its `options`, `withheld`, `metadata`, `system`, and `prompt`. Event schemas do not survive serialization, so replay closed choices rather than open ones. +- Use the requests as eval rows (a decider on each, compared with a label) or as probe inputs for a policy. + ## Parity with the code you replace `compareRuns` (from `@statelyai/agent/testing`) runs every scenario through the existing implementation and its replacement, and labels each difference. Use it when you port a loop to a machine and need to show where they behave the same and where the machine differs on purpose. diff --git a/skills/add-evals/SKILL.md b/skills/add-evals/SKILL.md index ac69509b..5dfefc23 100644 --- a/skills/add-evals/SKILL.md +++ b/skills/add-evals/SKILL.md @@ -84,11 +84,15 @@ Use `explorePaths` to assert the set of reachable final states, and `await simul - `explorePaths` and `canReach` use `options.payloads` **instead of** the decision's own offered options, so they explore values the machine would never offer. That is what a boundary proof needs. - `simulateAgent` keeps the offered options: a scripted event the decision did not offer is refused, and the run ends in the decision's `onError`. Script only offered events, or pick an input that offers the value. -**Decision evals.** A row is a decision request and the event a person says is right. Write rows by hand, or take them from recorded traces: the first attempt of each decision is a `request.start` event with `request.kind === "decision"` and `request.attempts.length === 0`, and its `request` carries `name`, `options`, and `metadata`. Cast the serialized `request` to `AgentExecutorDecisionRequest`, and compare events without caring about key order. If the decider reads anything outside the request (the ticket, the tool history), store that in the row too. +**Decision evals.** A row is a decision request and the event a person says is right. Write rows by hand, or take them from recorded traces with `getDecisionRequests(trace, { name })` from `@statelyai/agent/testing`: each decision's first attempt, with the states it was asked in, its `options`, and its `metadata`. Compare events without caring about key order. If the decider reads anything outside the request (the ticket, the tool history), store that in the row too. ```ts -// Your helper: rows for one decision state, from traces or a fixture file. -const rows: Array<{ id: string; request: AgentExecutorDecisionRequest; expected: ChosenEvent }> = loadDecisionRows("deciding"); +import { getDecisionRequests } from "@statelyai/agent/testing"; + +// One row per recorded decision; `expected` is the label a person confirmed for that scenario. +const rows = traces.flatMap(({ scenario, trace }) => + getDecisionRequests(trace, { name: "chooseResolution" }).map(({ request }) => ({ id: scenario.id, request, expected: labels[scenario.id] })), +); test.each(rows)("deciding: $id", async ({ request, expected }) => { const { event } = await decide(request); // the policy decider, or the model executor diff --git a/skills/improve-machine/SKILL.md b/skills/improve-machine/SKILL.md index 88121538..9f0cd629 100644 --- a/skills/improve-machine/SKILL.md +++ b/skills/improve-machine/SKILL.md @@ -37,7 +37,7 @@ The rounds are the same for both formats. What differs is what the diff may touc **No policy yet.** `loop-to-machine` leaves a single model decision inside the environment when the old code had no policy logic. Then: - Record `v0` in a new `CHANGELOG.md` next to the environment (in a `policy/` directory): "the model decides directly". This entry goes in the working tree. -- The first recommended candidate introduces the policy as `v1`: a policy file that reproduces the current behavior (it passes the environment's prompt and facts through to the model) plus the one rule the diagnosis found, and the routing line in the runner (`withPolicy(modelExecutors)`). The pattern is in `../loop-to-machine/policy.md`. +- The first recommended candidate introduces the policy as `v1`: a policy file that reproduces the current behavior (it passes the environment's prompt and facts through to the model) plus the one rule the diagnosis found, and the routing line in the runner (`withPolicy(modelExecutors)`). The policy runs as the decider through `createPolicyDecider` from `@statelyai/agent`; the pattern is in `../loop-to-machine/policy.md`. - Its `description` is the environment's goal with measurable success criteria. If the environment's goal has none, draft them from the confirmed objective and list the draft under questions for the user. - If the policy needs a fact the environment does not pass in the decision's `metadata.facts`, that is an environment change request. @@ -152,7 +152,7 @@ Run every check; a failure's output is the repair prompt. Cap repairs at 3. 1. **Environment unchanged:** every environment file hashes as recorded. A change is never repaired; discard the candidate. 2. **Types:** `tsc --noEmit` passes. 3. **Lint:** `lintAgentMachine(policy)` from `@statelyai/agent/testing` reports no errors. -4. **Paths:** for probe inputs built from recorded decision requests (one per training scenario that reaches the decision: its facts and offered options), `canReach(policy, (snapshot) => snapshot.status === "done" && !isOfferedEvent(snapshot.output), { input })` is `{ reachable: false, complete: true }`: every final output is one of the offered events. Pass `outputs` for any actor the policy invokes, and `errors: { "agent.decide": new Error("down") }` so the fallback path is checked too. +4. **Paths:** for probe inputs built from recorded decision requests (`getDecisionRequests(trace, { name })` from `@statelyai/agent/testing`, one per training scenario that reaches the decision: its facts and offered options), `canReach(policy, (snapshot) => snapshot.status === "done" && !isOfferedEvent(snapshot.output), { input })` is `{ reachable: false, complete: true }`: every final output is one of the offered events. Pass `outputs` for any actor the policy invokes, and `errors: { "agent.decide": new Error("down") }` so the fallback path is checked too. 5. **Source rules:** the policy file is still one `setupAgent` machine; its imports from the environment are only schemas, types, constants, and exported guard functions; it names no actor the environment owns. These are source checks: read the diff, or grep its imports and actor names. 6. **Overfit:** the diff contains no literal from the scenarios (ids, ticket text, customer ids, order ids). A branch that matches one scenario's data is memorization; discard it. 7. **Existing evals:** the rule and decision evals from `add-evals` still pass. Tests that pin measured numbers of the current version (a baseline failure list, a call count) are expected to change; update them in the same pull request and list each change in its body. diff --git a/skills/loop-to-machine/policy.md b/skills/loop-to-machine/policy.md index 812f136b..e928fb81 100644 --- a/skills/loop-to-machine/policy.md +++ b/skills/loop-to-machine/policy.md @@ -32,7 +32,7 @@ export const deskEvents = { ```ts import { z } from "zod"; -import { runAgent, setupAgent, type AgentDecisionExecutor, type AgentRequestExecutors } from "@statelyai/agent"; +import { createPolicyDecider, setupAgent, type AgentDecisionExecutor, type AgentRequestExecutors } from "@statelyai/agent"; import { deskEvents, factsSchema } from "./desk.js"; /** One proposal: an environment event with its `type`. */ @@ -119,22 +119,13 @@ export const policyMachine = policySetup.createMachine({ }, }); -/** Runs the policy as the environment's decider. */ +/** + * Runs the policy as the environment's decider. createPolicyDecider passes the request's + * `metadata.facts`, offered events, `system`, and `prompt` as the policy's input (its input + * schema validates them), and returns the policy's output event with its token usage. + */ export function policyDecider(executors: AgentRequestExecutors): AgentDecisionExecutor { - return async (request, info) => { - const result = await runAgent(policyMachine, { - input: { - facts: factsSchema.parse(request.metadata?.facts), - options: (request.options ?? []).map((option) => proposalSchema.parse(option.event)), - system: request.system, - prompt: request.prompt, - }, - executors, - signal: info?.signal, - }); - if (result.status !== "done") throw new Error(`policy settled '${result.status}'`); - return { event: result.output }; - }; + return createPolicyDecider(policyMachine, { executors }); } ``` diff --git a/src/decision-requests.test.ts b/src/decision-requests.test.ts new file mode 100644 index 00000000..b1618501 --- /dev/null +++ b/src/decision-requests.test.ts @@ -0,0 +1,90 @@ +import { describe, expect, test } from "vitest"; +import { z } from "zod"; +import { getDecisionRequests } from "./decision-requests.js"; +import type { AgentDecisionExecutor } from "./decision.js"; +import { runAgent, serializeTraceEvent, type AgentTraceEvent } from "./run-agent.js"; +import { setupAgent } from "./setup-agent.js"; + +const desk = setupAgent({ + context: z.object({ requested: z.number() }), + input: z.object({ requested: z.number() }), + events: { REFUND: z.object({ amount: z.number() }), ESCALATE: z.object({}), NOTE: z.object({}) }, +}).createMachine({ + context: ({ input }) => input, + initial: "triaging", + states: { + triaging: { + invoke: { + src: "agent.decide", + input: { model: "m", name: "triage", allowedEvents: ["NOTE"] }, + }, + on: { NOTE: { target: "deciding" } }, + }, + deciding: { + invoke: { + src: "agent.decide", + input: ({ context }) => ({ + model: "m", + name: "chooseResolution", + allowedEvents: ["REFUND", "ESCALATE"], + payloads: { REFUND: [{ amount: context.requested }, { amount: 1000 }] }, + metadata: { facts: { requested: context.requested } }, + }), + }, + on: { + REFUND: ({ context, event }) => + event.amount <= context.requested ? { target: "refunded" } : undefined, + ESCALATE: { target: "escalated" }, + }, + }, + refunded: { type: "final" }, + escalated: { type: "final" }, + }, +}); + +/** Notes, then proposes an illegal refund once before the legal one, so the decision is retried. */ +const decide: AgentDecisionExecutor = async (request) => { + if (request.name === "triage") return { event: { type: "NOTE" } }; + return { + event: + request.attempts.length === 0 ? { type: "REFUND", amount: 999 } : request.options![0]!.event, + }; +}; + +async function traceOf() { + const trace: AgentTraceEvent[] = []; + const result = await runAgent(desk, { + input: { requested: 40 }, + executors: { decide }, + onTrace: (event) => trace.push(event), + }); + expect(result.snapshot.value).toBe("refunded"); + return trace; +} + +describe("getDecisionRequests", () => { + test("lists each decision once, on its first attempt, with the state it was asked in", async () => { + const requests = getDecisionRequests(await traceOf()); + + expect(requests.map((r) => [r.states, r.request.name, r.request.attempts.length])).toEqual([ + [["triaging"], "triage", 0], + [["deciding"], "chooseResolution", 0], + ]); + }); + + test("reads a serialized trace, keeping the offered options and the facts", async () => { + const serialized = JSON.parse( + JSON.stringify((await traceOf()).map((e) => serializeTraceEvent(e))), + ); + const [row] = getDecisionRequests(serialized, { name: "chooseResolution" }); + + expect(row?.states).toEqual(["deciding"]); + expect(row?.request).toMatchObject({ + name: "chooseResolution", + metadata: { facts: { requested: 40 } }, + options: [{ event: { type: "REFUND", amount: 40 } }, { event: { type: "ESCALATE" } }], + }); + // The guard-refused option is on the request as `withheld`, not `options`. + expect(row?.request.withheld).toEqual([{ type: "REFUND", amount: 1000 }]); + }); +}); diff --git a/src/decision-requests.ts b/src/decision-requests.ts new file mode 100644 index 00000000..4cd3d49b --- /dev/null +++ b/src/decision-requests.ts @@ -0,0 +1,53 @@ +import type { AgentExecutorDecisionRequest } from "./decision.js"; +import { isFirstDecision, statePaths, valueOf } from "./internal/trace-states.js"; + +/** One decision a run asked for. See {@link getDecisionRequests}. */ +export interface TracedDecisionRequest { + /** The active states when the decision was asked, as dot paths (one per parallel region). */ + states: string[]; + /** The request as the executor received it, on its first attempt. */ + request: AgentExecutorDecisionRequest; +} + +/** Options for {@link getDecisionRequests}. */ +export interface GetDecisionRequestsOptions { + /** Keep only requests with this `name` (the `agent.decide` input's `name`). */ + name?: string; +} + +/** + * The decisions a run asked for, read from its trace: each decision's first + * `request.start`, with the states the run was in. A trace is one run's events, + * live or serialized with `serializeTraceEvent`. Retries of the same decision + * are left out, so each decision appears once. + * + * Use the requests as eval rows (run a decider on each and compare its event + * with a label) or as probe inputs for a policy. A serialized request keeps its + * `options`, `metadata`, `system`, and `prompt`; event schemas do not survive + * serialization, so replay closed choices (`options`) rather than open ones. + * + * @example + * ```ts no-check + * const rows = traces.flatMap((trace) => getDecisionRequests(trace, { name: "chooseResolution" })); + * for (const { request } of rows) console.log((await decide(request)).event); + * ``` + */ +export function getDecisionRequests( + trace: Iterable<{ readonly type: string }>, + options: GetDecisionRequestsOptions = {}, +): TracedDecisionRequest[] { + const requests: TracedDecisionRequest[] = []; + let current: string[] = []; + for (const event of trace as Iterable>) { + if (event.type === "machine.transition") { + current = statePaths(valueOf(event.snapshot)); + } else if (event.type === "request.start" && isFirstDecision(event.request)) { + if (options.name !== undefined && event.request.name !== options.name) continue; + requests.push({ + states: [...current], + request: event.request as unknown as AgentExecutorDecisionRequest, + }); + } + } + return requests; +} diff --git a/src/fold-traces.ts b/src/fold-traces.ts index e8e8d06c..609cb1c4 100644 --- a/src/fold-traces.ts +++ b/src/fold-traces.ts @@ -1,4 +1,4 @@ -import { isRecord } from "./internal/is-record.js"; +import { isFirstDecision, statePaths, valueOf } from "./internal/trace-states.js"; /** One state's numbers across many runs. See {@link foldTraces}. */ export interface StateStats { @@ -81,24 +81,3 @@ export function foldTraces( } return [...stats.values()]; } - -function valueOf(holder: unknown): unknown { - return isRecord(holder) ? holder.value : undefined; -} - -function isFirstDecision(request: unknown): request is Record { - return ( - isRecord(request) && - request.kind === "decision" && - (!Array.isArray(request.attempts) || request.attempts.length === 0) - ); -} - -// `'a'` → `['a']`; `{ a: 'b' }` → `['a.b']`; `{ p: { x: 'one', y: {} } }` → `['p.x.one', 'p.y']`. -function statePaths(value: unknown, prefix = ""): string[] { - if (typeof value === "string") return [prefix + value]; - if (!isRecord(value)) return []; - const entries = Object.entries(value); - if (entries.length === 0) return prefix ? [prefix.slice(0, -1)] : []; - return entries.flatMap(([key, child]) => statePaths(child, `${prefix}${key}.`)); -} diff --git a/src/index.ts b/src/index.ts index d8eaaac2..d9d81d58 100644 --- a/src/index.ts +++ b/src/index.ts @@ -93,6 +93,8 @@ export type { AgentRequest, AgentStepRequest } from "./steps.js"; export { AGENT_USAGE_EVENT_TYPE } from "./usage.js"; export { describeMachine } from "./describe-machine.js"; export { decideNext } from "./decide-next.js"; +export { createPolicyDecider } from "./policy-decider.js"; +export type { PolicyDeciderOptions } from "./policy-decider.js"; export type { AgentUsageEvent } from "./usage.js"; // The event log itself (replay, forking, hand-built entries, stores) lives in // `@statelyai/agent/log`. The root keeps only what the agent runtime reads and diff --git a/src/internal/trace-states.ts b/src/internal/trace-states.ts new file mode 100644 index 00000000..95edd9be --- /dev/null +++ b/src/internal/trace-states.ts @@ -0,0 +1,28 @@ +import { isRecord } from "./is-record.js"; + +/** The `value` of a trace event's `from` or `snapshot`. @internal */ +export function valueOf(holder: unknown): unknown { + return isRecord(holder) ? holder.value : undefined; +} + +/** A decision request on its first attempt (no failed attempts yet). @internal */ +export function isFirstDecision(request: unknown): request is Record { + return ( + isRecord(request) && + request.kind === "decision" && + (!Array.isArray(request.attempts) || request.attempts.length === 0) + ); +} + +/** + * Every active leaf of a state value as a dot path: + * `'a'` → `['a']`; `{ a: 'b' }` → `['a.b']`; `{ p: { x: 'one', y: {} } }` → `['p.x.one', 'p.y']`. + * @internal + */ +export function statePaths(value: unknown, prefix = ""): string[] { + if (typeof value === "string") return [prefix + value]; + if (!isRecord(value)) return []; + const entries = Object.entries(value); + if (entries.length === 0) return prefix ? [prefix.slice(0, -1)] : []; + return entries.flatMap(([key, child]) => statePaths(child, `${prefix}${key}.`)); +} diff --git a/src/policy-decider.test.ts b/src/policy-decider.test.ts new file mode 100644 index 00000000..960855e0 --- /dev/null +++ b/src/policy-decider.test.ts @@ -0,0 +1,223 @@ +import { describe, expect, test } from "vitest"; +import { createMachine } from "xstate"; +import { z } from "zod"; +import type { AgentDecisionExecutor, AgentExecutorDecisionRequest } from "./decision.js"; +import { createPolicyDecider } from "./policy-decider.js"; +import { runAgent } from "./run-agent.js"; +import { setupAgent } from "./setup-agent.js"; + +const proposal = z.discriminatedUnion("type", [ + z.object({ type: z.literal("REFUND"), amount: z.number() }), + z.object({ type: z.literal("ESCALATE") }), +]); +const facts = z.object({ requested: z.number() }); + +// The environment: one closed-choice decision, facts passed as metadata. +const desk = setupAgent({ + context: z.object({ requested: z.number() }), + input: z.object({ requested: z.number() }), + events: { REFUND: z.object({ amount: z.number() }), ESCALATE: z.object({}) }, +}).createMachine({ + context: ({ input }) => input, + initial: "deciding", + states: { + deciding: { + invoke: { + src: "agent.decide", + input: ({ context }) => ({ + model: "decider", + name: "chooseResolution", + system: "Desk rules.", + prompt: `Asked: ${context.requested}`, + allowedEvents: ["REFUND", "ESCALATE"], + payloads: { REFUND: [{ amount: context.requested }] }, + metadata: { facts: { requested: context.requested } }, + }), + }, + on: { + REFUND: ({ event }) => (event.amount <= 500 ? { target: "refunded" } : undefined), + ESCALATE: { target: "escalated" }, + }, + }, + refunded: { type: "final" }, + escalated: { type: "final" }, + }, +}); + +// The policy: refund small amounts with no model call; otherwise ask the model. +const policy = setupAgent({ + input: z.object({ + facts, + options: z.array(proposal), + system: z.string().optional(), + prompt: z.string().optional(), + }), + context: z.object({ + facts, + options: z.array(proposal), + system: z.string().optional(), + prompt: z.string().optional(), + chosen: proposal.nullable(), + }), + output: proposal, + events: { REFUND: z.object({ amount: z.number() }), ESCALATE: z.object({}) }, +}).createMachine({ + version: "1", + context: ({ input }) => ({ ...input, chosen: null }), + initial: "checking", + states: { + checking: { + type: "choice", + choice: ({ context }) => { + const refund = context.options.find((o) => o.type === "REFUND"); + return refund && context.facts.requested <= 100 + ? { target: "chosen", context: { chosen: refund } } + : { target: "asking" }; + }, + }, + asking: { + invoke: { + src: "agent.decide", + input: ({ context }) => ({ + model: "decider", + name: "pickResolution", + system: context.system, + prompt: context.prompt ?? "", + allowedEvents: ["REFUND", "ESCALATE"], + payloads: { + REFUND: context.options.flatMap((o) => + o.type === "REFUND" ? [{ amount: o.amount }] : [], + ), + }, + metadata: { facts: context.facts }, + }), + onError: { target: "gaveUp" }, + }, + on: { + REFUND: ({ event }) => ({ + target: "chosen", + context: { chosen: { type: "REFUND" as const, amount: event.amount } }, + }), + ESCALATE: { target: "chosen", context: { chosen: { type: "ESCALATE" as const } } }, + }, + }, + chosen: { type: "final", output: ({ context }) => context.chosen! }, + gaveUp: { type: "final", output: { type: "ESCALATE" as const } }, + }, +}); + +/** A model decider that records what it was asked and answers `answer`. */ +function model(answer: "first" | "ESCALATE", usage = { totalTokens: 7 }) { + const seen: AgentExecutorDecisionRequest[] = []; + const decide: AgentDecisionExecutor = async (request) => { + seen.push(request); + const event = answer === "first" ? request.options![0]!.event : { type: "ESCALATE" }; + return { event, usage }; + }; + return { decide, seen }; +} + +describe("createPolicyDecider", () => { + test("answers with a rule and no model call when the policy can decide alone", async () => { + const inner = model("ESCALATE"); + const result = await runAgent(desk, { + input: { requested: 40 }, + executors: { decide: createPolicyDecider(policy, { executors: { decide: inner.decide } }) }, + }); + + expect(result.status).toBe("done"); + expect(result.snapshot.value).toBe("refunded"); + expect(inner.seen).toEqual([]); + }); + + test("passes the request's facts, options, system, and prompt to the policy, and on to the model", async () => { + const inner = model("first"); + const result = await runAgent(desk, { + input: { requested: 240 }, + executors: { decide: createPolicyDecider(policy, { executors: { decide: inner.decide } }) }, + }); + + expect(result.snapshot.value).toBe("refunded"); + expect(inner.seen).toHaveLength(1); + expect(inner.seen[0]).toMatchObject({ + name: "pickResolution", + system: "Desk rules.", + prompt: "Asked: 240", + metadata: { facts: { requested: 240 } }, + options: [{ event: { type: "REFUND", amount: 240 } }, { event: { type: "ESCALATE" } }], + }); + }); + + test("reports the policy's token usage, and none when it made no model call", async () => { + const decide = createPolicyDecider(policy, { + executors: { decide: model("first", { totalTokens: 12 }).decide }, + }); + const request = (requested: number) => + ({ + kind: "decision", + id: "d", + name: "chooseResolution", + model: "decider", + events: [{ type: "REFUND" }, { type: "ESCALATE" }], + options: [ + { event: { type: "REFUND", amount: requested } }, + { event: { type: "ESCALATE" } }, + ], + attempts: [], + metadata: { facts: { requested } }, + }) as unknown as AgentExecutorDecisionRequest; + + expect(await decide(request(240))).toEqual({ + event: { type: "REFUND", amount: 240 }, + usage: { totalTokens: 12 }, + }); + expect(await decide(request(40))).toEqual({ event: { type: "REFUND", amount: 40 } }); + }); + + test("builds the policy's input with `input` when the default shape does not fit", async () => { + const seenInput: unknown[] = []; + const decide = createPolicyDecider(policy, { + executors: { decide: model("first").decide }, + input: (request) => { + const input = { + facts: { requested: Number((request.metadata as { amount: number }).amount) }, + options: (request.options ?? []).map((o) => proposal.parse(o.event)), + }; + seenInput.push(input); + return input; + }, + }); + + const result = await decide({ + kind: "decision", + id: "d", + name: "x", + model: "decider", + events: [{ type: "REFUND" }], + options: [{ event: { type: "REFUND", amount: 30 } }], + attempts: [], + metadata: { amount: 30 }, + } as unknown as AgentExecutorDecisionRequest); + + expect(result.event).toEqual({ type: "REFUND", amount: 30 }); + expect(seenInput).toEqual([ + { facts: { requested: 30 }, options: [{ type: "REFUND", amount: 30 }] }, + ]); + }); + + test("throws when the policy does not finish with an event", async () => { + // A policy that waits for a person never settles 'done', so it has no event to return. + const stuck = createMachine({ initial: "waiting", states: { waiting: {} } }); + const decide = createPolicyDecider(stuck, { executors: {} }); + await expect( + decide({ + kind: "decision", + id: "d", + name: "x", + model: "m", + events: [], + attempts: [], + } as unknown as AgentExecutorDecisionRequest), + ).rejects.toThrow(/policy .* settled 'idle'/); + }); +}); diff --git a/src/policy-decider.ts b/src/policy-decider.ts new file mode 100644 index 00000000..400e22a4 --- /dev/null +++ b/src/policy-decider.ts @@ -0,0 +1,79 @@ +import type { AnyStateMachine, InputFrom } from "xstate"; +import type { AgentDecisionExecutor, AgentExecutorDecisionRequest } from "./decision.js"; +import { AgentError } from "./errors.js"; +import { isRecord } from "./internal/is-record.js"; +import { runAgent, type AgentRunOptions } from "./run-agent.js"; + +/** Options for {@link createPolicyDecider}: everything {@link runAgent} takes for the policy run, except how the run opens. */ +export interface PolicyDeciderOptions extends Omit< + AgentRunOptions, + "input" | "event" | "attempts" | "snapshot" | "events" | "onIdle" | "signal" +> { + /** + * Builds the policy's input from the decision request. Default: + * `{ facts: request.metadata?.facts, options, system: request.system, prompt: request.prompt }`, + * where `options` are the offered events (`request.options[].event`). The + * policy's input schema validates it. + */ + input?: (request: AgentExecutorDecisionRequest) => InputFrom; +} + +/** + * Runs a policy machine as a decide executor: each decision request starts one + * run of `policy`, and its output is the chosen event. The policy can answer + * with a rule (a `choice` over the facts, no model call) and invoke + * `agent.decide` for the rest, answered by `options.executors`. + * + * The environment still judges the event: one it does not accept is retried + * like any other decider's choice. The policy's token usage is reported on the + * result, and left out when the policy made no model call. A run that does not + * end `done` with an event throws `AgentError` with code `policy-not-done`. + * + * @example + * ```ts no-check + * const decide = createPolicyDecider(policyMachine, { executors: modelExecutors }); + * await runAgent(deskMachine, { input, executors: { ...modelExecutors, decide } }); + * ``` + */ +export function createPolicyDecider( + policy: TPolicy, + options: PolicyDeciderOptions = {}, +): AgentDecisionExecutor { + const { input = defaultInput, ...runOptions } = options; + return async (request, info) => { + const result = await runAgent(policy, { + ...(runOptions as AgentRunOptions), + input: input(request), + ...(info?.signal ? { signal: info.signal } : {}), + }); + const id = (policy.config as { id?: string }).id ?? policy.id; + if (result.status !== "done") { + const why = result.status === "error" ? `: ${String(result.error)}` : ""; + throw new AgentError("policy-not-done", `policy '${id}' settled '${result.status}'${why}`, { + cause: result.status === "error" ? result.error : undefined, + }); + } + const event: unknown = result.output; + if (!isRecord(event) || typeof event.type !== "string") { + throw new AgentError( + "policy-not-done", + `policy '${id}' output is not an event: ${JSON.stringify(event)}`, + ); + } + const { modelCalls, ...usage } = result.usage; + return { event: event as { type: string }, ...(modelCalls > 0 ? { usage } : {}) }; + }; +} + +function defaultInput( + request: AgentExecutorDecisionRequest, +): InputFrom { + const facts = (request.metadata as { facts?: unknown } | undefined)?.facts; + // Absent keys stay absent: an `undefined` in a run's input keeps it from being journaled. + return { + ...(facts === undefined ? {} : { facts }), + options: (request.options ?? []).map((option) => option.event), + ...(request.system === undefined ? {} : { system: request.system }), + ...(request.prompt === undefined ? {} : { prompt: request.prompt }), + } as InputFrom; +} diff --git a/src/testing/index.ts b/src/testing/index.ts index 055c3272..9cdcc142 100644 --- a/src/testing/index.ts +++ b/src/testing/index.ts @@ -49,6 +49,8 @@ export type { } from "../seam.js"; export { foldTraces } from "../fold-traces.js"; export type { FoldTracesOptions, StateStats } from "../fold-traces.js"; +export { getDecisionRequests } from "../decision-requests.js"; +export type { GetDecisionRequestsOptions, TracedDecisionRequest } from "../decision-requests.js"; export { compareRuns } from "../compare-runs.js"; export type { CompareRunsOptions, From 015895875b2989c3dabadd02f3970c78c4bff578 Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 23 Oct 2026 12:18:59 +0300 Subject: [PATCH 2/4] fix(createPolicyDecider): a thread per decision; JSON-only policy input --- .changeset/policy-decider-helpers.md | 2 +- docs/decisions.md | 6 ++- skills/add-evals/SKILL.md | 9 +++- skills/add-traces/SKILL.md | 2 +- src/policy-decider.test.ts | 71 +++++++++++++++++++++++++++- src/policy-decider.ts | 61 +++++++++++++++++++----- 6 files changed, 133 insertions(+), 18 deletions(-) diff --git a/.changeset/policy-decider-helpers.md b/.changeset/policy-decider-helpers.md index 82ec480b..cbb40e4c 100644 --- a/.changeset/policy-decider-helpers.md +++ b/.changeset/policy-decider-helpers.md @@ -2,6 +2,6 @@ "@statelyai/agent": minor --- -**`createPolicyDecider` runs a policy machine as a decide executor.** Each decision request starts one policy run, whose output event is the choice. The input defaults to the request's `metadata.facts`, offered events, `system`, and `prompt`; the policy's token usage is reported on the result. A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. +**`createPolicyDecider` runs a policy machine as a decide executor.** Each decision request starts one policy run, whose output event is the choice. The input defaults to the request's `metadata.facts`, offered events, `system`, and `prompt`; the policy's token usage is reported on the result. A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. With a `store`, each decision's policy run gets its own thread (`policy:` by default, or `threadId: (request, info) => ...`). The input must be JSON: the default drops `undefined` keys, and anything else that is not JSON throws before the run. **`getDecisionRequests` reads the decisions a run asked for from its trace.** From `@statelyai/agent/testing`: `getDecisionRequests(trace, { name? })` returns each decision's first-attempt request with the states it was asked in, from a live or serialized trace, for eval rows and policy probes. diff --git a/docs/decisions.md b/docs/decisions.md index e4dc198b..d74a7955 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -273,9 +273,11 @@ await runAgent(deskMachine, { input, executors: { ...modelExecutors, decide } }) - The policy's own `agent.decide` invokes are answered by `executors`. Pass the environment's `metadata` on to them, so the model sees what it would without the policy. - The result carries the policy's token usage, and none when it decided with a rule. - A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. Give each of the policy's decisions an `onError` that falls back to an event the environment always accepts. -- Every other `runAgent` option passes through: `store`, `threadId`, and `metadata` to journal the policy's runs, `onTrace` to trace them. +- Every other `runAgent` option passes through: `store` and `metadata` to journal the policy's runs, `onTrace` to trace them. +- Each decision's policy run gets its own `store` thread, `policy:${info.callKey}` by default: a thread that already has entries is resumed, so a shared one would return an earlier choice. Pass `threadId: (request, info) => ...` to name them yourself; it is required with a `store` when the decider is called without a `callKey`. +- The policy's input must be JSON, since it opens the run's event log. The default drops keys whose value is `undefined`; any other value that is not JSON throws before the run. -To route by request name, wrap it: `decide: (request, info) => (request.name === "chooseResolution" ? policy(request, info) : modelExecutors.decide!(request, info))`. +To route by request name, wrap it: `decide: (request, info) => (request.name === "chooseResolution" ? policyDecide(request, info) : modelExecutors.decide!(request, info))`, where `policyDecide` is the executor `createPolicyDecider` returned. ## Decisions at the boundary of a plain machine diff --git a/skills/add-evals/SKILL.md b/skills/add-evals/SKILL.md index 5dfefc23..815c40c7 100644 --- a/skills/add-evals/SKILL.md +++ b/skills/add-evals/SKILL.md @@ -89,9 +89,14 @@ Use `explorePaths` to assert the set of reachable final states, and `await simul ```ts import { getDecisionRequests } from "@statelyai/agent/testing"; -// One row per recorded decision; `expected` is the label a person confirmed for that scenario. +// One row per recorded decision; `expected` is the label a person confirmed for that decision. +// A scenario can decide more than once, so label each decision by its position, not the scenario alone. const rows = traces.flatMap(({ scenario, trace }) => - getDecisionRequests(trace, { name: "chooseResolution" }).map(({ request }) => ({ id: scenario.id, request, expected: labels[scenario.id] })), + getDecisionRequests(trace, { name: "chooseResolution" }).map(({ request }, i) => ({ + id: `${scenario.id}#${i}`, + request, + expected: labels[`${scenario.id}#${i}`], + })), ); test.each(rows)("deciding: $id", async ({ request, expected }) => { diff --git a/skills/add-traces/SKILL.md b/skills/add-traces/SKILL.md index 290992b5..7109bab1 100644 --- a/skills/add-traces/SKILL.md +++ b/skills/add-traces/SKILL.md @@ -100,7 +100,7 @@ await writeFile(`runs/${file}.trace.json`, JSON.stringify(trace)); Decide with the user whether `runs/` is committed or gitignored. - In production, send them to OpenTelemetry instead: `createOtelTraceHandler({ tracer })` from `@statelyai/agent/otel` is an `onTrace` handler. -- **A policy run inside a decider** is a separate run. Pass it the same `onTrace` (and its own `store` thread, `:policy:`) if you want per-state stats for the policy too. +- **A policy run inside a decider** is a separate run. Pass it the same `onTrace` (and a `store`) if you want per-state stats for the policy too. `createPolicyDecider` gives each decision's policy run its own thread, `policy:`; never give it one fixed `threadId`, or the second decision resumes the first. - **A plain XState machine** (built with `createMachine` or `setup`, decided with `decideNext`): drive it with `createAgentRuntime` and `runToQuiescence`, which accept plain machines, and pass each `decideNext` result as `{ event, attempts }` so refused choices reach the log. Push `{ type: "request.start", request: result.request }` onto the trace before sending the event, since a host-driven decision emits no trace itself. - **A plain `createActor` run you cannot change:** attach `recordLog(actor)` from `@statelyai/agent/log` before `actor.start()`, and pass `inspect: traceTransitions((event) => trace.push(serializeTraceEvent(event)))` to `createActor`. `recordLog` cannot record decision attempts. diff --git a/src/policy-decider.test.ts b/src/policy-decider.test.ts index 960855e0..70570210 100644 --- a/src/policy-decider.test.ts +++ b/src/policy-decider.test.ts @@ -2,6 +2,7 @@ import { describe, expect, test } from "vitest"; import { createMachine } from "xstate"; import { z } from "zod"; import type { AgentDecisionExecutor, AgentExecutorDecisionRequest } from "./decision.js"; +import { createInMemoryEventLogStore } from "./event-log-store.js"; import { createPolicyDecider } from "./policy-decider.js"; import { runAgent } from "./run-agent.js"; import { setupAgent } from "./setup-agent.js"; @@ -10,7 +11,8 @@ const proposal = z.discriminatedUnion("type", [ z.object({ type: z.literal("REFUND"), amount: z.number() }), z.object({ type: z.literal("ESCALATE") }), ]); -const facts = z.object({ requested: z.number() }); +// `coupon` is optional: a present key holding undefined survives the schema. +const facts = z.object({ requested: z.number(), coupon: z.string().optional() }); // The environment: one closed-choice decision, facts passed as metadata. const desk = setupAgent({ @@ -205,6 +207,73 @@ describe("createPolicyDecider", () => { ]); }); + const decisionRequest = (facts: Record, requested = facts.requested as number) => + ({ + kind: "decision", + id: "d", + name: "chooseResolution", + model: "decider", + events: [{ type: "REFUND" }, { type: "ESCALATE" }], + options: [{ event: { type: "REFUND", amount: requested } }, { event: { type: "ESCALATE" } }], + attempts: [], + metadata: { facts }, + }) as unknown as AgentExecutorDecisionRequest; + + test("gives every decision its own policy thread, so a store never replays an earlier choice", async () => { + const store = createInMemoryEventLogStore(); + const decide = createPolicyDecider(policy, { + executors: { decide: model("ESCALATE").decide }, + store, + }); + + const first = await decide(decisionRequest({ requested: 240 }), { callKey: "exec:site#0" }); + const second = await decide(decisionRequest({ requested: 40 }), { callKey: "exec:site#1" }); + + expect(first.event).toEqual({ type: "ESCALATE" }); + expect(second.event).toEqual({ type: "REFUND", amount: 40 }); + const { threadIds } = await store.listThreads!({ prefix: "policy:" }); + expect(threadIds.sort()).toEqual(["policy:exec:site#0", "policy:exec:site#1"]); + }); + + test("takes the policy thread from `threadId`, and needs one when a store has no call key to go on", async () => { + const store = createInMemoryEventLogStore(); + const decide = createPolicyDecider(policy, { + executors: { decide: model("ESCALATE").decide }, + store, + threadId: (request) => + `policy:${String((request.metadata as { facts: { requested: number } }).facts.requested)}`, + }); + await decide(decisionRequest({ requested: 240 })); + expect((await store.listThreads!({ prefix: "policy:" })).threadIds).toEqual(["policy:240"]); + + const without = createPolicyDecider(policy, { + executors: { decide: model("ESCALATE").decide }, + store, + }); + await expect(without(decisionRequest({ requested: 240 }))).rejects.toThrow(/threadId/); + }); + + test("drops undefined keys nested in the default input, so the policy's log and usage survive", async () => { + const decide = createPolicyDecider(policy, { + executors: { decide: model("first", { totalTokens: 12 }).decide }, + }); + + const result = await decide(decisionRequest({ requested: 240, coupon: undefined })); + + expect(result).toEqual({ event: { type: "REFUND", amount: 240 }, usage: { totalTokens: 12 } }); + }); + + test("rejects a policy input that is not JSON before the policy runs", async () => { + const inner = model("first"); + const decide = createPolicyDecider(policy, { + executors: { decide: inner.decide }, + input: () => ({ facts: { requested: 240, at: new Date() }, options: [] }) as never, + }); + + await expect(decide(decisionRequest({ requested: 240 }))).rejects.toThrow(/at/); + expect(inner.seen).toEqual([]); + }); + test("throws when the policy does not finish with an event", async () => { // A policy that waits for a person never settles 'done', so it has no event to return. const stuck = createMachine({ initial: "waiting", states: { waiting: {} } }); diff --git a/src/policy-decider.ts b/src/policy-decider.ts index 400e22a4..04bb3c7d 100644 --- a/src/policy-decider.ts +++ b/src/policy-decider.ts @@ -1,14 +1,25 @@ import type { AnyStateMachine, InputFrom } from "xstate"; import type { AgentDecisionExecutor, AgentExecutorDecisionRequest } from "./decision.js"; import { AgentError } from "./errors.js"; +import { assertJsonSerializable } from "./event-log.js"; import { isRecord } from "./internal/is-record.js"; import { runAgent, type AgentRunOptions } from "./run-agent.js"; +import type { AgentRequestExecutorInfo } from "./text-logic.js"; /** Options for {@link createPolicyDecider}: everything {@link runAgent} takes for the policy run, except how the run opens. */ export interface PolicyDeciderOptions extends Omit< AgentRunOptions, - "input" | "event" | "attempts" | "snapshot" | "events" | "onIdle" | "signal" + "input" | "event" | "attempts" | "snapshot" | "events" | "onIdle" | "signal" | "threadId" > { + /** + * The `store` thread for one decision's policy run. Every decision, and every + * retry of one, needs its own thread: a thread that already has entries is + * resumed, so a shared one would return an earlier decision's choice. Default + * with a `store`: `policy:${info.callKey}`, which is unique per call and + * attempt and stays the same when a crashed run re-issues the call. Required + * with a `store` when the decider is called without a `callKey`. + */ + threadId?: (request: AgentExecutorDecisionRequest, info?: AgentRequestExecutorInfo) => string; /** * Builds the policy's input from the decision request. Default: * `{ facts: request.metadata?.facts, options, system: request.system, prompt: request.prompt }`, @@ -25,7 +36,9 @@ export interface PolicyDeciderOptions extends O * `agent.decide` for the rest, answered by `options.executors`. * * The environment still judges the event: one it does not accept is retried - * like any other decider's choice. The policy's token usage is reported on the + * like any other decider's choice. The policy's input must be JSON (it opens + * the policy run's event log): the default drops keys whose value is + * `undefined`, and any other value that is not JSON throws before the run. The policy's token usage is reported on the * result, and left out when the policy made no model call. A run that does not * end `done` with an event throws `AgentError` with code `policy-not-done`. * @@ -39,14 +52,18 @@ export function createPolicyDecider( policy: TPolicy, options: PolicyDeciderOptions = {}, ): AgentDecisionExecutor { - const { input = defaultInput, ...runOptions } = options; + const { input = defaultInput, threadId, ...runOptions } = options; + const id = (policy.config as { id?: string }).id ?? policy.id; return async (request, info) => { + const policyInput = input(request); + assertJsonSerializable(policyInput, "policy input"); + const thread = runOptions.store === undefined ? undefined : policyThread(request, info); const result = await runAgent(policy, { ...(runOptions as AgentRunOptions), - input: input(request), + input: policyInput, + ...(thread === undefined ? {} : { threadId: thread }), ...(info?.signal ? { signal: info.signal } : {}), }); - const id = (policy.config as { id?: string }).id ?? policy.id; if (result.status !== "done") { const why = result.status === "error" ? `: ${String(result.error)}` : ""; throw new AgentError("policy-not-done", `policy '${id}' settled '${result.status}'${why}`, { @@ -63,17 +80,39 @@ export function createPolicyDecider( const { modelCalls, ...usage } = result.usage; return { event: event as { type: string }, ...(modelCalls > 0 ? { usage } : {}) }; }; + + function policyThread(request: AgentExecutorDecisionRequest, info?: AgentRequestExecutorInfo) { + if (threadId) return threadId(request, info); + if (info?.callKey) return `policy:${info.callKey}`; + throw new AgentError( + "missing-thread-id", + `policy '${id}' has a store but no threadId, and the decision came with no callKey to derive one from`, + ); + } } function defaultInput( request: AgentExecutorDecisionRequest, ): InputFrom { const facts = (request.metadata as { facts?: unknown } | undefined)?.facts; - // Absent keys stay absent: an `undefined` in a run's input keeps it from being journaled. - return { - ...(facts === undefined ? {} : { facts }), + // An `undefined` anywhere in a run's input keeps it from being journaled; an absent key means the same. + return withoutUndefinedKeys({ + facts, options: (request.options ?? []).map((option) => option.event), - ...(request.system === undefined ? {} : { system: request.system }), - ...(request.prompt === undefined ? {} : { prompt: request.prompt }), - } as InputFrom; + system: request.system, + prompt: request.prompt, + }) as InputFrom; +} + +/** A copy of `value` with every object key whose value is `undefined` removed, at any depth. */ +function withoutUndefinedKeys(value: unknown): unknown { + if (Array.isArray(value)) return value.map(withoutUndefinedKeys); + // Plain objects only: a Date or a class instance passes through for the JSON check to reject. + if (!isRecord(value) || ![Object.prototype, null].includes(Object.getPrototypeOf(value))) + return value; + return Object.fromEntries( + Object.entries(value) + .filter(([, child]) => child !== undefined) + .map(([key, child]) => [key, withoutUndefinedKeys(child)]), + ); } From 89e3416f2d29962c3c84f4571bb4bcb112c0e19c Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 23 Oct 2026 22:34:13 +0300 Subject: [PATCH 3/4] fix(createPolicyDecider): fingerprint the policy input in the default thread id --- .changeset/policy-decider-helpers.md | 2 +- docs/decisions.md | 2 +- src/policy-decider.test.ts | 23 ++++++++++++++++++++++- src/policy-decider.ts | 21 +++++++++++++++------ 4 files changed, 39 insertions(+), 9 deletions(-) diff --git a/.changeset/policy-decider-helpers.md b/.changeset/policy-decider-helpers.md index cbb40e4c..f5e3de71 100644 --- a/.changeset/policy-decider-helpers.md +++ b/.changeset/policy-decider-helpers.md @@ -2,6 +2,6 @@ "@statelyai/agent": minor --- -**`createPolicyDecider` runs a policy machine as a decide executor.** Each decision request starts one policy run, whose output event is the choice. The input defaults to the request's `metadata.facts`, offered events, `system`, and `prompt`; the policy's token usage is reported on the result. A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. With a `store`, each decision's policy run gets its own thread (`policy:` by default, or `threadId: (request, info) => ...`). The input must be JSON: the default drops `undefined` keys, and anything else that is not JSON throws before the run. +**`createPolicyDecider` runs a policy machine as a decide executor.** Each decision request starts one policy run, whose output event is the choice. The input defaults to the request's `metadata.facts`, offered events, `system`, and `prompt`; the policy's token usage is reported on the result. A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. With a `store`, each decision's policy run gets its own thread (`policy::` by default, or `threadId: (request, info) => ...`). The input must be JSON: the default drops `undefined` keys, and anything else that is not JSON throws before the run. **`getDecisionRequests` reads the decisions a run asked for from its trace.** From `@statelyai/agent/testing`: `getDecisionRequests(trace, { name? })` returns each decision's first-attempt request with the states it was asked in, from a live or serialized trace, for eval rows and policy probes. diff --git a/docs/decisions.md b/docs/decisions.md index d74a7955..57d3db25 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -274,7 +274,7 @@ await runAgent(deskMachine, { input, executors: { ...modelExecutors, decide } }) - The result carries the policy's token usage, and none when it decided with a rule. - A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. Give each of the policy's decisions an `onError` that falls back to an event the environment always accepts. - Every other `runAgent` option passes through: `store` and `metadata` to journal the policy's runs, `onTrace` to trace them. -- Each decision's policy run gets its own `store` thread, `policy:${info.callKey}` by default: a thread that already has entries is resumed, so a shared one would return an earlier choice. Pass `threadId: (request, info) => ...` to name them yourself; it is required with a `store` when the decider is called without a `callKey`. +- Each decision's policy run gets its own `store` thread, `policy:${info.callKey}:${hash}` by default, where `hash` fingerprints the policy input: a thread that already has entries is resumed, so a shared one would return an earlier choice. The hash keeps a fork that asks something else at the same call site off its parent's thread. Pass `threadId: (request, info) => ...` to name them yourself; it is required with a `store` when the decider is called without a `callKey`. - The policy's input must be JSON, since it opens the run's event log. The default drops keys whose value is `undefined`; any other value that is not JSON throws before the run. To route by request name, wrap it: `decide: (request, info) => (request.name === "chooseResolution" ? policyDecide(request, info) : modelExecutors.decide!(request, info))`, where `policyDecide` is the executor `createPolicyDecider` returned. diff --git a/src/policy-decider.test.ts b/src/policy-decider.test.ts index 70570210..47d57736 100644 --- a/src/policy-decider.test.ts +++ b/src/policy-decider.test.ts @@ -232,7 +232,28 @@ describe("createPolicyDecider", () => { expect(first.event).toEqual({ type: "ESCALATE" }); expect(second.event).toEqual({ type: "REFUND", amount: 40 }); const { threadIds } = await store.listThreads!({ prefix: "policy:" }); - expect(threadIds.sort()).toEqual(["policy:exec:site#0", "policy:exec:site#1"]); + expect(threadIds.map((thread) => thread.replace(/:[^:]+$/, "")).sort()).toEqual([ + "policy:exec:site#0", + "policy:exec:site#1", + ]); + }); + + test("a different request at the same call key gets its own thread; the same one resumes", async () => { + const store = createInMemoryEventLogStore(); + const decide = createPolicyDecider(policy, { + executors: { decide: model("ESCALATE").decide }, + store, + }); + + // A fork inherits the parent's call key but asks with other facts. + const parent = await decide(decisionRequest({ requested: 240 }), { callKey: "exec:site#0" }); + const fork = await decide(decisionRequest({ requested: 40 }), { callKey: "exec:site#0" }); + const reissued = await decide(decisionRequest({ requested: 40 }), { callKey: "exec:site#0" }); + + expect(parent.event).toEqual({ type: "ESCALATE" }); + expect(fork.event).toEqual({ type: "REFUND", amount: 40 }); + expect(reissued.event).toEqual({ type: "REFUND", amount: 40 }); + expect((await store.listThreads!({ prefix: "policy:" })).threadIds).toHaveLength(2); }); test("takes the policy thread from `threadId`, and needs one when a store has no call key to go on", async () => { diff --git a/src/policy-decider.ts b/src/policy-decider.ts index 04bb3c7d..b657d3e0 100644 --- a/src/policy-decider.ts +++ b/src/policy-decider.ts @@ -1,10 +1,11 @@ import type { AnyStateMachine, InputFrom } from "xstate"; import type { AgentDecisionExecutor, AgentExecutorDecisionRequest } from "./decision.js"; import { AgentError } from "./errors.js"; -import { assertJsonSerializable } from "./event-log.js"; +import { assertJsonSerializable, canonicalEventJson } from "./event-log.js"; import { isRecord } from "./internal/is-record.js"; import { runAgent, type AgentRunOptions } from "./run-agent.js"; import type { AgentRequestExecutorInfo } from "./text-logic.js"; +import { djb2Hex } from "./utils.js"; /** Options for {@link createPolicyDecider}: everything {@link runAgent} takes for the policy run, except how the run opens. */ export interface PolicyDeciderOptions extends Omit< @@ -15,8 +16,10 @@ export interface PolicyDeciderOptions extends O * The `store` thread for one decision's policy run. Every decision, and every * retry of one, needs its own thread: a thread that already has entries is * resumed, so a shared one would return an earlier decision's choice. Default - * with a `store`: `policy:${info.callKey}`, which is unique per call and - * attempt and stays the same when a crashed run re-issues the call. Required + * with a `store`: `policy:${info.callKey}:${hash}`, where `hash` fingerprints + * the policy input. The call key is unique per call and attempt and stays the + * same when a crashed run re-issues the call; the hash keeps a fork that asks + * something else at the same call site off the parent's thread. Required * with a `store` when the decider is called without a `callKey`. */ threadId?: (request: AgentExecutorDecisionRequest, info?: AgentRequestExecutorInfo) => string; @@ -57,7 +60,8 @@ export function createPolicyDecider( return async (request, info) => { const policyInput = input(request); assertJsonSerializable(policyInput, "policy input"); - const thread = runOptions.store === undefined ? undefined : policyThread(request, info); + const thread = + runOptions.store === undefined ? undefined : policyThread(request, policyInput, info); const result = await runAgent(policy, { ...(runOptions as AgentRunOptions), input: policyInput, @@ -81,9 +85,14 @@ export function createPolicyDecider( return { event: event as { type: string }, ...(modelCalls > 0 ? { usage } : {}) }; }; - function policyThread(request: AgentExecutorDecisionRequest, info?: AgentRequestExecutorInfo) { + function policyThread( + request: AgentExecutorDecisionRequest, + policyInput: unknown, + info?: AgentRequestExecutorInfo, + ) { if (threadId) return threadId(request, info); - if (info?.callKey) return `policy:${info.callKey}`; + // The call key names the call site, not the request: a fork shares it. + if (info?.callKey) return `policy:${info.callKey}:${djb2Hex(canonicalEventJson(policyInput))}`; throw new AgentError( "missing-thread-id", `policy '${id}' has a store but no threadId, and the decision came with no callKey to derive one from`, From 3fb3b028da101e6ce0cc5140c277b7744b93f32c Mon Sep 17 00:00:00 2001 From: David Khourshid Date: Fri, 23 Oct 2026 22:42:05 +0300 Subject: [PATCH 4/4] fix(createPolicyDecider): SHA-256 the policy input in the default thread id --- .changeset/policy-decider-helpers.md | 2 +- docs/decisions.md | 2 +- src/policy-decider.test.ts | 4 +++- src/policy-decider.ts | 18 ++++++++++++------ 4 files changed, 17 insertions(+), 9 deletions(-) diff --git a/.changeset/policy-decider-helpers.md b/.changeset/policy-decider-helpers.md index f5e3de71..611ff913 100644 --- a/.changeset/policy-decider-helpers.md +++ b/.changeset/policy-decider-helpers.md @@ -2,6 +2,6 @@ "@statelyai/agent": minor --- -**`createPolicyDecider` runs a policy machine as a decide executor.** Each decision request starts one policy run, whose output event is the choice. The input defaults to the request's `metadata.facts`, offered events, `system`, and `prompt`; the policy's token usage is reported on the result. A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. With a `store`, each decision's policy run gets its own thread (`policy::` by default, or `threadId: (request, info) => ...`). The input must be JSON: the default drops `undefined` keys, and anything else that is not JSON throws before the run. +**`createPolicyDecider` runs a policy machine as a decide executor.** Each decision request starts one policy run, whose output event is the choice. The input defaults to the request's `metadata.facts`, offered events, `system`, and `prompt`; the policy's token usage is reported on the result. A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. With a `store`, each decision's policy run gets its own thread (`policy::` by default, or `threadId: (request, info) => ...`). The input must be JSON: the default drops `undefined` keys, and anything else that is not JSON throws before the run. **`getDecisionRequests` reads the decisions a run asked for from its trace.** From `@statelyai/agent/testing`: `getDecisionRequests(trace, { name? })` returns each decision's first-attempt request with the states it was asked in, from a live or serialized trace, for eval rows and policy probes. diff --git a/docs/decisions.md b/docs/decisions.md index 57d3db25..119670aa 100644 --- a/docs/decisions.md +++ b/docs/decisions.md @@ -274,7 +274,7 @@ await runAgent(deskMachine, { input, executors: { ...modelExecutors, decide } }) - The result carries the policy's token usage, and none when it decided with a rule. - A run that does not end `done` with an event throws `AgentError` with code `policy-not-done`. Give each of the policy's decisions an `onError` that falls back to an event the environment always accepts. - Every other `runAgent` option passes through: `store` and `metadata` to journal the policy's runs, `onTrace` to trace them. -- Each decision's policy run gets its own `store` thread, `policy:${info.callKey}:${hash}` by default, where `hash` fingerprints the policy input: a thread that already has entries is resumed, so a shared one would return an earlier choice. The hash keeps a fork that asks something else at the same call site off its parent's thread. Pass `threadId: (request, info) => ...` to name them yourself; it is required with a `store` when the decider is called without a `callKey`. +- Each decision's policy run gets its own `store` thread, `policy:${info.callKey}:${hash}` by default, where `hash` is the SHA-256 of the policy input: a thread that already has entries is resumed, so a shared one would return an earlier choice. The hash keeps a fork that asks something else at the same call site off its parent's thread. Pass `threadId: (request, info) => ...` to name them yourself; it is required with a `store` when the decider is called without a `callKey`. - The policy's input must be JSON, since it opens the run's event log. The default drops keys whose value is `undefined`; any other value that is not JSON throws before the run. To route by request name, wrap it: `decide: (request, info) => (request.name === "chooseResolution" ? policyDecide(request, info) : modelExecutors.decide!(request, info))`, where `policyDecide` is the executor `createPolicyDecider` returned. diff --git a/src/policy-decider.test.ts b/src/policy-decider.test.ts index 47d57736..101a8227 100644 --- a/src/policy-decider.test.ts +++ b/src/policy-decider.test.ts @@ -253,7 +253,9 @@ describe("createPolicyDecider", () => { expect(parent.event).toEqual({ type: "ESCALATE" }); expect(fork.event).toEqual({ type: "REFUND", amount: 40 }); expect(reissued.event).toEqual({ type: "REFUND", amount: 40 }); - expect((await store.listThreads!({ prefix: "policy:" })).threadIds).toHaveLength(2); + const { threadIds } = await store.listThreads!({ prefix: "policy:" }); + expect(threadIds).toHaveLength(2); + expect(threadIds[0]).toMatch(/^policy:exec:site#0:[0-9a-f]{64}$/); }); test("takes the policy thread from `threadId`, and needs one when a store has no call key to go on", async () => { diff --git a/src/policy-decider.ts b/src/policy-decider.ts index b657d3e0..4b25b45f 100644 --- a/src/policy-decider.ts +++ b/src/policy-decider.ts @@ -5,7 +5,6 @@ import { assertJsonSerializable, canonicalEventJson } from "./event-log.js"; import { isRecord } from "./internal/is-record.js"; import { runAgent, type AgentRunOptions } from "./run-agent.js"; import type { AgentRequestExecutorInfo } from "./text-logic.js"; -import { djb2Hex } from "./utils.js"; /** Options for {@link createPolicyDecider}: everything {@link runAgent} takes for the policy run, except how the run opens. */ export interface PolicyDeciderOptions extends Omit< @@ -16,8 +15,8 @@ export interface PolicyDeciderOptions extends O * The `store` thread for one decision's policy run. Every decision, and every * retry of one, needs its own thread: a thread that already has entries is * resumed, so a shared one would return an earlier decision's choice. Default - * with a `store`: `policy:${info.callKey}:${hash}`, where `hash` fingerprints - * the policy input. The call key is unique per call and attempt and stays the + * with a `store`: `policy:${info.callKey}:${hash}`, where `hash` is the + * SHA-256 of the policy input's canonical JSON. The call key is unique per call and attempt and stays the * same when a crashed run re-issues the call; the hash keeps a fork that asks * something else at the same call site off the parent's thread. Required * with a `store` when the decider is called without a `callKey`. @@ -61,7 +60,7 @@ export function createPolicyDecider( const policyInput = input(request); assertJsonSerializable(policyInput, "policy input"); const thread = - runOptions.store === undefined ? undefined : policyThread(request, policyInput, info); + runOptions.store === undefined ? undefined : await policyThread(request, policyInput, info); const result = await runAgent(policy, { ...(runOptions as AgentRunOptions), input: policyInput, @@ -85,14 +84,15 @@ export function createPolicyDecider( return { event: event as { type: string }, ...(modelCalls > 0 ? { usage } : {}) }; }; - function policyThread( + async function policyThread( request: AgentExecutorDecisionRequest, policyInput: unknown, info?: AgentRequestExecutorInfo, ) { if (threadId) return threadId(request, info); // The call key names the call site, not the request: a fork shares it. - if (info?.callKey) return `policy:${info.callKey}:${djb2Hex(canonicalEventJson(policyInput))}`; + if (info?.callKey) + return `policy:${info.callKey}:${await sha256(canonicalEventJson(policyInput))}`; throw new AgentError( "missing-thread-id", `policy '${id}' has a store but no threadId, and the decision came with no callKey to derive one from`, @@ -125,3 +125,9 @@ function withoutUndefinedKeys(value: unknown): unknown { .map(([key, child]) => [key, withoutUndefinedKeys(child)]), ); } + +/** Hex SHA-256 of `text`, with Web Crypto (Node, browsers, workers). */ +async function sha256(text: string): Promise { + const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(text)); + return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""); +}