From 37175fac578364ffe094c189aa9c96175d7462f6 Mon Sep 17 00:00:00 2001 From: gcharang <21151592+gcharang@users.noreply.github.com> Date: Sat, 15 Aug 2026 07:45:21 +0400 Subject: [PATCH 1/3] fix: fail closed on guarded structured output --- .changeset/gate-structured-output.md | 12 + .github/workflows/ci.yml | 8 +- README.md | 8 +- docs/breakwater-architecture.md | 6 +- docs/breakwater-purpose-and-boundaries.md | 2 +- docs/getting-started.md | 6 +- docs/policy-engine-design.md | 14 +- .../breakwater-improvement-roadmap.md | 33 +- docs/proposals/model-gateway-policy.md | 6 +- docs/security-threat-model.md | 4 +- packages/breakwater/README.md | 29 +- packages/breakwater/package.json | 2 +- .../scripts/packed-consumer-test.mjs | 20 +- packages/breakwater/src/agent/agent.test.ts | 98 ++++++ packages/breakwater/src/agent/index.ts | 35 ++- packages/breakwater/src/index.ts | 2 + .../policy-engine/content-inspection.test.ts | 6 +- .../breakwater/src/policy-engine/index.ts | 258 +++++++++++----- .../src/policy-engine/policy-engine.test.ts | 282 ++++++++++++++++-- packages/flowsafe/README.md | 10 +- packages/flowsafe/package.json | 4 +- .../flowsafe/scripts/agent-host-pack-test.mjs | 2 +- .../flowsafe/src/agent-host/catalog.test.ts | 33 ++ packages/flowsafe/src/agent-host/catalog.ts | 7 +- .../src/agent-host/thread-host.test.ts | 7 + .../agent-runner/durable-agent-runner.test.ts | 202 +++++++++++++ .../src/agent-runner/durable-agent-runner.ts | 117 +++++++- 27 files changed, 1041 insertions(+), 172 deletions(-) create mode 100644 .changeset/gate-structured-output.md diff --git a/.changeset/gate-structured-output.md b/.changeset/gate-structured-output.md new file mode 100644 index 0000000..a4fa89b --- /dev/null +++ b/.changeset/gate-structured-output.md @@ -0,0 +1,12 @@ +--- +"@proofoftech/breakwater": minor +"@proofoftech/flowsafe": minor +--- + +Fail closed on structured-output coverage gaps. `createGuardedAgent()` rejects structured output before model execution because Mastra exposes parsed values to messages, persistence, and observability hooks before a post-generation wrapper could inspect them. It also rejects object-only policies that no supported guarded invocation can cover. + +Processor-visible object chunks are validated as JSON, evaluated through their canonical serialization, and replaced with the same canonical clone. Standalone object-only policies abort when an invocation exposes no object to the processor. Policy lists and decision-driving descriptors are snapshotted at construction, evaluator callables retain their original receiver, and per-stream audit metadata stays bounded by configured policies and channels. + +Flowsafe recognizes the new guarded-agent host protocol and rejects structured output on durable stream, generate, and prepare before Mastra can bypass the narrow handle. Durable entry points snapshot data-property call options before validation and delegation, reject accessors, and use the same snapshot for later run registration. Both packages pin their tested `@mastra/core` 1.50.0 contract. + +Hold-back cost under large streams is measured by opt-in evidence tests (`BREAKWATER_PERF=1`) and recorded in the policy-engine design guide. diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9e34a40..f3ddf37 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -133,10 +133,10 @@ jobs: # Mastra compat matrix: the libraries pin behavioral contracts to # @mastra/core internals (resume-context merge-over semantics, snapshot - # shape, the reprocess-part key, the six-table inventory). The peer range - # (^1.50.0) admits any 1.x, so run the library suites against the newest - # published 1.x -- drift becomes a CI signal here instead of a consumer's - # runtime surprise. The tripwire tests (runtime.test.ts merge pin, + # shape, the reprocess-part key, the six-table inventory). The supported + # peer is exactly 1.50.0; run the library suites against the newest + # published 1.x as a signal for a deliberate future compatibility update. + # The tripwire tests (runtime.test.ts merge pin, # ProcessorRunner reprocess-key probe, mastra-schema-guard) are the teeth. mastra-compat: runs-on: ubuntu-latest diff --git a/README.md b/README.md index 433f591..bf9abba 100644 --- a/README.md +++ b/README.md @@ -34,8 +34,8 @@ Anchorage is not a model provider, identity provider, hosted SaaS, generic proce Fleet control requires Node 22.22.0 or later. Flowsafe requires Node 22 or later. Breakwater requires Node 22.3 or later for its built-in Agent CLI -executor. All three packages are ESM-only. Breakwater and Flowsafe declare -`@mastra/core` `^1.50.0` as their peer range. React 18 or 19 is needed only for +executor. All three packages are ESM-only. Breakwater and Flowsafe require +`@mastra/core` `1.50.0`. React 18 or 19 is needed only for the optional Flowsafe approval UI. Hosts that run `flowsafe-provision` or the Fleet Control Wrangler backend must provide Wrangler `>=4.118 <5`; Wrangler is not a Flowsafe peer dependency. @@ -43,7 +43,7 @@ not a Flowsafe peer dependency. ## Start with breakwater ```bash -npm install @mastra/core@^1.50.0 @proofoftech/breakwater +npm install @mastra/core@1.50.0 @proofoftech/breakwater ``` Create a guarded agent and pass the authenticated actor through Mastra's `RequestContext`: @@ -139,7 +139,7 @@ explains every permission, required store, and accepted limit: ## Add durable approvals ```bash -npm install @mastra/core@^1.50.0 @proofoftech/breakwater @proofoftech/flowsafe +npm install @mastra/core@1.50.0 @proofoftech/breakwater @proofoftech/flowsafe ``` Start from the copy-ready baseline Worker in [`packages/flowsafe/deploy/`](packages/flowsafe/deploy/README.md). It wires one Durable Object per run, D1 snapshots and approvals, authenticated run routes, server-derived connector grants, live-streaming opt-in, SLA sweep, retention, audit export, and a sample gated workflow. diff --git a/docs/breakwater-architecture.md b/docs/breakwater-architecture.md index 49936d5..58ae577 100644 --- a/docs/breakwater-architecture.md +++ b/docs/breakwater-architecture.md @@ -98,7 +98,7 @@ output: model/tools -> app output -> policy output The durable resume line describes which `processInput` hook runs during rehydration. Before installing the registries, Flowsafe restores the complete input and LLM-request processor lists for later loop hooks. It also restores the same tools, memory, model, application output processors, and mandatory policy output processor as initial preparation. An RBAC denial stops rehydration before registry installation or resumed tool execution. -The call allowlist contains `requestContext`, `runId`, `memory`, and `abortSignal`. Unknown own properties fail even when their value is `undefined`. Construction fixes `maxSteps` and `toolChoice`, enables policy hold-back, and disables background continuations. +The call allowlist contains `requestContext`, `runId`, `memory`, and `abortSignal`. Calls are copied into frozen allowlisted snapshots, and unknown own properties fail even when their value is `undefined`. Construction fixes `maxSteps` and `toolChoice`, enables policy hold-back, and disables background continuations. Application input processors may implement only `processInput`. Application output processors must implement both stream and final-result enforcement. Processor workflows and the reserved IDs `breakwater-rbac` and `breakwater-policy-engine` fail construction. @@ -112,7 +112,9 @@ Application input processors may implement only `processInput`. Application outp - structured-object stream snapshots; - custom synchronous or asynchronous policy evaluators. -Structured-object values are available on Mastra's streaming path. Under the supported Mastra version, non-streaming `generate()` exposes structured output through answer text rather than a separate result object. A policy scoped only to `object` therefore has no non-streaming result coverage; the engine requires an audit sink for that configuration and records the limitation. +Under the supported Mastra version, structured objects parsed by `generate()` and the chunks core's `StructuredOutputProcessor` emits never pass through the agent's output processors. Mastra also copies a parsed value into messages and may send it to persistence and observability hooks before `generate()` returns. A post-generation wrapper gate is therefore not a containment boundary. The guarded agent rejects structured output before execution and rejects object-only policies at construction. A standalone `PolicyEngine` validates processor-visible object chunks as JSON, evaluates their canonical snapshots, forwards the same canonical clones, and aborts at the result boundary when an object-only policy saw no such chunk. + +The guarded handle also carries a versioned host protocol. Flowsafe checks that protocol before durable wrapping and rejects guarded structured output on every durable entry point, preserving the narrow handle's refusal even though Mastra's durable runner invokes the raw agent through processor lists. ## Streaming behavior diff --git a/docs/breakwater-purpose-and-boundaries.md b/docs/breakwater-purpose-and-boundaries.md index aa2905a..1cd665c 100644 --- a/docs/breakwater-purpose-and-boundaries.md +++ b/docs/breakwater-purpose-and-boundaries.md @@ -30,7 +30,7 @@ write. `createGuardedAgent()` is the supported in-process boundary for protected agent execution. It accepts construction-time agent configuration instead of an existing raw `Agent`, then returns a narrow handle with unstructured `generate()` and `stream()` methods. -The factory fixes the step budget and tool choice, requires an audit logger, forces streaming policy hold-back, and disables agent background continuations. A call must include trusted `requestContext`; the only other accepted keys are `runId`, `memory`, and `abortSignal`. Unknown keys fail even when their value is `undefined`. +The factory fixes the step budget and tool choice, requires an audit logger, forces streaming policy hold-back, and disables agent background continuations. A call must include trusted `requestContext`; the only other accepted top-level keys are `runId`, `memory`, and `abortSignal`. Unknown keys, including `structuredOutput`, fail even when their value is `undefined`. The direct path authorizes before application processors run. Durable preparation lists the same gates in this order: diff --git a/docs/getting-started.md b/docs/getting-started.md index 9f79b37..9434e41 100644 --- a/docs/getting-started.md +++ b/docs/getting-started.md @@ -8,7 +8,7 @@ This guide takes a Mastra application from package installation to one guarded c complete repository and private fleet-control package require Node.js 22.22.0 or later - An ESM TypeScript project using `moduleResolution: "NodeNext"`, `"Node16"`, or `"Bundler"` -- `@mastra/core` in the `^1.50.0` peer range +- `@mastra/core` `1.50.0` - A Cloudflare account, D1 database, and Durable Objects only when deploying flowsafe React is not required by flowsafe unless you import `@proofoftech/flowsafe/approval-ui`. @@ -16,7 +16,7 @@ React is not required by flowsafe unless you import `@proofoftech/flowsafe/appro ## Install breakwater ```bash -npm install @mastra/core@^1.50.0 @proofoftech/breakwater +npm install @mastra/core@1.50.0 @proofoftech/breakwater ``` ### Add agent-boundary policy @@ -66,7 +66,7 @@ export const agent = createGuardedAgent({ }); ``` -`createGuardedAgent()` forces policy hold-back, disables background continuations, and rejects per-call processor, tool, model, callback, hook, structured-output, and execution-limit overrides. Application input processors can enforce only initial input. Application output processors must enforce both streamed and final results. +`createGuardedAgent()` forces policy hold-back, disables background continuations, and rejects per-call processor, tool, model, callback, hook, structured-output, and execution-limit overrides. Under the supported Mastra version, parsed structured output reaches messages and persistence before a wrapper gate could inspect it, so guarded structured output remains unavailable. Object-only policies are rejected at construction. Application input processors can enforce only initial input. Application output processors must enforce both streamed and final results. The narrow handle prevents accidental bypass through Mastra's larger `Agent` API. It is a trusted in-process API, not a sandbox against hostile code running with the same imports and credentials. diff --git a/docs/policy-engine-design.md b/docs/policy-engine-design.md index ce9c72b..0920f1e 100644 --- a/docs/policy-engine-design.md +++ b/docs/policy-engine-design.md @@ -26,7 +26,13 @@ interface PolicyEvaluator { Each decision is either `{ allowed: true }` or `{ allowed: false, reason }`. An evaluator exception is a policy-engine failure, is audited as an error, and fails the request closed. -Policies run in array order. The first denial aborts the phase. +Policies run in array order. The engine snapshots the list and each evaluator's +name, phase/channel selectors, hold-back hint, and evaluator reference at +construction. Class-based evaluators keep their original receiver, so private +fields and helper methods continue to work. Later replacement of the evaluator +method or mutation of caller-owned selector arrays does not change enforcement; +mutable state owned by the evaluator instance or its closures remains the +application's responsibility. The first denial aborts the phase. ## Phases and channels @@ -36,11 +42,11 @@ Input processing joins textual message parts and evaluates them under the `answe | --- | --- | | `answer` | Client-visible answer text | | `reasoning` | Reasoning stream deltas | -| `object` | JSON-stringified structured-output snapshots | +| `object` | Canonical JSON structured-output snapshots | A policy defaults to both phases and the `answer` channel. Set `phases` and `channels` when a policy applies more narrowly. -Mastra exposes structured objects to the processor on the streaming path. Under the supported core version, the non-streaming output result has no separate object field. JSON carried in answer text is still inspected by policies that include `answer`; an object-only policy has no final `generate()` coverage. +Under the supported core version, the engine sees the `object` channel only for object chunks that flow through the processor chain (model-native streaming). It requires those values to be JSON data, evaluates the canonical serialization, and forwards the same canonical clone. A `generate()` result's parsed object and core's `StructuredOutputProcessor` chunks never pass through the chain, and Mastra may expose the parsed value before `generate()` returns. `createGuardedAgent` therefore rejects structured output and rejects object-only policies at construction. JSON carried in answer text is still inspected by policies that include `answer`. A standalone engine with an object-only policy requires an audit sink and aborts at the result boundary unless a processor-visible object chunk provided coverage. ## Built-in content policies @@ -105,6 +111,8 @@ Properties: - A classifier that must see the full output should opt into `holdBackChars: Infinity`, accepting full buffering. - Hold-back changes delta boundaries. Consumers must treat text deltas as chunks, not semantic tokens. +Measured cost (2026-08-15, Node 22.22.0, `@mastra/core` 1.50.0; opt-in evidence tests in `packages/breakwater/src/policy-engine/policy-engine.test.ts`, run with `BREAKWATER_PERF=1`): a 4 MB stream in 2 KB deltas against string-pattern policies processes at roughly 1 MB/s including per-chunk harness overhead, with peak held text of 17 characters — the pattern-bound window, not the stream length. Any RegExp policy forces the unbounded window: a 1 MB stream keeps all of it pending and releases nothing until the channel ends. Prefer string patterns, overriding `holdBackChars` when the match bound is known, for large-stream leak prevention. + ## Tool policy `ToolPolicyEvaluator` receives the connector manifest, input, request context, and connector identity before execution. diff --git a/docs/proposals/breakwater-improvement-roadmap.md b/docs/proposals/breakwater-improvement-roadmap.md index 81d87b9..de89b79 100644 --- a/docs/proposals/breakwater-improvement-roadmap.md +++ b/docs/proposals/breakwater-improvement-roadmap.md @@ -36,7 +36,7 @@ testable, and appropriately placed. | **Shipped** | Optional connector invocation permissions | `PermissionManifest.requiredPermissions` is enforced against the trusted principal-permissions projection before dry-run and approval — authorization and approval answer different questions | | **Shipped** | Secure single-deployment policy preset | `singleTenantConnectorPolicies()` validates durable stores, audit, egress, permission wiring, background policy, and deployment-wide isolation posture | | **P1** | Add manifest-conformance tooling and stronger egress posture | Runtime safety depends on honest manifests and use of `runtime.fetch` | -| **P1** | Close known agent-output enforcement gaps | Structured output and override seams must not silently weaken mandatory policy | +| **Shipped** | Fail closed on known agent-output enforcement gaps | The guarded handle rejects structured output; processor-visible object chunks are canonicalized and missing required coverage aborts | | **P1** | Introduce stable decision codes and richer audit correlation | Reason strings are not a durable API for alerts, metrics, or incident response | | **P2** | Add policy bundles, versioning, and drift reporting | Useful once several agents/connectors share centrally managed policy | | **P2** | Add stricter rate-limit algorithms and operational maintenance | Fixed windows and durable stores need production-scale alternatives and cleanup | @@ -461,26 +461,23 @@ Infrastructure must still supply process/container/network controls for raw sockets, DNS/IP restrictions, and compromised dependencies. Do not market `runtime.fetch` as a complete sandbox. -### 13. Close Agent Output Coverage Gaps +### 13. Fail Closed on Agent Output Coverage Gaps (shipped) -Known processor/API limitations must remain loud: +#### Shipped implementation -- Under the pinned Mastra surface, non-streaming structured output does not - expose a parsed object to the output-result processor. -- Object-only policy therefore has weaker non-streaming coverage. -- The stream hold-back implementation depends on a Mastra reprocessing state - key guarded by compatibility tests. +Under the pinned core, parsed `generate()` objects and `StructuredOutputProcessor` chunks bypass output processors. Mastra also copies the parsed value into messages and may send it to memory, tracing, scorers, and logs before `generate()` returns. A post-generation wrapper gate is therefore too late. `createGuardedAgent()` rejects `structuredOutput` on both public methods before model execution and rejects object-only policies at construction. Guarded structured output remains planned until a verified pre-persistence seam exists. + +The versioned host protocol lets Flowsafe preserve that refusal even though Mastra's durable runner uses the branded raw agent and processor lists instead of the narrow handle. Flowsafe rejects structured output on its guarded durable stream, generate, and prepare entry points. -Improvements: +For processor-visible object chunks, `PolicyEngine` rejects non-JSON values, evaluates the canonical serialization, and forwards the same canonical clone. An object-only standalone policy requires an audit sink and aborts at the result boundary when no object chunk reached the processor. Hold-back cost is measured by opt-in evidence tests (`BREAKWATER_PERF=1`): a bounded string-pattern window holds only the pattern-bound trailing window regardless of stream size, while any RegExp policy buffers the whole stream; numbers are recorded in [Policy engine](../policy-engine-design.md). -- Gate the final parsed object in the guarded-agent wrapper when Mastra returns - it outside the processor result. -- Refuse configurations whose chosen invocation mode cannot expose a required - channel. -- Keep an upgrade tripwire around Mastra's processor and stream semantics. -- Add leak tests proving no forbidden prefix is emitted before a late detector - fires. -- Measure hold-back memory and latency under large streams. +Historical limitations that motivated this section: + +- Under the pinned Mastra surface, non-streaming structured output did not + expose a parsed object to the output-result processor. +- Object-only policy therefore had no enforceable guarded non-streaming path. +- The stream hold-back implementation depends on a Mastra reprocessing state + key guarded by compatibility tests (still true; unchanged). ### 14. Add Stable Decision Codes and Safe Error Surfaces @@ -724,7 +721,7 @@ The shipped host deliberately omits a public raw-resume route. It accepts struct 1. ~~Publish secure policy presets.~~ Shipped as `singleTenantConnectorPolicies()`. 2. Add connector conformance tests and global-fetch linting. 3. Generate the manifest/policy coverage report. -4. Close structured-output policy coverage. +4. Close structured-output policy coverage. Shipped fail-closed refusal; gated support still requires a pre-persistence seam (see §13). 5. Add stable decision codes. ### Phase E: Operations and scale diff --git a/docs/proposals/model-gateway-policy.md b/docs/proposals/model-gateway-policy.md index d248334..d02f39e 100644 --- a/docs/proposals/model-gateway-policy.md +++ b/docs/proposals/model-gateway-policy.md @@ -1,6 +1,6 @@ # Proposal: model gateway policy -> Status: partially implemented. `createGuardedAgent()` ships the non-overridable guarded-agent boundary, and `PolicyEngine` ships content inspection. Provider allowlisting, D1 spend budgets, model-endpoint egress policy, guarded structured-output validation, and an Anchorage scoring gate remain design only. Supported behavior is documented in [Breakwater architecture](../breakwater-architecture.md) and [Policy engine](../policy-engine-design.md). +> Status: partially implemented. `createGuardedAgent()` ships the non-overridable guarded-agent boundary and `PolicyEngine` ships content inspection. Guarded structured output, provider allowlisting, D1 spend budgets, model-endpoint egress policy, schema-shape validation, and an Anchorage scoring gate remain design only. Supported behavior is documented in [Breakwater architecture](../breakwater-architecture.md) and [Policy engine](../policy-engine-design.md). Mastra's `Agent` class handles model routing, provider normalization, and tool calling natively. This document covers the policy layer Anchorage adds on top of `Agent.generate()`. @@ -16,7 +16,7 @@ Mastra's `Agent` class handles model routing, provider normalization, and tool c | Policy | Status | Description | |---|---|---| -| Output schema validation | Planned | The guarded handle currently rejects structured-output call options | +| Output schema validation | Planned | Guarded `generate()` and `stream()` reject structured output because Mastra exposes parsed values before a wrapper gate can contain them; support requires a verified pre-persistence seam | | Content moderation | Partial | `PolicyEngine`, `piiSecrets()`, and `classifierPolicy()` ship; default prompt-injection and toxicity policies do not | | Quality scoring | Planned | Mastra construction-time scorers can pass through, but Anchorage adds no scoring policy or post-gate contract | @@ -24,7 +24,7 @@ Mastra's `Agent` class handles model routing, provider normalization, and tool c The shipped guarded path uses Breakwater's `RBACMiddleware` and `PolicyEngine` as mandatory Mastra processors. `createGuardedAgent()` fixes the model, processors, execution limits, and tool choice at construction, then rejects per-call overrides that could remove those controls. -No separate model-gateway export exists. The guarded handle does not implement the planned provider allowlist, budget lookup, model-endpoint check, structured-output validation, or scoring gate. +No separate model-gateway export exists. The guarded handle does not implement the planned structured-output gate, provider allowlist, budget lookup, model-endpoint check, or scoring gate. ## Planned budget enforcement diff --git a/docs/security-threat-model.md b/docs/security-threat-model.md index 14ad0c1..e1cce54 100644 --- a/docs/security-threat-model.md +++ b/docs/security-threat-model.md @@ -240,9 +240,9 @@ A host with only one human reviewer must consciously choose availability or sepa ## Guarded agent boundary -`createGuardedAgent()` fixes RBAC, policy, execution limits, tool choice, and application processor ordering at construction. Its public handle exposes only unstructured `generate()` and `stream()` calls with a mandatory `RequestContext`. +`createGuardedAgent()` fixes RBAC, policy, execution limits, tool choice, and application processor ordering at construction. Its public handle exposes only unstructured `generate()` and `stream()`, both with a mandatory `RequestContext`. It rejects structured output because Mastra exposes parsed values to messages and persistence before a wrapper could inspect them. -The handle rejects unknown call options even when a key is present with `undefined`. It disables background continuations, forces policy hold-back, and keeps the raw Mastra agent package-private. Flowsafe validates an unforgeable package-local brand before accepting a catalog module. +The handle rejects unknown call options even when a key is present with `undefined`, and the factory rejects object-only policies that no supported invocation can cover. Policy selectors and evaluator callables are captured at construction so later caller mutation cannot change channel coverage or replace the evaluator; mutable state internal to an evaluator remains trusted application state. The handle disables background continuations, forces policy hold-back, and keeps the raw Mastra agent package-private. Flowsafe validates an unforgeable package-local brand and a versioned host protocol before accepting a catalog module. Its lower-level durable methods snapshot data-property options before validation and delegation, reject accessors, and reject guarded structured output before Mastra can bypass the narrow handle. This boundary protects trusted application code from accidental or unsupported invocation paths. It is not a sandbox against hostile code running in the same process; such code can import Mastra, construct another agent, access host credentials, or bypass the supported HTTP topology. diff --git a/packages/breakwater/README.md b/packages/breakwater/README.md index 9d6cabc..2904fe1 100644 --- a/packages/breakwater/README.md +++ b/packages/breakwater/README.md @@ -38,10 +38,10 @@ schedule work, or run durable agent loops on Cloudflare. ## Install it breakwater is ESM-only, requires Node.js 22.3 or newer (engine range `>=22.3.0`), -and supports `@mastra/core` in its declared `^1.50.0` peer range. +and requires `@mastra/core` `1.50.0`. ```bash -npm install @proofoftech/breakwater @mastra/core +npm install @proofoftech/breakwater @mastra/core@1.50.0 ``` Connector authors who define Zod schemas should also declare Zod directly: @@ -113,7 +113,9 @@ const result = await agent.generate('Summarize the account.', { }); ``` -The handle exposes only unstructured `generate()` and `stream()`. Each call requires `requestContext` and may accept only `runId`, `memory`, and `abortSignal`. The factory fixes `maxSteps` and `toolChoice`, forces streaming policy hold-back, disables background continuations, and rejects processor or model overrides. +The handle exposes only unstructured `generate()` and `stream()`. Each call requires `requestContext` and may accept only `runId`, `memory`, and `abortSignal`. The factory fixes `maxSteps` and `toolChoice`, forces streaming policy hold-back, disables background continuations, and rejects processor, model, and structured-output overrides. + +Under the pinned Mastra version, parsed structured output bypasses the output-processor chain and reaches messages, persistence, and observability hooks before a wrapper could inspect it. Both guarded methods therefore reject `structuredOutput` before model execution. The factory also rejects object-only policies because no supported guarded invocation can expose their required channel. Structured-output support requires a future pre-persistence gate, not a post-generation wrapper. `allowedRoles` is an exact allowlist with no role hierarchy. Application input processors may implement only `processInput`. Application output processors must implement both `processOutputStream` and `processOutputResult`. @@ -140,11 +142,15 @@ The narrow handle prevents accidental use of raw Mastra execution methods. It is ## Choose agent policies `PolicyEngine` accepts any `PolicyEvaluator`. Policies can select the -`input` or `output` phase and the output channels they inspect: +`input` or `output` phase and the output channels they inspect. Construction +snapshots the policy list, selector arrays, names, hold-back hints, and +evaluator callable so later replacement cannot change enforcement. Class-based +evaluators retain their original receiver. Evaluator-owned instance or closure +state remains application-owned and is not deep-cloned: - `answer` is client-visible text and is also the only input channel. - `reasoning` is the model reasoning trace. -- `object` is the latest JSON-stringified structured-output snapshot. +- `object` is the latest canonical JSON structured-output snapshot. The included policies are: @@ -181,12 +187,13 @@ secret detectors provide bounded windows. A regular expression in text released at the end of an earlier segment cannot be withdrawn if a match completes in a later segment. -Under `@mastra/core` 1.50, the `object` channel is available on -`agent.stream()` but not as a distinct field in the final non-streaming -`generate()` result. JSON carried as answer text is still inspected by -answer-inclusive policies. An object-only policy therefore requires an audit -sink, and the engine emits a one-time coverage warning on the non-streaming -path. +Under `@mastra/core` 1.50.0, only object chunks that traverse the processor +chain reach a standalone engine. The engine validates those chunks as JSON, +evaluates the canonical snapshot, and forwards the same canonical clone. +Parsed `generate()` results and core's structured-output processor chunks +bypass that chain. If an object-only policy sees no processor-visible object +chunk, the engine requires an audit sink and aborts at the result boundary. +JSON carried as answer text is still inspected by answer-inclusive policies. ## Enforce tool permissions diff --git a/packages/breakwater/package.json b/packages/breakwater/package.json index bbe3db7..597836a 100644 --- a/packages/breakwater/package.json +++ b/packages/breakwater/package.json @@ -47,7 +47,7 @@ "prepublishOnly": "pnpm run build" }, "peerDependencies": { - "@mastra/core": "^1.50.0" + "@mastra/core": "1.50.0" }, "dependencies": { "zod": "^4.4.3" diff --git a/packages/breakwater/scripts/packed-consumer-test.mjs b/packages/breakwater/scripts/packed-consumer-test.mjs index 1e8e7f6..ce7edb0 100644 --- a/packages/breakwater/scripts/packed-consumer-test.mjs +++ b/packages/breakwater/scripts/packed-consumer-test.mjs @@ -70,7 +70,7 @@ try { ); assert.equal(manifest.dependencies?.zod, '^4.4.3'); assert.equal(manifest.devDependencies?.zod, undefined); - assert.equal(manifest.peerDependencies?.['@mastra/core'], '^1.50.0'); + assert.equal(manifest.peerDependencies?.['@mastra/core'], '1.50.0'); for (const documentation of [ 'README.md', 'CONNECTORS.md', @@ -133,6 +133,7 @@ try { CONNECTOR_GRANTS_CONTEXT_KEY, createGuardedAgent, createCodexConnector, + GUARDED_AGENT_HOST_PROTOCOL, inspectLegacyConnectorIdempotency, migrateLegacyConnectorIdempotency, singleTenantConnectorPolicies, @@ -140,11 +141,14 @@ try { type AgentCliErrorMetadata, type ConnectorApprovalGrant, type ConnectorExecutionIdentity, + type GuardedAgentCallOptions, type GuardedAgentHandle, + type GuardedAgentHostProtocol, type LegacyConnectorIdempotencyMigrationRequest, type LegacyConnectorIdempotencyMigrationResult, type SingleTenantConnectorPoliciesOptions, } from '@proofoftech/breakwater'; +import type { RequestContext } from '@mastra/core/request-context'; import { isGuardedAgentHandle } from '@proofoftech/breakwater/agent'; import { connectorManifest, @@ -231,6 +235,18 @@ const guarded: GuardedAgentHandle = createGuardedAgent({ maxSteps: 1, toolChoice: 'none', }); +const guardedProtocol: GuardedAgentHostProtocol = + guarded[GUARDED_AGENT_HOST_PROTOCOL]; +function checkGuardedCallOptions( + handle: GuardedAgentHandle, + options: GuardedAgentCallOptions, + requestContext: RequestContext, +): void { + void handle.generate('hello', options); + void handle.stream('hello', options); + // @ts-expect-error structured output is intentionally unavailable. + void handle.generate('hello', { requestContext, structuredOutput: {} }); +} const tool = createCodexConnector({ exec: async () => ({ stdout: 'ok', stderr: '', @@ -256,6 +272,8 @@ void migrationResult; void preset; void presetFromSubpath; void isGuardedAgentHandle(guarded); +void guardedProtocol; +void checkGuardedCallOptions; void isPermissionIdentifier(permission); void isPrincipalPermissions(projection); void metadata; diff --git a/packages/breakwater/src/agent/agent.test.ts b/packages/breakwater/src/agent/agent.test.ts index 951f654..cb84bb4 100644 --- a/packages/breakwater/src/agent/agent.test.ts +++ b/packages/breakwater/src/agent/agent.test.ts @@ -15,6 +15,8 @@ import { denyPatterns, type PolicyEvaluator } from '../policy-engine/index.js'; import { ACTOR_CONTEXT_KEY, type PrincipalKind } from '../rbac/index.js'; import { createGuardedAgent, + GUARDED_AGENT_HOST_PROTOCOL, + type GuardedAgentCallOptions, type GuardedAgentConfig, type GuardedAgentHandle, type GuardedInputProcessor, @@ -28,6 +30,22 @@ const usage = { totalTokens: 2, }; +class PrivateFieldPolicy implements PolicyEvaluator { + readonly name = 'private-field-policy'; + readonly phases = ['output'] as const; + readonly #blocked: string; + + constructor(blocked: string) { + this.#blocked = blocked; + } + + evaluate({ text }: Parameters[0]) { + return text.includes(this.#blocked) + ? { allowed: false as const, reason: 'matched private field' } + : { allowed: true as const }; + } +} + function testModel( text = 'model answer', onCall: () => void = () => {}, @@ -224,6 +242,48 @@ describe('createGuardedAgent direct execution', () => { ]); }); + it('keeps guarded policy enforcement independent of caller mutations', async () => { + const phases: Array<'input' | 'output'> = ['output']; + const channels: Array<'answer' | 'object'> = ['answer']; + const policy = denyPatterns(['blocked'], { phases, channels }); + const policies = [policy]; + const agent = guarded({ + model: testModel('blocked output'), + policies, + }); + + policies.length = 0; + phases[0] = 'input'; + channels[0] = 'object'; + policy.name = 'mutated'; + policy.evaluate = () => ({ allowed: true }); + + const result = await agent.generate('hello', { + requestContext: actorContext(), + }); + + expect(result.finishReason).toBe('other'); + expect(result.tripwire?.reason).toMatch(/deny-patterns/); + }); + + it('preserves a class policy receiver through guarded execution', async () => { + const policy = new PrivateFieldPolicy('blocked'); + const agent = guarded({ + model: testModel('blocked output'), + policies: [policy], + }); + policy.evaluate = () => ({ allowed: true }); + + const result = await agent.generate('hello', { + requestContext: actorContext(), + }); + + expect(result.finishReason).toBe('other'); + expect(result.tripwire?.reason).toMatch( + /private-field-policy: matched private field/, + ); + }); + it('withholds denied streamed output before it reaches the consumer', async () => { const agent = guarded({ model: testModel('blocked output'), @@ -938,14 +998,52 @@ describe('Mastra Agent execution-entry inventory', () => { }); }); +describe('createGuardedAgent structured output boundary', () => { + it.each([ + 'generate', + 'stream', + ] as const)('refuses structuredOutput on %s before model execution', async (method) => { + const modelCall = vi.fn(); + const agent = guarded({ model: testModel('unreachable', modelCall) }); + const options = { + requestContext: actorContext(), + structuredOutput: { schema: {} }, + }; + + await expect( + ( + agent[method] as ( + messages: string, + callOptions: unknown, + ) => Promise + )('hello', options), + ).rejects.toThrowError(/structuredOutput.*not allowed/); + expect(modelCall).not.toHaveBeenCalled(); + }); + + it('rejects object-only policies at construction', () => { + expect(() => + guarded({ + policies: [denyPatterns(['blocked'], { channels: ['object'] })], + }), + ).toThrowError(/object-only policy.*cannot be enforced/is); + }); +}); + function compileTimeSurface( handle: GuardedAgentHandle, requestContext: RequestContext, ): void { if (Date.now() < 0) { void handle.generate('hello', { requestContext }); + const options: GuardedAgentCallOptions = { requestContext }; + void handle.generate('hello', options); + void handle.stream('hello', options); // @ts-expect-error Structured output is intentionally unavailable. void handle.generate('hello', { requestContext, structuredOutput: {} }); + // @ts-expect-error Structured output is intentionally unavailable. + void handle.stream('hello', { requestContext, structuredOutput: {} }); + void handle[GUARDED_AGENT_HOST_PROTOCOL].supportsDurableStructuredOutput; // @ts-expect-error Raw resume is intentionally unavailable. void handle.resumeStream({}, { requestContext }); // @ts-expect-error Standalone durable resume is intentionally unavailable. diff --git a/packages/breakwater/src/agent/index.ts b/packages/breakwater/src/agent/index.ts index e7d8c45..044bb43 100644 --- a/packages/breakwater/src/agent/index.ts +++ b/packages/breakwater/src/agent/index.ts @@ -39,6 +39,17 @@ const GUARDED_CALL_OPTION_KEYS = new Set([ 'abortSignal', ]); +/** Well-known inter-package key for guarded host compatibility metadata. */ +export const GUARDED_AGENT_HOST_PROTOCOL = Symbol.for( + '@proofoftech/breakwater/guarded-agent-host/v1', +); + +/** Runtime metadata consumed by hosts that cannot call the narrow handle. */ +export interface GuardedAgentHostProtocol { + readonly version: 1; + readonly supportsDurableStructuredOutput: false; +} + const UNSAFE_CONSTRUCTION_KEYS = new Set([ 'agent', 'inputProcessors', @@ -220,6 +231,8 @@ export interface GuardedAgentHandle { readonly allowedPrincipalKinds: readonly PrincipalKind[]; /** Fixed maximum execution steps. */ readonly maxSteps: number; + /** Host compatibility metadata; not an alternate execution surface. */ + readonly [GUARDED_AGENT_HOST_PROTOCOL]: GuardedAgentHostProtocol; /** Generate one unstructured result through all mandatory gates. */ generate( @@ -409,7 +422,8 @@ function guardedCallOptions(options: unknown): GuardedAgentCallOptions { `GuardedAgent: call option '${String(key)}' is not allowed`, ); } - if (Object.getOwnPropertyDescriptor(options, key)?.get) { + const descriptor = Object.getOwnPropertyDescriptor(options, key); + if (descriptor?.get || descriptor?.set) { throw new TypeError( `GuardedAgent: call option '${key}' must be a data property`, ); @@ -424,7 +438,14 @@ function guardedCallOptions(options: unknown): GuardedAgentCallOptions { if (candidate.runId !== undefined && typeof candidate.runId !== 'string') { throw new TypeError('GuardedAgent: runId must be a string'); } - return candidate as GuardedAgentCallOptions; + return Object.freeze({ + requestContext: candidate.requestContext, + ...(candidate.runId !== undefined ? { runId: candidate.runId } : {}), + ...(candidate.memory !== undefined ? { memory: candidate.memory } : {}), + ...(candidate.abortSignal !== undefined + ? { abortSignal: candidate.abortSignal } + : {}), + }); } function directAuthorizationError(reason: string): never { @@ -439,6 +460,7 @@ class GuardedAgent< readonly allowedRoles: readonly Role[]; readonly allowedPrincipalKinds: readonly PrincipalKind[]; readonly maxSteps: number; + readonly [GUARDED_AGENT_HOST_PROTOCOL]: GuardedAgentHostProtocol; readonly #audit: AuditLogger; readonly #applicationInputProcessors: readonly GuardedInputProcessor[]; readonly #applicationOutputProcessors: readonly GuardedOutputProcessor[]; @@ -489,6 +511,11 @@ class GuardedAgent< holdBack: true, resource: `agent:${options.id}`, }); + if (policy.objectOnlyPolicyNames.length > 0) { + throw new TypeError( + `createGuardedAgent: object-only polic${policy.objectOnlyPolicyNames.length === 1 ? 'y' : 'ies'} [${policy.objectOnlyPolicyNames.join(', ')}] cannot be enforced because guarded structured output is unavailable under tested @mastra/core 1.50.0; include 'answer' in each affected policy's channels`, + ); + } const rbac = new RBACMiddleware({ allowedRoles, allowedPrincipalKinds, @@ -512,6 +539,10 @@ class GuardedAgent< this.allowedRoles = allowedRoles; this.allowedPrincipalKinds = allowedPrincipalKinds; this.maxSteps = maxSteps; + this[GUARDED_AGENT_HOST_PROTOCOL] = Object.freeze({ + version: 1, + supportsDurableStructuredOutput: false, + }); this.#audit = audit; this.#applicationInputProcessors = applicationInputProcessors; this.#applicationOutputProcessors = applicationOutputProcessors; diff --git a/packages/breakwater/src/index.ts b/packages/breakwater/src/index.ts index a4968e9..fbf15fc 100644 --- a/packages/breakwater/src/index.ts +++ b/packages/breakwater/src/index.ts @@ -16,12 +16,14 @@ export type { GuardedAgentCallOptions, GuardedAgentConfig, GuardedAgentHandle, + GuardedAgentHostProtocol, GuardedInputProcessor, GuardedOutputProcessor, GuardedToolChoice, } from './agent/index.js'; export { createGuardedAgent, + GUARDED_AGENT_HOST_PROTOCOL, isGuardedAgentHandle, } from './agent/index.js'; export type { diff --git a/packages/breakwater/src/policy-engine/content-inspection.test.ts b/packages/breakwater/src/policy-engine/content-inspection.test.ts index e44a0cf..796bd93 100644 --- a/packages/breakwater/src/policy-engine/content-inspection.test.ts +++ b/packages/breakwater/src/policy-engine/content-inspection.test.ts @@ -664,7 +664,7 @@ describe('piiSecrets', () => { // #when — the clean partial passes... await expect( engine.processOutputStream(makeStreamArgs([clean], state)), - ).resolves.toBe(clean); + ).resolves.toStrictEqual(clean); // #then — the REPLACEMENT snapshot (not a delta) is fully rescanned await expect( @@ -776,8 +776,8 @@ describe('classifierPolicy', () => { evaluateEveryChars: 10_000, }), ], - // The D1 guard: an object-only policy needs a sink to carry the - // one-time non-streaming-result coverage warning. + // The D1 guard: an object-only policy needs a sink to record a + // fail-closed result-phase coverage error when no object is observed. audit: new AuditLogger(), }); const state: Record = {}; diff --git a/packages/breakwater/src/policy-engine/index.ts b/packages/breakwater/src/policy-engine/index.ts index e4ebb49..c9c0f8b 100644 --- a/packages/breakwater/src/policy-engine/index.ts +++ b/packages/breakwater/src/policy-engine/index.ts @@ -6,8 +6,10 @@ // // Output is gated in two places so agent.stream() cannot leak forbidden text: // processOutputStream gates each streamed chunk against the output accumulated -// so far, and processOutputResult is the authoritative final gate (and the only -// one for non-streaming agent.generate()). +// so far, and processOutputResult is the authoritative final processor gate +// (and the only one for non-streaming agent.generate()). Structured objects +// that Mastra returns outside the processor chain are not covered; the guarded +// agent therefore rejects structured output. // // Output is gated per CHANNEL — 'answer' (client-visible text), 'reasoning' // (the model's reasoning trace), 'object' (structured-output snapshots) — @@ -28,6 +30,7 @@ import type { } from '@mastra/core/processors'; import type { RequestContext } from '@mastra/core/request-context'; import type { ChunkType } from '@mastra/core/stream'; +import { type JSONType, z } from 'zod'; import { type AuditLogger, agentAuditDetail } from '../audit/index.js'; import { type Actor, actorFromRequestContext } from '../rbac/index.js'; @@ -40,13 +43,16 @@ export type PolicyPhase = 'input' | 'output'; * Which output surface the gated text belongs to. 'answer' is the * client-visible answer text (and the channel input gating always runs * under); 'reasoning' is the model's reasoning trace; 'object' is structured - * output, gated as the JSON-stringified latest snapshot. + * output, gated as the canonical JSON latest snapshot. * - * Under `@mastra/core` 1.50, the `object` channel is available only during - * streaming because the final output result has no structured-object field. - * Direct structured output is also emitted as answer text, so policies that - * include `answer` still inspect it. An object-only policy requires an audit - * sink so the engine can report the non-streaming coverage gap. + * Under `@mastra/core` 1.50, the engine sees the `object` channel only for + * object chunks that flow through the processor chain (model-native + * streaming). A `generate()` result's parsed object and core's + * `StructuredOutputProcessor` chunks never pass through the chain. + * `createGuardedAgent` rejects structured output because a wrapper gate would + * run only after Mastra had exposed the parsed value. An object-only policy + * still requires an audit sink so a standalone engine records a fail-closed + * result-phase coverage error. */ export type OutputChannel = 'answer' | 'reasoning' | 'object'; @@ -135,6 +141,42 @@ const CHANNELS_STATE_KEY = 'breakwater.channels'; // collide across instances (two denyPatterns both named 'deny-patterns'); // indexes cannot. const POLICY_STATE_KEY = 'breakwater.policyState'; +const OBJECT_CHANNEL_EVALUATED_STATE_KEY = 'breakwater.objectChannelEvaluated'; +const STREAM_EVALUATED_POLICIES_STATE_KEY = + 'breakwater.streamEvaluatedPolicies'; +const JSON_VALUE_SCHEMA = z.json(); + +interface CanonicalJsonValue { + value: JSONType; + snapshot: string; +} + +interface StreamEvaluatedPolicies { + names: string[]; + channels: OutputChannel[]; +} + +function streamEvaluatedPoliciesOf( + state: Record, +): StreamEvaluatedPolicies { + let evaluated = state[STREAM_EVALUATED_POLICIES_STATE_KEY] as + | StreamEvaluatedPolicies + | undefined; + if (!evaluated) { + evaluated = { names: [], channels: [] }; + state[STREAM_EVALUATED_POLICIES_STATE_KEY] = evaluated; + } + return evaluated; +} + +function canonicalJsonValue(value: unknown): CanonicalJsonValue { + const parsed = JSON_VALUE_SCHEMA.parse(value); + const snapshot = JSON.stringify(parsed); + return { + value: JSON.parse(snapshot) as JSONType, + snapshot, + }; +} interface ChannelTexts { answer: string; @@ -238,9 +280,32 @@ function holdBackWindowFor( return window; } +function snapshotPolicies( + policies: readonly PolicyEvaluator[], +): readonly PolicyEvaluator[] { + return Object.freeze( + policies.map((policy) => { + const name = policy.name; + const phases = policy.phases; + const channels = policy.channels; + const holdBackChars = policy.holdBackChars; + const evaluate = policy.evaluate.bind(policy); + return Object.freeze({ + name, + ...(phases !== undefined ? { phases: Object.freeze([...phases]) } : {}), + ...(channels !== undefined + ? { channels: Object.freeze([...channels]) } + : {}), + ...(holdBackChars !== undefined ? { holdBackChars } : {}), + evaluate, + }); + }), + ); +} + /** Configuration for `PolicyEngine`. */ export interface PolicyEngineOptions { - /** Policies evaluated in array order. */ + /** Policies snapshotted at construction and evaluated in array order. */ policies: readonly PolicyEvaluator[]; /** Optional audit logger for policy decisions and evaluator failures. */ audit?: AuditLogger; @@ -274,9 +339,9 @@ export interface PolicyEngineOptions { * * Under `@mastra/core` 1.50, a final output result has no structured-object * field. The constructor therefore requires an audit sink when a policy - * selects `object` without `answer`. The first final-result call then emits - * one coverage warning for that engine instance. Policies that include - * `answer` inspect the JSON text emitted by direct structured output. + * selects `object` without `answer`. A final-result call fails closed unless + * this engine actually evaluated an object chunk. Policies that include + * `answer` inspect JSON carried as answer text. * * The constructor also rejects an explicit input policy whose channels * exclude `answer`, because input evaluation has no other channel. @@ -290,10 +355,10 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { readonly #holdBack: boolean; readonly #holdBackWindow: Record; readonly #objectOnlyPolicyNames: readonly string[]; - #objectChannelFenceWarned = false; constructor(options: PolicyEngineOptions) { - for (const policy of options.policies) { + const policies = snapshotPolicies(options.policies); + for (const policy of policies) { if ( policy.phases?.includes('input') && policy.channels !== undefined && @@ -304,32 +369,31 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { ); } } - this.#policies = options.policies; + this.#policies = policies; this.#audit = options.audit; this.#resource = options.resource ?? this.id; this.#holdBack = options.holdBack ?? false; this.#holdBackWindow = { - answer: holdBackWindowFor(options.policies, 'answer'), - reasoning: holdBackWindowFor(options.policies, 'reasoning'), + answer: holdBackWindowFor(policies, 'answer'), + reasoning: holdBackWindowFor(policies, 'reasoning'), }; - this.#objectOnlyPolicyNames = options.policies - .filter( - (policy) => - policy.channels?.includes('object') && - !policy.channels.includes('answer'), - ) - .map((policy) => policy.name); - // D1 fence (construction time): an object-only policy has zero - // result-phase coverage under @mastra/core 1.50.0, and the runtime - // warning it would otherwise get rides the OPTIONAL audit sink — with no - // sink the gap is entirely silent. Reject the combination rather than - // ship an unenforceable policy that can never surface. Reuses the - // object-only set computed above. + this.#objectOnlyPolicyNames = Object.freeze( + policies + .filter( + (policy) => + policy.channels?.includes('object') && + !policy.channels.includes('answer'), + ) + .map((policy) => policy.name), + ); + // D1 fence (construction time): an object-only policy needs an audit sink + // to record a fail-closed coverage error when processors expose no object. + // Reuses the object-only set computed above. if (this.#objectOnlyPolicyNames.length > 0 && options.audit === undefined) { const names = this.#objectOnlyPolicyNames.join(', '); const plural = this.#objectOnlyPolicyNames.length === 1 ? 'y' : 'ies'; throw new TypeError( - `PolicyEngine: polic${plural} [${names}] scoped to the 'object' channel without 'answer' cannot be enforced on non-streaming generate() results under @mastra/core 1.50.0 (OutputResult carries no structured-object field) and would silently no-op without an audit sink to carry the one-time warning — provide options.audit, or include 'answer' in channels.`, + `PolicyEngine: polic${plural} [${names}] scoped to the 'object' channel without 'answer' require an audit sink for fail-closed coverage errors when no object reaches the processor — provide options.audit, or include 'answer' in channels.`, ); } } @@ -354,11 +418,12 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { async processOutputResult( args: ProcessOutputResultArgs, ): Promise { - this.#warnObjectChannelGapOnce(args.requestContext); + this.#assertObjectChannelCoverage(args); // result.text is the authoritative generation output (non-optional in // core); messages also carry earlier conversation turns the output - // policies should not re-gate. This is the final gate for both - // agent.generate() and, after the stream drains, agent.stream(). + // policies should not re-gate. This is the final processor gate for + // answer and reasoning on both agent.generate() and, after the stream + // drains, agent.stream(). const actor = actorFromRequestContext(args.requestContext) ?? null; const evaluated = await this.#evaluate( { @@ -371,40 +436,55 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { actor, args.abort, ); + const evaluatedChannels: OutputChannel[] = + evaluated.length > 0 ? ['answer'] : []; // Reasoning is gated from the per-step aggregates. OutputResult carries - // no structured-object field, so the object channel is gated in-stream - // only ('object'/'object-result' chunks); structured output transported - // as JSON answer text is covered by the answer pass above. + // no structured-object field, so this processor gates the object channel + // only when 'object'/'object-result' chunks reach processOutputStream. + // JSON carried as answer text is covered by the answer pass above. const reasoningText = args.result.steps .map((step) => step.reasoningText) .filter((text): text is string => typeof text === 'string' && text !== '') .join('\n'); if (reasoningText !== '') { - evaluated.push( - ...(await this.#evaluate( - { - phase: 'output', - channel: 'reasoning', - messages: args.messages, - text: reasoningText, - requestContext: args.requestContext, - }, - actor, - args.abort, - )), + const reasoningEvaluated = await this.#evaluate( + { + phase: 'output', + channel: 'reasoning', + messages: args.messages, + text: reasoningText, + requestContext: args.requestContext, + }, + actor, + args.abort, ); + evaluated.push(...reasoningEvaluated); + if (reasoningEvaluated.length > 0) { + evaluatedChannels.push('reasoning'); + } + } + const streamedEvaluated = args.state[STREAM_EVALUATED_POLICIES_STATE_KEY] as + | StreamEvaluatedPolicies + | undefined; + if (streamedEvaluated) { + evaluated.push(...streamedEvaluated.names); + evaluatedChannels.push(...streamedEvaluated.channels); } - // ONE terminal allowed record per result — channel passes aggregated, - // deduplicated — not one record per channel. this.#recordAllowed( 'output', actor, [...new Set(evaluated)], args.requestContext, + [...new Set(evaluatedChannels)], ); return args.messages; } + /** Names of policies scoped to `object` without `answer` (never result-covered by the engine alone). */ + get objectOnlyPolicyNames(): readonly string[] { + return this.#objectOnlyPolicyNames; + } + // agent.stream() emits chunks to the client before processOutputResult // runs, so the result gate alone lets forbidden output through mid-stream. // Each gated chunk feeds its channel's accumulated text — text-delta → @@ -428,6 +508,7 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { const texts = channelTextsOf(args.state); let channel: OutputChannel; let delta: DeltaChunk | undefined; + let forwardedPart = part; if (part.type === 'text-delta') { // The typeof guard stops a malformed chunk (payload.text not a string) // from coercing e.g. "undefined" into the tracked text. @@ -445,9 +526,28 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { // (ChunkType, stream/types.d.ts): replace, never concatenate. The // stringify lib type lies — it returns undefined for undefined input // (a malformed chunk), which must not corrupt the tracked text. - const snapshot: string | undefined = JSON.stringify(part.object); - texts.object = snapshot ?? ''; + let canonical: CanonicalJsonValue; + try { + canonical = canonicalJsonValue(part.object); + } catch { + this.#audit?.record({ + actor: actorFromRequestContext(args.requestContext) ?? null, + action: 'agent.output.policy', + resource: this.#resource, + decision: 'error', + reason: 'structured object is not JSON data', + detail: agentAuditDetail(args.requestContext, { + channel: 'object', + }), + }); + args.abort('structured object is not JSON data'); + } + texts.object = canonical.snapshot; channel = 'object'; + forwardedPart = { + ...part, + object: canonical.value, + } as unknown as ChunkType; } else { return this.#holdBack ? this.#forwardUngated(args) : part; } @@ -457,7 +557,7 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { // abortOnError=true: Mastra's stream driver emits the chunk on a raw throw // and only suppresses it on a TripWire (abort), so an evaluator crash here // must abort, not rethrow. - await this.#evaluate( + const evaluated = await this.#evaluate( { phase: 'output', channel, @@ -470,14 +570,28 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { true, args.state, ); - if (!this.#holdBack) return part; + if (evaluated.length > 0) { + const streamedEvaluated = streamEvaluatedPoliciesOf(args.state); + for (const name of evaluated) { + if (!streamedEvaluated.names.includes(name)) { + streamedEvaluated.names.push(name); + } + } + if (!streamedEvaluated.channels.includes(channel)) { + streamedEvaluated.channels.push(channel); + } + } + if (channel === 'object') { + args.state[OBJECT_CHANNEL_EVALUATED_STATE_KEY] = true; + } + if (!this.#holdBack) return forwardedPart; if (delta && (channel === 'answer' || channel === 'reasoning')) { return this.#releaseHeld(args, channel, delta); } // Intermediate 'object' snapshots are suppressed under hold-back // (evaluated, never emitted); the final object-result is emitted once it // passes. Trade-off: consumers get only the final object. - return part.type === 'object-result' ? part : null; + return part.type === 'object-result' ? forwardedPart : null; } // Hold-back release for a just-evaluated delta. The full accumulated @@ -639,33 +753,28 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { return evaluated; } - // D1 fence (see the class doc + OutputChannel's coverage caveat): a policy - // scoped to 'object' without 'answer' has zero coverage at the result - // phase under @mastra/core 1.50.0 — warn once per engine instance instead - // of silently doing nothing on every processOutputResult call. The - // constructor now REJECTS this configuration when no audit sink is present - // (D1), so #audit is guaranteed to exist whenever there is an object-only - // policy to warn about; the optional-chain remains only as defense in depth. - #warnObjectChannelGapOnce(requestContext?: RequestContext): void { + #assertObjectChannelCoverage(args: ProcessOutputResultArgs): void { if ( - this.#objectChannelFenceWarned || - this.#objectOnlyPolicyNames.length === 0 + this.#objectOnlyPolicyNames.length === 0 || + args.state[OBJECT_CHANNEL_EVALUATED_STATE_KEY] === true ) { return; } - this.#objectChannelFenceWarned = true; const names = this.#objectOnlyPolicyNames.join(', '); const plural = this.#objectOnlyPolicyNames.length === 1 ? 'y' : 'ies'; this.#audit?.record({ - actor: null, + actor: actorFromRequestContext(args.requestContext) ?? null, action: 'agent.output.policy', resource: this.#resource, decision: 'error', - reason: `polic${plural} [${names}] scoped to the 'object' channel without 'answer' cannot be enforced on non-streaming generate() results under @mastra/core 1.50.0 (OutputResult carries no structured-object field) — gate via agent.stream() or include 'answer' in channels`, - detail: agentAuditDetail(requestContext, { + reason: 'required object output channel was not observable', + detail: agentAuditDetail(args.requestContext, { policies: [...this.#objectOnlyPolicyNames], }), }); + args.abort( + `PolicyEngine: polic${plural} [${names}] require the 'object' output channel, but this invocation exposed no object to the processor`, + ); } #recordAllowed( @@ -673,13 +782,17 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { actor: Actor | null, evaluated: string[], requestContext?: RequestContext, + channels?: readonly OutputChannel[], ): void { this.#audit?.record({ actor, action: `agent.${phase}.policy`, resource: this.#resource, decision: 'allowed', - detail: agentAuditDetail(requestContext, { evaluated }), + detail: agentAuditDetail(requestContext, { + evaluated, + ...(channels !== undefined ? { channels } : {}), + }), }); } } @@ -690,7 +803,8 @@ export class PolicyEngine implements Processor<'breakwater-policy-engine'> { * channels by default (answer, reasoning, object) — leak prevention is its * purpose, and a secret is no less leaked through a reasoning trace; narrow * with `options.channels` when a channel must stay ungated. The 'object' - * channel is enforced in-stream only (see OutputChannel's coverage caveat). + * channel is enforced only for object chunks that reach this processor; see + * {@link OutputChannel} for the guarded structured-output limitation. * * Substring matching is plain toLowerCase — no Unicode folding or * normalization — so alternate spellings evade it (e.g. 'strasse' does not diff --git a/packages/breakwater/src/policy-engine/policy-engine.test.ts b/packages/breakwater/src/policy-engine/policy-engine.test.ts index 040018b..996061d 100644 --- a/packages/breakwater/src/policy-engine/policy-engine.test.ts +++ b/packages/breakwater/src/policy-engine/policy-engine.test.ts @@ -25,6 +25,22 @@ import { class Tripwire extends Error {} +class PrivateFieldPolicy implements PolicyEvaluator { + readonly name = 'private-field-policy'; + readonly phases = ['output'] as const; + readonly #blocked: string; + + constructor(blocked: string) { + this.#blocked = blocked; + } + + evaluate({ text }: Parameters[0]) { + return text.includes(this.#blocked) + ? { allowed: false as const, reason: 'matched private field' } + : { allowed: true as const }; + } +} + let messageSeq = 0; function makeMessage( @@ -396,8 +412,8 @@ describe('PolicyEngine constructor validation (K2)', () => { }); it('rejects an object-only policy constructed without an audit sink (D1)', () => { - // #given — channels include 'object' but not 'answer': zero result-phase - // coverage AND no sink to carry the one-time warning, so the gap is silent + // #given — channels include 'object' but not 'answer', and no sink can + // record the fail-closed coverage error if an invocation exposes no object const policy: PolicyEvaluator = { name: 'object-only', channels: ['object'], @@ -411,7 +427,7 @@ describe('PolicyEngine constructor validation (K2)', () => { }); it('allows an object-only policy when an audit sink is provided (D1)', () => { - // #given — the same policy, now with a sink to carry the one-time warning + // #given — the same policy, now with a sink for coverage errors const audit = new AuditLogger(); const policy: PolicyEvaluator = { name: 'object-only', @@ -422,10 +438,48 @@ describe('PolicyEngine constructor validation (K2)', () => { // #when / #then expect(() => new PolicyEngine({ policies: [policy], audit })).not.toThrow(); }); + + it('snapshots the policy list and decision-driving policy fields', async () => { + const phases: Array<'input' | 'output'> = ['output']; + const channels: Array<'answer' | 'object'> = ['answer']; + const policy = denyPatterns(['blocked'], { phases, channels }); + const policies = [policy]; + const engine = new PolicyEngine({ policies }); + + policies.length = 0; + phases[0] = 'input'; + channels[0] = 'object'; + policy.name = 'mutated'; + policy.evaluate = () => ({ allowed: true }); + + await expect( + engine.processOutputResult(makeOutputArgs('blocked')), + ).rejects.toThrowError(/deny-patterns: matched blocked pattern/); + }); + + it('snapshots hold-back hints before caller mutation', async () => { + const policy = denyPatterns(['blocked'], { phases: ['output'] }); + const engine = new PolicyEngine({ policies: [policy], holdBack: true }); + policy.holdBackChars = 0; + + await expect( + engine.processOutputStream(makeStreamArgs([textDelta('clean')], {})), + ).resolves.toBeNull(); + }); + + it('preserves a class evaluator receiver while capturing its method', async () => { + const policy = new PrivateFieldPolicy('blocked'); + const engine = new PolicyEngine({ policies: [policy] }); + policy.evaluate = () => ({ allowed: true }); + + await expect( + engine.processOutputResult(makeOutputArgs('blocked')), + ).rejects.toThrowError(/private-field-policy: matched private field/); + }); }); describe('PolicyEngine object-channel result-phase fence (D1)', () => { - it('warns once per engine instance when a policy is scoped to object without answer', async () => { + it('fails closed when a policy requires an object the invocation never exposes', async () => { // #given — zero result-phase coverage: OutputResult has no object field, // and this policy never sees the answer channel either const audit = new AuditLogger(); @@ -434,22 +488,15 @@ describe('PolicyEngine object-channel result-phase fence (D1)', () => { audit, }); - // #when — two result-phase calls - await engine.processOutputResult(makeOutputArgs('clean')); - await engine.processOutputResult(makeOutputArgs('clean')); + // #when / #then — the result gate aborts instead of logging and allowing + await expect( + engine.processOutputResult(makeOutputArgs('clean')), + ).rejects.toThrowError(/require the 'object' output channel/); - // #then — exactly one warning, not one per call. #recordAllowed also - // writes 'agent.output.policy' events (decision: 'allowed') on every - // call, so the fence warning is distinguished by decision: 'error'. - const warnings = audit - .events() - .filter( - (event) => - event.action === 'agent.output.policy' && event.decision === 'error', - ); - expect(warnings).toHaveLength(1); - expect(warnings[0]).toMatchObject({ + expect(audit.events()).toHaveLength(1); + expect(audit.events()[0]).toMatchObject({ decision: 'error', + reason: 'required object output channel was not observable', detail: { policies: ['deny-patterns'] }, }); }); @@ -500,6 +547,41 @@ describe('PolicyEngine object-channel result-phase fence (D1)', () => { ), ).toBe(false); }); + + it('accepts a result after the processor actually evaluated an object chunk', async () => { + const audit = new AuditLogger(); + const engine = new PolicyEngine({ + policies: [denyPatterns(['x'], { channels: ['object'] })], + audit, + }); + const state: Record = {}; + const part = objectResult({ answer: 'clean' }); + + await engine.processOutputStream(makeStreamArgs([part], state)); + const resultArgs = makeOutputArgs('clean'); + resultArgs.state = state; + + await expect(engine.processOutputResult(resultArgs)).resolves.toEqual( + resultArgs.messages, + ); + expect(audit.events().some((event) => event.decision === 'error')).toBe( + false, + ); + }); + + it('exposes a frozen list of object-only policies', () => { + const engine = new PolicyEngine({ + policies: [ + denyPatterns(['x'], { channels: ['object'] }), + denyPatterns(['y'], { channels: ['answer', 'object'] }), + maxTextLength(10), + ], + audit: new AuditLogger(), + }); + + expect(engine.objectOnlyPolicyNames).toEqual(['deny-patterns']); + expect(Object.isFrozen(engine.objectOnlyPolicyNames)).toBe(true); + }); }); describe('PolicyEngine.processOutputStream', () => { @@ -604,6 +686,23 @@ describe('PolicyEngine.processOutputStream', () => { expect(audit.events()).toHaveLength(0); }); + it('bounds terminal audit metadata by policies and channels, not chunk count', async () => { + const engine = new PolicyEngine({ + policies: [denyPatterns(['blocked'], { phases: ['output'] })], + }); + const state: Record = {}; + + for (let index = 0; index < 100; index += 1) { + const part = textDelta('clean'); + await engine.processOutputStream(makeStreamArgs([part], state)); + } + + expect(state['breakwater.streamEvaluatedPolicies']).toEqual({ + names: ['deny-patterns'], + channels: ['answer'], + }); + }); + it('fails closed on an evaluator crash mid-stream: aborts, not rethrows', async () => { // #given — Mastra's stream driver emits the chunk on a raw throw and only // suppresses it on an abort (TripWire), so a crash must surface as an abort @@ -731,6 +830,45 @@ describe('PolicyEngine output channels — streaming', () => { ).rejects.toThrowError(/deny-patterns: matched blocked pattern/); }); + it('forwards the same canonical object snapshot that policy inspected', async () => { + const engine = new PolicyEngine({ policies: [] }); + const transformed = Object.defineProperty({ answer: 'ok' }, 'secret', { + value: 'hidden', + enumerable: false, + }); + const part = objectResult(transformed); + + const result = await engine.processOutputStream(makeStreamArgs([part])); + + expect(result).toMatchObject({ + type: 'object-result', + object: { answer: 'ok' }, + }); + expect((result as { object?: unknown }).object).not.toBe(transformed); + expect((result as { object?: unknown }).object).not.toHaveProperty( + 'secret', + ); + }); + + it('aborts before forwarding a structured chunk that is not JSON data', async () => { + const audit = new AuditLogger(); + const engine = new PolicyEngine({ policies: [], audit }); + const part = objectResult({ + answer: 'ok', + toJSON: () => ({ answer: 'ok' }), + }); + + await expect( + engine.processOutputStream(makeStreamArgs([part])), + ).rejects.toThrowError(/structured object is not JSON data/); + expect(audit.events()).toHaveLength(1); + expect(audit.events()[0]).toMatchObject({ + decision: 'error', + reason: 'structured object is not JSON data', + detail: { channel: 'object' }, + }); + }); + it('evaluates object snapshots as replacements, not concatenations', async () => { // #given — a cap that the CONCATENATION of the two snapshots would // exceed but the latest snapshot alone does not (partials are growing @@ -747,10 +885,10 @@ describe('PolicyEngine output channels — streaming', () => { // #then — both pass because the second REPLACES the first await expect( engine.processOutputStream(makeStreamArgs([partial], state)), - ).resolves.toBe(partial); + ).resolves.toStrictEqual(partial); await expect( engine.processOutputStream(makeStreamArgs([partial, final], state)), - ).resolves.toBe(final); + ).resolves.toStrictEqual(final); }); it('keeps channel caps independent: long reasoning does not trip an answer cap', async () => { @@ -1032,7 +1170,7 @@ describe('PolicyEngine hold-back buffering', () => { ).resolves.toBeNull(); await expect( engine.processOutputStream(makeStreamArgs([partial, final], state)), - ).resolves.toBe(final); + ).resolves.toStrictEqual(final); }); it('drops pending text when the stream errors instead of emitting after the failure', async () => { @@ -1476,3 +1614,105 @@ describe('extractMessageText', () => { ); }); }); + +// Opt-in perf evidence for hold-back cost (roadmap §13: "measure hold-back +// memory and latency under large streams"). Skipped unless +// BREAKWATER_PERF=1 so CI never carries timing variance; run with: +// BREAKWATER_PERF=1 pnpm --filter @proofoftech/breakwater exec vitest run src/policy-engine/policy-engine.test.ts -t 'hold-back cost' +describe('PolicyEngine hold-back cost (opt-in perf evidence)', () => { + const textOf = (chunk: ChunkType | null | undefined): string => { + const text = (chunk as { payload?: { text?: unknown } } | null | undefined) + ?.payload?.text; + return typeof text === 'string' ? text : ''; + }; + + it.skipIf(!process.env.BREAKWATER_PERF)( + 'measures latency and peak buffering on a 4 MB stream', + async () => { + // #given — a bounded string-pattern window (18-char pattern → 17-char + // hold window) and 4 MB of clean text in 2 KB deltas + const TOTAL_CHARS = 4 * 1024 * 1024; + const CHUNK = 'x'.repeat(2048); + const engine = new PolicyEngine({ + policies: [ + denyPatterns(['needle-not-present'], { phases: ['output'] }), + ], + holdBack: true, + }); + const state: Record = {}; + let emitted = 0; + let peakPending = 0; + const pendingOf = () => { + const hold = state['breakwater.holdBack'] as + | { answer?: { pending: string } } + | undefined; + return hold?.answer?.pending.length ?? 0; + }; + + // #when + const started = performance.now(); + for (let sent = 0; sent < TOTAL_CHARS; sent += CHUNK.length) { + const part = await engine.processOutputStream( + makeStreamArgs([textDelta(CHUNK)], state), + ); + emitted += textOf(part).length; + peakPending = Math.max(peakPending, pendingOf()); + } + const elapsedMs = performance.now() - started; + + // #then — bounded window held (pattern-bound string window + one + // in-flight chunk), everything eventually released in aggregate, and + // throughput stays far above interactive stream rates. Bounds are + // generous regression fences, not benchmarks; the reported numbers are + // recorded in docs/policy-engine-design.md. + expect(peakPending).toBeLessThanOrEqual( + 'needle-not-present'.length - 1 + CHUNK.length, + ); + const hold = state['breakwater.holdBack'] as { + answer?: { pending: string }; + }; + expect(emitted + (hold.answer?.pending.length ?? 0)).toBe(TOTAL_CHARS); + expect(elapsedMs).toBeLessThan(10_000); + console.log( + `hold-back perf: ${TOTAL_CHARS / 1024 / 1024} MB in ${Math.round(elapsedMs)} ms — ${Math.round(TOTAL_CHARS / 1024 / (elapsedMs / 1000))} KB/s, peak pending ${peakPending} chars`, + ); + }, + ); + + it.skipIf(!process.env.BREAKWATER_PERF)( + 'measures the RegExp (Infinity-window) buffer-all trade-off on the same stream', + async () => { + // #given — any RegExp forces an unbounded hold window: the entire + // stream stays pending until finish. This MEASURES that documented + // trade-off rather than asserting a bound that cannot hold. + const TOTAL_CHARS = 1024 * 1024; + const CHUNK = 'y'.repeat(2048); + const engine = new PolicyEngine({ + policies: [denyPatterns([/z-\d+/], { phases: ['output'] })], + holdBack: true, + }); + const state: Record = {}; + let emitted = 0; + let peakPending = 0; + const pendingOf = () => { + const hold = state['breakwater.holdBack'] as + | { answer?: { pending: string } } + | undefined; + return hold?.answer?.pending.length ?? 0; + }; + + // #when + for (let sent = 0; sent < TOTAL_CHARS; sent += CHUNK.length) { + const part = await engine.processOutputStream( + makeStreamArgs([textDelta(CHUNK)], state), + ); + emitted += textOf(part).length; + peakPending = Math.max(peakPending, pendingOf()); + } + + // #then — nothing released mid-stream, everything buffered + expect(emitted).toBe(0); + expect(peakPending).toBe(TOTAL_CHARS); + }, + ); +}); diff --git a/packages/flowsafe/README.md b/packages/flowsafe/README.md index bb7f967..fb90242 100644 --- a/packages/flowsafe/README.md +++ b/packages/flowsafe/README.md @@ -9,7 +9,7 @@ Flowsafe runs Mastra workflows and agents through Cloudflare Durable Objects, st ## Install ```bash -npm install @mastra/core@^1.50.0 @proofoftech/flowsafe +npm install @mastra/core@1.50.0 @proofoftech/flowsafe ``` Install `@proofoftech/breakwater` when resumed steps call approval-protected connectors: @@ -25,9 +25,9 @@ Compatibility: - Node.js 22 or later (engine range `>=22`) - ESM only - TypeScript `moduleResolution: "NodeNext"`, `"Node16"`, or `"Bundler"` -- `@mastra/core` in the declared `^1.50.0` peer range +- `@mastra/core` `1.50.0` - `react` and `react-dom` `>=18 <20` (React 18 or 19) for the optional approval UI -- `@proofoftech/breakwater` `>=0.9.0 <1.0.0` when used +- `@proofoftech/breakwater` `>=0.12.0 <1.0.0` when used - host-provided Wrangler `>=4.118 <5` for the optional `flowsafe-provision` CLI ## Choose an export @@ -345,11 +345,11 @@ A configured resolver runs on every authorized entry, and its resolution is proj Permission authorization audit detail records `requiredPermissions` and `permissionPolicyVersion`; it does not record effective permissions or identity-provider groups. -`createThreadAgentHost()` validates Breakwater's guarded-handle brand before it registers the agent with Mastra. It persists the thread/agent binding and original run principal, so eviction recovery and approval resume cannot switch agents or actors. +`createThreadAgentHost()` validates Breakwater's guarded-handle brand and versioned host protocol before it registers the agent with Mastra. It persists the thread/agent binding and original run principal, so eviction recovery and approval resume cannot switch agents or actors. Agent resume is approval-only. `createAgentApprovalResumer()` rejects legacy agent targets without the original principal, then rechecks that principal against the current catalog. A human must still satisfy the selected agent's `allowedRoles`. An automated principal's kind must still appear in `allowedAutomation`; `approval.resume` is implied for a declared kind. The thread host then enforces any `requiredPermissions` through the current resolver policy. The resumer delegates non-agent workflow records to the existing resume function. -`createFlowsafeDurableAgent()` remains the lower-level compatibility API. It routes Mastra's durable-agent workflow through `RunnerRuntime`, but it does not guard an arbitrary raw agent. Use `agent-host` when an HTTP surface must enforce catalog and Breakwater invariants. +`createFlowsafeDurableAgent()` remains the lower-level compatibility API. It routes Mastra's durable-agent workflow through `RunnerRuntime`, but it does not guard an arbitrary raw agent. Each durable entry snapshots data-property call options before validating or delegating and rejects accessor-backed options. When passed a Breakwater guarded agent, it rejects structured output on stream, generate, and prepare because Mastra's durable runner bypasses the narrow handle. Use `agent-host` when an HTTP surface must enforce catalog and Breakwater invariants. Agent event replay lasts only as long as the configured Mastra cache. The default in-memory cache does not survive process restart. A 409 stream response means the client must read the authoritative status route. diff --git a/packages/flowsafe/package.json b/packages/flowsafe/package.json index 67cd34c..98db169 100644 --- a/packages/flowsafe/package.json +++ b/packages/flowsafe/package.json @@ -82,8 +82,8 @@ "prepublishOnly": "pnpm run build" }, "peerDependencies": { - "@mastra/core": "^1.50.0", - "@proofoftech/breakwater": ">=0.9.0 <1.0.0", + "@mastra/core": "1.50.0", + "@proofoftech/breakwater": ">=0.12.0 <1.0.0", "react": ">=18 <20", "react-dom": ">=18 <20" }, diff --git a/packages/flowsafe/scripts/agent-host-pack-test.mjs b/packages/flowsafe/scripts/agent-host-pack-test.mjs index 23163ce..457b353 100644 --- a/packages/flowsafe/scripts/agent-host-pack-test.mjs +++ b/packages/flowsafe/scripts/agent-host-pack-test.mjs @@ -80,7 +80,7 @@ try { readFileSync(join(packageRoot, 'package.json'), 'utf8'), ); assert.equal(manifest.dependencies['@mastra/cloudflare-d1'], '1.1.1'); - assert.equal(manifest.peerDependencies['@mastra/core'], '^1.50.0'); + assert.equal(manifest.peerDependencies['@mastra/core'], '1.50.0'); assert.equal(manifest.dependencies.jose, sourceManifest.dependencies.jose); assert.equal( manifest.peerDependencies['@proofoftech/breakwater'], diff --git a/packages/flowsafe/src/agent-host/catalog.test.ts b/packages/flowsafe/src/agent-host/catalog.test.ts index 8a76e07..ae7cac5 100644 --- a/packages/flowsafe/src/agent-host/catalog.test.ts +++ b/packages/flowsafe/src/agent-host/catalog.test.ts @@ -5,6 +5,10 @@ import { describe, expect, it, vi } from 'vitest'; import type { ApprovalRole } from '../approval-api/index.js'; import { type AgentMeta, isPermissionIdentifier } from './types.js'; +const GUARDED_AGENT_HOST_PROTOCOL = Symbol.for( + '@proofoftech/breakwater/guarded-agent-host/v1', +); + vi.mock('@proofoftech/breakwater/agent', () => ({ isGuardedAgentHandle: (value: unknown) => typeof value === 'object' && @@ -22,6 +26,10 @@ function handle( ): GuardedAgentHandle { return { guarded: true, + [GUARDED_AGENT_HOST_PROTOCOL]: { + version: 1, + supportsDurableStructuredOutput: false, + }, id, allowedRoles, allowedPrincipalKinds, @@ -98,6 +106,31 @@ describe('agent catalog', () => { ]), ).toThrow('metadata roles must exactly match guarded agent roles'); }); + + it('rejects a guarded handle without the durable-host protocol', () => { + const legacy = { ...handle() } as Record; + delete legacy[GUARDED_AGENT_HOST_PROTOCOL]; + + expect(() => + createAgentModuleCatalog([ + { meta, agent: legacy as unknown as GuardedAgentHandle }, + ]), + ).toThrow(/without the durable-host protocol.*>=0\.12\.0/); + }); + + it('rejects an incompatible durable-host protocol', () => { + const incompatible = handle() as unknown as Record; + incompatible[GUARDED_AGENT_HOST_PROTOCOL] = { + version: 1, + supportsDurableStructuredOutput: true, + }; + + expect(() => + createAgentModuleCatalog([ + { meta, agent: incompatible as unknown as GuardedAgentHandle }, + ]), + ).toThrow(/malformed Breakwater guarded-agent host protocol/); + }); }); describe('agent permission declaration', () => { diff --git a/packages/flowsafe/src/agent-host/catalog.ts b/packages/flowsafe/src/agent-host/catalog.ts index c0f9359..c807b0f 100644 --- a/packages/flowsafe/src/agent-host/catalog.ts +++ b/packages/flowsafe/src/agent-host/catalog.ts @@ -1,7 +1,7 @@ // SPDX-License-Identifier: Apache-2.0 import { isGuardedAgentHandle } from '@proofoftech/breakwater/agent'; - +import { breakwaterGuardedAgentHostProtocol } from '../agent-runner/durable-agent-runner.js'; import { AGENT_ENTRY_PATHS } from '../agent-runner/index.js'; import { type ApprovalRole, @@ -221,6 +221,11 @@ export function validateAgentModule(module: AgentModule): AgentModule { if (!isGuardedAgentHandle(module.agent)) { fail(`agent '${meta.id}' must be created by createGuardedAgent`); } + if (breakwaterGuardedAgentHostProtocol(module.agent) === undefined) { + fail( + `agent '${meta.id}' was built by a @proofoftech/breakwater without the durable-host protocol; >=0.12.0 is required`, + ); + } if (module.agent.id !== meta.id) { fail( `metadata id '${meta.id}' does not match guarded agent id '${module.agent.id}'`, diff --git a/packages/flowsafe/src/agent-host/thread-host.test.ts b/packages/flowsafe/src/agent-host/thread-host.test.ts index adfc8d8..eb085f5 100644 --- a/packages/flowsafe/src/agent-host/thread-host.test.ts +++ b/packages/flowsafe/src/agent-host/thread-host.test.ts @@ -41,6 +41,9 @@ const mocked = vi.hoisted(() => ({ getHistory: vi.fn(), })); const RESOURCE_ID = resourceIdFromKey('acme_thread'); +const GUARDED_AGENT_HOST_PROTOCOL = Symbol.for( + '@proofoftech/breakwater/guarded-agent-host/v1', +); vi.mock('@proofoftech/breakwater/agent', () => ({ isGuardedAgentHandle: (value: unknown) => @@ -132,6 +135,10 @@ function guarded( ): GuardedAgentHandle { return { guarded: true, + [GUARDED_AGENT_HOST_PROTOCOL]: { + version: 1, + supportsDurableStructuredOutput: false, + }, id, allowedRoles: ['operator'], allowedPrincipalKinds: ['human', ...automationKinds], diff --git a/packages/flowsafe/src/agent-runner/durable-agent-runner.test.ts b/packages/flowsafe/src/agent-runner/durable-agent-runner.test.ts index 1c1a8be..7edbb0d 100644 --- a/packages/flowsafe/src/agent-runner/durable-agent-runner.test.ts +++ b/packages/flowsafe/src/agent-runner/durable-agent-runner.test.ts @@ -38,6 +38,7 @@ import { type Role, } from '@proofoftech/breakwater'; import { afterEach, describe, expect, it, vi } from 'vitest'; +import { z } from 'zod'; import { InvalidRunRequestError, @@ -124,6 +125,20 @@ function testAgent(id = 'writer'): Agent { }); } +function guardedTestAgent(): Agent { + return createGuardedAgent({ + id: 'writer', + name: 'Writer', + instructions: 'Answer the request.', + model: 'openai/gpt-4o-mini', + allowedRoles: ['operator'], + policies: [], + audit: new AuditLogger(), + maxSteps: 2, + toolChoice: 'auto', + }) as unknown as Agent; +} + function actorContext(role: Role = 'operator'): RequestContext { const context = new RequestContext(); context.set(ACTOR_CONTEXT_KEY, { id: 'actor-1', role }); @@ -235,6 +250,10 @@ const INPUT = { } as unknown as DurableAgenticWorkflowInput; describe('createFlowsafeDurableAgent', () => { + afterEach(() => { + vi.restoreAllMocks(); + }); + it('registers the raw agent and durable-agentic-loop workflow on the runtime', () => { // #given a runtime with nothing registered const { runtime, register, registerAgent } = fakeRuntime(); @@ -277,6 +296,138 @@ describe('createFlowsafeDurableAgent', () => { expect(isRuntimeDrivenAgent({})).toBe(false); expect(isRuntimeDrivenAgent(undefined)).toBe(false); }); + + it('rejects structured durable methods for a guarded agent before core dispatch', async () => { + const { runtime } = fakeRuntime(); + const durable = createFlowsafeDurableAgent({ + agent: guardedTestAgent(), + runtime, + }); + const schema = z.object({ answer: z.string() }); + const superStream = vi.spyOn(DurableAgent.prototype, 'stream'); + const superGenerate = vi.spyOn(DurableAgent.prototype, 'generate'); + const superPrepare = vi.spyOn(DurableAgent.prototype, 'prepare'); + + await expect( + durable.stream('hello', { + runId: 'run-1', + structuredOutput: { schema }, + } as never), + ).rejects.toThrow(/structuredOutput is not supported.*guarded agent/is); + await expect( + durable.generate('hello', { + runId: 'run-1', + structuredOutput: { schema }, + } as never), + ).rejects.toThrow(/structuredOutput is not supported.*guarded agent/is); + await expect( + durable.prepare('hello', { + runId: 'run-1', + structuredOutput: { schema }, + } as never), + ).rejects.toThrow(/structuredOutput is not supported.*guarded agent/is); + expect(superStream).not.toHaveBeenCalled(); + expect(superGenerate).not.toHaveBeenCalled(); + expect(superPrepare).not.toHaveBeenCalled(); + + await expect( + durable.generate('hello', { + runId: 'run-1', + structuredOutput: undefined, + } as never), + ).rejects.toThrow(/structuredOutput is not supported.*guarded agent/is); + expect(superGenerate).not.toHaveBeenCalled(); + }); + + it('snapshots durable call options before delegating to core', async () => { + const { runtime } = fakeRuntime(); + const durable = createFlowsafeDurableAgent({ + agent: guardedTestAgent(), + runtime, + }); + const superStream = vi + .spyOn(DurableAgent.prototype, 'stream') + .mockResolvedValue({ output: {} } as never); + const superGenerate = vi + .spyOn(DurableAgent.prototype, 'generate') + .mockResolvedValue({} as never); + const superPrepare = vi + .spyOn(DurableAgent.prototype, 'prepare') + .mockResolvedValue({} as never); + + for (const [method, coreMethod] of [ + ['stream', superStream], + ['generate', superGenerate], + ['prepare', superPrepare], + ] as const) { + const options: Record = { runId: 'run-1' }; + const pending = ( + durable[method] as unknown as ( + messages: string, + options: Record, + ) => Promise + )('hello', options); + options.runId = 'mutated-run'; + options.structuredOutput = { schema: z.object({ answer: z.string() }) }; + + await expect(pending).resolves.toBeDefined(); + const forwarded = coreMethod.mock.calls.at(-1)?.[1] as + | Record + | undefined; + expect(forwarded).toEqual({ runId: 'run-1' }); + expect(forwarded).not.toBe(options); + expect(Object.isFrozen(forwarded)).toBe(true); + } + }); + + it.each([ + 'stream', + 'generate', + 'prepare', + ] as const)('rejects accessor-backed %s options without invoking the accessor', async (method) => { + const { runtime } = fakeRuntime(); + const durable = createFlowsafeDurableAgent({ + agent: guardedTestAgent(), + runtime, + }); + const structuredOutput = vi.fn(() => undefined); + const options = { runId: 'run-1' } as Record; + Object.defineProperty(options, 'structuredOutput', { + enumerable: true, + get: structuredOutput, + }); + + await expect( + ( + durable[method] as unknown as ( + messages: string, + options: Record, + ) => Promise + )('hello', options), + ).rejects.toThrow(/structuredOutput.*data property/); + expect(structuredOutput).not.toHaveBeenCalled(); + }); + + it('rejects an accessor-backed runId without reading it or delegating', async () => { + const { runtime } = fakeRuntime(); + const durable = createFlowsafeDurableAgent({ + agent: guardedTestAgent(), + runtime, + }); + const superGenerate = vi.spyOn(DurableAgent.prototype, 'generate'); + const runId = vi.fn(() => 'run-1'); + const options: Record = {}; + Object.defineProperty(options, 'runId', { + enumerable: true, + get: runId, + }); + + await expect(durable.generate('hello', options as never)).rejects.toThrow( + /runId.*data property/, + ); + expect(runId).not.toHaveBeenCalled(); + expect(superGenerate).not.toHaveBeenCalled(); + }); }); describe('FlowsafeDurableAgent.executeWorkflow', () => { @@ -322,6 +473,57 @@ describe('FlowsafeDurableAgent.executeWorkflow', () => { }); describe('FlowsafeDurableAgent.streamUntilPersisted', () => { + it('snapshots call options before installing persistence state', async () => { + const { runtime } = fakeRuntime(); + const agent = createFlowsafeDurableAgent({ + agent: guardedTestAgent(), + runtime, + }); + const failure = new Error('stream unavailable'); + const stream = vi.spyOn(agent, 'stream').mockRejectedValue(failure); + const options: Record = { runId: 'run-1' }; + + const pending = agent.streamUntilPersisted( + 'hello', + options as never, + 'operator-1', + 'human', + ); + options.runId = 'mutated-run'; + options.structuredOutput = { schema: z.object({ answer: z.string() }) }; + + await expect(pending).rejects.toBe(failure); + expect(stream).toHaveBeenCalledOnce(); + expect(stream.mock.calls[0]?.[1]).toMatchObject({ runId: 'run-1' }); + expect(stream.mock.calls[0]?.[1]).not.toHaveProperty('structuredOutput'); + }); + + it('rejects accessor-backed call options before installing persistence state', async () => { + const { runtime } = fakeRuntime(); + const agent = createFlowsafeDurableAgent({ + agent: guardedTestAgent(), + runtime, + }); + const stream = vi.spyOn(agent, 'stream'); + const structuredOutput = vi.fn(() => undefined); + const options = { runId: 'run-1' } as Record; + Object.defineProperty(options, 'structuredOutput', { + enumerable: true, + get: structuredOutput, + }); + + await expect( + agent.streamUntilPersisted( + 'hello', + options as never, + 'operator-1', + 'human', + ), + ).rejects.toThrow(/structuredOutput.*data property/); + expect(structuredOutput).not.toHaveBeenCalled(); + expect(stream).not.toHaveBeenCalled(); + }); + it('does not resolve until the runtime has persisted the first summary', async () => { const { runtime, start } = fakeRuntime(); let releaseStart!: () => void; diff --git a/packages/flowsafe/src/agent-runner/durable-agent-runner.ts b/packages/flowsafe/src/agent-runner/durable-agent-runner.ts index aff2bb5..c6a3875 100644 --- a/packages/flowsafe/src/agent-runner/durable-agent-runner.ts +++ b/packages/flowsafe/src/agent-runner/durable-agent-runner.ts @@ -88,6 +88,75 @@ import { */ export const DURABLE_AGENTIC_LOOP_WORKFLOW_ID = 'durable-agentic-loop'; +const BREAKWATER_GUARDED_AGENT_HOST_PROTOCOL = Symbol.for( + '@proofoftech/breakwater/guarded-agent-host/v1', +); + +interface BreakwaterGuardedAgentHostProtocol { + readonly version: 1; + readonly supportsDurableStructuredOutput: false; +} + +function snapshotDurableCallOptions(options: T): T; +function snapshotDurableCallOptions(options: undefined): undefined; +function snapshotDurableCallOptions( + options: T | undefined, +): T | undefined; +function snapshotDurableCallOptions( + options: T | undefined, +): T | undefined { + if (options === undefined) return undefined; + if ( + options === null || + typeof options !== 'object' || + Array.isArray(options) + ) { + throw new TypeError('FlowsafeDurableAgent: call options must be an object'); + } + const snapshot: Record = {}; + for (const key of Reflect.ownKeys(options)) { + const descriptor = Object.getOwnPropertyDescriptor(options, key); + if (!descriptor) continue; + if (descriptor.get || descriptor.set) { + throw new TypeError( + `FlowsafeDurableAgent: call option '${String(key)}' must be a data property`, + ); + } + Object.defineProperty(snapshot, key, { + configurable: false, + enumerable: descriptor.enumerable, + value: descriptor.value, + writable: false, + }); + } + return Object.freeze(snapshot) as T; +} + +/** @internal Shared with the guarded agent catalog's compatibility check. */ +export function breakwaterGuardedAgentHostProtocol( + agent: unknown, +): BreakwaterGuardedAgentHostProtocol | undefined { + if ((typeof agent !== 'object' && typeof agent !== 'function') || !agent) { + return undefined; + } + const protocol = (agent as Record)[ + BREAKWATER_GUARDED_AGENT_HOST_PROTOCOL + ]; + if (protocol === undefined) return undefined; + if ( + !protocol || + typeof protocol !== 'object' || + (protocol as { version?: unknown }).version !== 1 || + (protocol as { supportsDurableStructuredOutput?: unknown }) + .supportsDurableStructuredOutput !== false + ) { + throw new TypeError( + 'FlowsafeDurableAgent: malformed Breakwater guarded-agent host protocol', + ); + } + return protocol as BreakwaterGuardedAgentHostProtocol; +} + function bindThreadCompletion( output: T, completion: Promise, @@ -205,6 +274,7 @@ export class FlowsafeDurableAgent< readonly [RUNTIME_DRIVEN_AGENT] = true; readonly #runtime: RunnerRuntime; readonly #wrappedAgent: Agent; + readonly #isBreakwaterGuardedAgent: boolean; readonly #threadRuntime?: Mastra['agentThreadStreamRuntime']; readonly #persistenceWaiters = new Map< string, @@ -222,6 +292,7 @@ export class FlowsafeDurableAgent< >(); constructor(options: FlowsafeDurableAgentOptions) { + const guardedProtocol = breakwaterGuardedAgentHostProtocol(options.agent); super({ agent: options.agent, id: options.id, @@ -238,6 +309,7 @@ export class FlowsafeDurableAgent< }); this.#runtime = options.runtime; this.#wrappedAgent = options.agent; + this.#isBreakwaterGuardedAgent = guardedProtocol !== undefined; this.#threadRuntime = options.threadRuntime; } @@ -264,6 +336,19 @@ export class FlowsafeDurableAgent< } } + #assertGuardedStructuredOutput(options: unknown): void { + if ( + this.#isBreakwaterGuardedAgent && + options !== null && + typeof options === 'object' && + Object.hasOwn(options, 'structuredOutput') + ) { + throw new TypeError( + 'FlowsafeDurableAgent: structuredOutput is not supported for a Breakwater guarded agent because Mastra durable execution bypasses the narrow guarded handle', + ); + } + } + /** * Enforce a caller-minted run ID before the inherited durable * `stream()` runs: without this the @@ -278,15 +363,17 @@ export class FlowsafeDurableAgent< ): Promise< Awaited['stream']>> > { - this.#assertCallerRunId(options?.runId); - const result = await super.stream(messages, options); - if (!options?.untilIdle) { + const callOptions = snapshotDurableCallOptions(options); + this.#assertCallerRunId(callOptions?.runId); + this.#assertGuardedStructuredOutput(callOptions); + const result = await super.stream(messages, callOptions); + if (!callOptions?.untilIdle) { await this.#threadRuntime?.registerRun( this as unknown as Parameters< Mastra['agentThreadStreamRuntime']['registerRun'] >[0], result.output, - (options ?? {}) as Parameters< + (callOptions ?? {}) as Parameters< Mastra['agentThreadStreamRuntime']['registerRun'] >[2], this.pubsub, @@ -318,14 +405,16 @@ export class FlowsafeDurableAgent< ): Promise< Awaited['stream']>> > { - this.#assertCallerRunId(options.runId); + const callOptions = snapshotDurableCallOptions(options); + this.#assertCallerRunId(callOptions?.runId); + this.#assertGuardedStructuredOutput(callOptions); if (!isExecutionPrincipalId(requestedBy)) { throw new InvalidRunRequestError('requestedBy is malformed'); } if (!isExecutionPrincipalKind(requestedByKind)) { throw new InvalidRunRequestError('requestedByKind is malformed'); } - const runId = options.runId; + const runId = callOptions.runId; if (this.#persistenceWaiters.has(runId)) { throw new InvalidRunRequestError( `run '${runId}' already has a pending durable start`, @@ -345,10 +434,10 @@ export class FlowsafeDurableAgent< if (scheduleDispatch) { this.#startScheduleDispatches.set(runId, scheduleDispatch); } - const onError = options.onError; + const onError = callOptions.onError; try { const result = await this.stream(messages, { - ...options, + ...callOptions, onError: async (data) => { reject( data.error instanceof Error @@ -385,8 +474,10 @@ export class FlowsafeDurableAgent< ): Promise< Awaited['generate']>> > { - this.#assertCallerRunId(options?.runId); - return super.generate(messages, options); + const callOptions = snapshotDurableCallOptions(options); + this.#assertCallerRunId(callOptions?.runId); + this.#assertGuardedStructuredOutput(callOptions); + return super.generate(messages, callOptions); } /** @@ -406,8 +497,10 @@ export class FlowsafeDurableAgent< ): Promise< Awaited['prepare']>> > { - this.#assertCallerRunId(options?.runId); - return super.prepare(messages, options); + const callOptions = snapshotDurableCallOptions(options); + this.#assertCallerRunId(callOptions?.runId); + this.#assertGuardedStructuredOutput(callOptions); + return super.prepare(messages, callOptions); } async #rehydrateRegistry(options: { From 802860568879e98049cf6cd1f86536075696888d Mon Sep 17 00:00:00 2001 From: gcharang <21151592+gcharang@users.noreply.github.com> Date: Sat, 15 Aug 2026 12:54:47 +0400 Subject: [PATCH 2/3] fix(fleet-control): suppress terminal re-entry audit --- .changeset/quiet-fleet-terminal-reentry.md | 5 +++++ docs/fleet-control.md | 4 ++-- packages/fleet-control/README.md | 2 +- packages/fleet-control/src/provision.ts | 1 - packages/fleet-control/test/provision.test.ts | 14 ++++++++++++-- 5 files changed, 20 insertions(+), 6 deletions(-) create mode 100644 .changeset/quiet-fleet-terminal-reentry.md diff --git a/.changeset/quiet-fleet-terminal-reentry.md b/.changeset/quiet-fleet-terminal-reentry.md new file mode 100644 index 0000000..7bb1c3f --- /dev/null +++ b/.changeset/quiet-fleet-terminal-reentry.md @@ -0,0 +1,5 @@ +--- +"@proofoftech/fleet-control": patch +--- + +Delete an already-decommissioned fleet ledger row without emitting a duplicate decommission event mislabeled as forced. diff --git a/docs/fleet-control.md b/docs/fleet-control.md index c54e1b7..7c7e42d 100644 --- a/docs/fleet-control.md +++ b/docs/fleet-control.md @@ -185,7 +185,7 @@ await forceDecommissionDeployment({ }); ``` -The function always acquires `store.withDeploymentLease()` before it reads the ledger. An absent deployment succeeds without a provider mutation. A terminal `decommissioned` record also succeeds and removes the retained ledger row. +The function always acquires `store.withDeploymentLease()` before it reads the ledger. An absent deployment succeeds without a provider mutation. A terminal `decommissioned` record also succeeds, removes the retained ledger row, and emits no audit event. A concurrent provision or decommission receives the same lease-acquisition error as `decommissionDeployment()`. A `database-reserved` row has not authorized provider creation, so force decommission removes that reservation without a provider call. @@ -199,7 +199,7 @@ The force path uses persisted resource identities only. It never fetches an arti Each provider mutation receives the active lease fence. `WranglerLoopBackend` requires route API implementations of both `getDatabase` and `deleteDatabase` before the first D1 lookup. The exact-ID lookup, deletion, and confirmation all run within that route fence; force deletion never falls back to Wrangler. The ledger persists `decommissioning`, `traffic-removed`, `credentials-revoked`, and `database-deleting`, so an interrupted call repeats the incomplete idempotent stage. -Provider 404 responses mean the resource is already absent. After D1 is absent, the function persists `decommissioned`, emits `DecommissionAuditEvent` with `forced: true`, and calls `lease.delete()`. If audit delivery fails, the terminal row remains and a retry emits the event again before deleting the row. Normal decommission uses the same optional event with `forced: false`. +Provider 404 responses mean the resource is already absent. After D1 is absent, the function persists `decommissioned`, emits `DecommissionAuditEvent` with `forced: true`, and calls `lease.delete()`. If initial audit delivery fails, the terminal row remains. A retry deletes it without repeating provider mutations or redelivering the event. Normal decommission uses the same optional event with `forced: false`. Force decommission does not delete the ordinary Worker script, application R2 buckets, or control-plane retention data. It removes the deployment’s ingress, live Worker secrets, database, and fleet ownership record. After the call returns, the host deletes its separate retention row and revokes its gateway key. Workers for Platforms fails closed unless its backend implements equivalent spec-free primitives; its dispatch route and trusted-resource topology cannot use the ordinary Worker route API contract. diff --git a/packages/fleet-control/README.md b/packages/fleet-control/README.md index a3949dc..34086ec 100644 --- a/packages/fleet-control/README.md +++ b/packages/fleet-control/README.md @@ -57,7 +57,7 @@ Use `forceDecommissionDeployment()` only when the host has lost the retained cre Provider 404 responses converge as already absent, so retry the same call after an interrupted teardown. A `database-reserved` row has not authorized provider creation and can be removed without a provider call. A `database-create-authorized` row has an unresolved creation outcome and only a synthetic ID, so force decommission fails closed and retains that row for spec-aware recovery. -Pass `options.audit` to receive a `DecommissionAuditEvent`. Normal decommission emits `forced: false`; force decommission emits the same event with `forced: true`. Fleet Control emits the force event before it deletes the ledger row. If the sink fails, the terminal record remains and the next call retries audit delivery without repeating provider mutations. +Pass `options.audit` to receive a `DecommissionAuditEvent`. Normal decommission emits `forced: false`; an actual force teardown emits the same event with `forced: true`. Fleet Control emits the force event before it deletes the ledger row. If the sink fails, the terminal record remains, and the next call deletes it without repeating provider mutations or redelivering the event. Force decommission of an already-terminal record also deletes the ledger row without emitting an event. The function never reads an artifact, computes a specification digest, deletes host-retained control-plane secrets, or deletes application R2 buckets. `WranglerLoopBackend` requires `PlainWorkerRouteApi.getDatabase` and `deleteDatabase`; it never falls back to Wrangler for force deletion. Other backends fail closed unless they implement the narrow `forceDecommissionStep` contract for equivalent provider primitives. diff --git a/packages/fleet-control/src/provision.ts b/packages/fleet-control/src/provision.ts index 315dd00..6ee7146 100644 --- a/packages/fleet-control/src/provision.ts +++ b/packages/fleet-control/src/provision.ts @@ -1677,7 +1677,6 @@ export async function forceDecommissionDeployment( ); } if (current.phase === 'decommissioned') { - await emitDecommissionAudit(input.options?.audit, current, true); await lease.delete(); return; } diff --git a/packages/fleet-control/test/provision.test.ts b/packages/fleet-control/test/provision.test.ts index 560c405..8b1d8de 100644 --- a/packages/fleet-control/test/provision.test.ts +++ b/packages/fleet-control/test/provision.test.ts @@ -2226,10 +2226,14 @@ describe('fleet provisioning', () => { spec: spec(), secrets, }); + const auditFlags: boolean[] = []; await decommissionDeployment({ backend, store: terminalStore, spec: spec(), + audit: (event) => { + auditFlags.push(event.forced); + }, }); Object.defineProperty(backend, 'forceDecommissionStep', { value: undefined, @@ -2242,11 +2246,17 @@ describe('fleet provisioning', () => { store: terminalStore, tenantTag: 'acme', environment: 'production', + options: { + audit: (event) => { + auditFlags.push(event.forced); + }, + }, }), ).resolves.toBeUndefined(); expect(terminalStore.leaseCalls).toBe(priorLeaseCalls + 1); expect(terminalStore.record).toBeUndefined(); expect(backend.forceSteps).toEqual([]); + expect(auditFlags).toEqual([false]); }); it('uses one audit event shape for normal and forced decommission', async () => { @@ -2388,7 +2398,7 @@ describe('fleet provisioning', () => { await expect(force).resolves.toBeUndefined(); }); - it('retries a completed force teardown when audit delivery fails', async () => { + it('silently removes a completed force teardown after audit delivery fails', async () => { const backend = new FakeBackend('plain-worker'); const store = new MemoryStore(); await provisionDeployment({ backend, store, spec: spec(), secrets }); @@ -2417,7 +2427,7 @@ describe('fleet provisioning', () => { 'delete-database', ]); expect(store.record).toBeUndefined(); - expect(auditAttempts).toBe(2); + expect(auditAttempts).toBe(1); }); it('rejects a concurrent lifecycle operation for the same deployment', async () => { From 9924bb48c1cea2549954aa3c3caeb1f5b0ddf6aa Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sat, 15 Aug 2026 09:05:29 +0000 Subject: [PATCH 3/3] release: version packages --- .changeset/gate-structured-output.md | 12 ------------ .changeset/quiet-fleet-terminal-reentry.md | 5 ----- packages/agent-starter/CHANGELOG.md | 8 ++++++++ packages/agent-starter/package.json | 2 +- packages/breakwater/CHANGELOG.md | 12 ++++++++++++ packages/breakwater/package.json | 2 +- packages/fleet-control/CHANGELOG.md | 8 ++++++++ packages/fleet-control/package.json | 2 +- packages/flowsafe/CHANGELOG.md | 12 ++++++++++++ packages/flowsafe/package.json | 2 +- packages/showcase/CHANGELOG.md | 8 ++++++++ packages/showcase/package.json | 2 +- 12 files changed, 53 insertions(+), 22 deletions(-) delete mode 100644 .changeset/gate-structured-output.md delete mode 100644 .changeset/quiet-fleet-terminal-reentry.md diff --git a/.changeset/gate-structured-output.md b/.changeset/gate-structured-output.md deleted file mode 100644 index a4fa89b..0000000 --- a/.changeset/gate-structured-output.md +++ /dev/null @@ -1,12 +0,0 @@ ---- -"@proofoftech/breakwater": minor -"@proofoftech/flowsafe": minor ---- - -Fail closed on structured-output coverage gaps. `createGuardedAgent()` rejects structured output before model execution because Mastra exposes parsed values to messages, persistence, and observability hooks before a post-generation wrapper could inspect them. It also rejects object-only policies that no supported guarded invocation can cover. - -Processor-visible object chunks are validated as JSON, evaluated through their canonical serialization, and replaced with the same canonical clone. Standalone object-only policies abort when an invocation exposes no object to the processor. Policy lists and decision-driving descriptors are snapshotted at construction, evaluator callables retain their original receiver, and per-stream audit metadata stays bounded by configured policies and channels. - -Flowsafe recognizes the new guarded-agent host protocol and rejects structured output on durable stream, generate, and prepare before Mastra can bypass the narrow handle. Durable entry points snapshot data-property call options before validation and delegation, reject accessors, and use the same snapshot for later run registration. Both packages pin their tested `@mastra/core` 1.50.0 contract. - -Hold-back cost under large streams is measured by opt-in evidence tests (`BREAKWATER_PERF=1`) and recorded in the policy-engine design guide. diff --git a/.changeset/quiet-fleet-terminal-reentry.md b/.changeset/quiet-fleet-terminal-reentry.md deleted file mode 100644 index 7bb1c3f..0000000 --- a/.changeset/quiet-fleet-terminal-reentry.md +++ /dev/null @@ -1,5 +0,0 @@ ---- -"@proofoftech/fleet-control": patch ---- - -Delete an already-decommissioned fleet ledger row without emitting a duplicate decommission event mislabeled as forced. diff --git a/packages/agent-starter/CHANGELOG.md b/packages/agent-starter/CHANGELOG.md index 8f3e254..61b5d56 100644 --- a/packages/agent-starter/CHANGELOG.md +++ b/packages/agent-starter/CHANGELOG.md @@ -1,5 +1,13 @@ # anchorage-agent-starter +## 0.0.13 + +### Patch Changes + +- Updated dependencies [37175fa] + - @proofoftech/breakwater@0.12.0 + - @proofoftech/flowsafe@0.17.0 + ## 0.0.12 ### Patch Changes diff --git a/packages/agent-starter/package.json b/packages/agent-starter/package.json index e362170..b955d79 100644 --- a/packages/agent-starter/package.json +++ b/packages/agent-starter/package.json @@ -1,6 +1,6 @@ { "name": "anchorage-agent-starter", - "version": "0.0.12", + "version": "0.0.13", "private": true, "description": "Production-shaped Cloudflare Workers starter for durable, approval-gated Anchorage agents", "type": "module", diff --git a/packages/breakwater/CHANGELOG.md b/packages/breakwater/CHANGELOG.md index e7688f3..eb82399 100644 --- a/packages/breakwater/CHANGELOG.md +++ b/packages/breakwater/CHANGELOG.md @@ -1,5 +1,17 @@ # @proofoftech/breakwater +## 0.12.0 + +### Minor Changes + +- 37175fa: Fail closed on structured-output coverage gaps. `createGuardedAgent()` rejects structured output before model execution because Mastra exposes parsed values to messages, persistence, and observability hooks before a post-generation wrapper could inspect them. It also rejects object-only policies that no supported guarded invocation can cover. + + Processor-visible object chunks are validated as JSON, evaluated through their canonical serialization, and replaced with the same canonical clone. Standalone object-only policies abort when an invocation exposes no object to the processor. Policy lists and decision-driving descriptors are snapshotted at construction, evaluator callables retain their original receiver, and per-stream audit metadata stays bounded by configured policies and channels. + + Flowsafe recognizes the new guarded-agent host protocol and rejects structured output on durable stream, generate, and prepare before Mastra can bypass the narrow handle. Durable entry points snapshot data-property call options before validation and delegation, reject accessors, and use the same snapshot for later run registration. Both packages pin their tested `@mastra/core` 1.50.0 contract. + + Hold-back cost under large streams is measured by opt-in evidence tests (`BREAKWATER_PERF=1`) and recorded in the policy-engine design guide. + ## 0.11.1 ### Patch Changes diff --git a/packages/breakwater/package.json b/packages/breakwater/package.json index 597836a..78d2267 100644 --- a/packages/breakwater/package.json +++ b/packages/breakwater/package.json @@ -1,6 +1,6 @@ { "name": "@proofoftech/breakwater", - "version": "0.11.1", + "version": "0.12.0", "publishConfig": { "access": "public" }, diff --git a/packages/fleet-control/CHANGELOG.md b/packages/fleet-control/CHANGELOG.md index c029c56..4a83acc 100644 --- a/packages/fleet-control/CHANGELOG.md +++ b/packages/fleet-control/CHANGELOG.md @@ -1,5 +1,13 @@ # @proofoftech/fleet-control +## 0.3.2 + +### Patch Changes + +- 8028605: Delete an already-decommissioned fleet ledger row without emitting a duplicate decommission event mislabeled as forced. +- Updated dependencies [37175fa] + - @proofoftech/flowsafe@0.17.0 + ## 0.3.1 ### Patch Changes diff --git a/packages/fleet-control/package.json b/packages/fleet-control/package.json index cba8f65..e3e05ce 100644 --- a/packages/fleet-control/package.json +++ b/packages/fleet-control/package.json @@ -1,6 +1,6 @@ { "name": "@proofoftech/fleet-control", - "version": "0.3.1", + "version": "0.3.2", "publishConfig": { "access": "public" }, diff --git a/packages/flowsafe/CHANGELOG.md b/packages/flowsafe/CHANGELOG.md index 9b18b90..34a6021 100644 --- a/packages/flowsafe/CHANGELOG.md +++ b/packages/flowsafe/CHANGELOG.md @@ -1,5 +1,17 @@ # @proofoftech/flowsafe +## 0.17.0 + +### Minor Changes + +- 37175fa: Fail closed on structured-output coverage gaps. `createGuardedAgent()` rejects structured output before model execution because Mastra exposes parsed values to messages, persistence, and observability hooks before a post-generation wrapper could inspect them. It also rejects object-only policies that no supported guarded invocation can cover. + + Processor-visible object chunks are validated as JSON, evaluated through their canonical serialization, and replaced with the same canonical clone. Standalone object-only policies abort when an invocation exposes no object to the processor. Policy lists and decision-driving descriptors are snapshotted at construction, evaluator callables retain their original receiver, and per-stream audit metadata stays bounded by configured policies and channels. + + Flowsafe recognizes the new guarded-agent host protocol and rejects structured output on durable stream, generate, and prepare before Mastra can bypass the narrow handle. Durable entry points snapshot data-property call options before validation and delegation, reject accessors, and use the same snapshot for later run registration. Both packages pin their tested `@mastra/core` 1.50.0 contract. + + Hold-back cost under large streams is measured by opt-in evidence tests (`BREAKWATER_PERF=1`) and recorded in the policy-engine design guide. + ## 0.16.1 ### Patch Changes diff --git a/packages/flowsafe/package.json b/packages/flowsafe/package.json index 98db169..ce729ba 100644 --- a/packages/flowsafe/package.json +++ b/packages/flowsafe/package.json @@ -1,6 +1,6 @@ { "name": "@proofoftech/flowsafe", - "version": "0.16.1", + "version": "0.17.0", "publishConfig": { "access": "public" }, diff --git a/packages/showcase/CHANGELOG.md b/packages/showcase/CHANGELOG.md index 74bb317..3b46801 100644 --- a/packages/showcase/CHANGELOG.md +++ b/packages/showcase/CHANGELOG.md @@ -1,5 +1,13 @@ # showcase +## 0.0.19 + +### Patch Changes + +- Updated dependencies [37175fa] + - @proofoftech/breakwater@0.12.0 + - @proofoftech/flowsafe@0.17.0 + ## 0.0.18 ### Patch Changes diff --git a/packages/showcase/package.json b/packages/showcase/package.json index eec8c64..3b43b84 100644 --- a/packages/showcase/package.json +++ b/packages/showcase/package.json @@ -1,6 +1,6 @@ { "name": "showcase", - "version": "0.0.18", + "version": "0.0.19", "private": true, "description": "Anchorage showcase — six runnable workflows and seven guardrail scenarios behind one React frontend", "type": "module",