From e9a263398c3bca4e5eef42bbc8324e755f4f7372 Mon Sep 17 00:00:00 2001 From: William Cory Date: Fri, 7 Aug 2026 23:00:17 -0700 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20feat(engine,kernel):=20one=20capabi?= =?UTF-8?q?lity-scoped=20flow=20wire=20for=20every=20placement?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FlowWire is the transport-neutral flow execution contract: one serializable Request in, one Response out, the same Flow definitions and FlowEngine underneath whether the serving function runs in a browser Service Worker's message listener, a local Bun process's HTTP route, an edge worker's fetch handler, or a cloud sandbox guest's socket loop. serveHttp adds the shared body-in/body-out projection those HTTP placements would otherwise each hand-roll, mapping refusals onto 400/403/404 while always encoding a Response body. CapabilityEnvelope is the one serializable authority that travels with a request: versioned capability patterns every placement interprets the same way, by intersecting the executing host's ambient CapabilitySet — an envelope can only narrow, and both the missing-interpreter and uninterpretable cases refuse execution instead of running wide. The exit now crosses the wire in its JSON codec form, and client rejections carry the refusal's code and detail so a caller can tell refused apart from ran-and-failed without decoding the reason by hand. Co-Authored-By: Claude Fable 5 --- packages/engine/src/FlowWire.ts | 417 ++++++++++++++++++ packages/engine/src/index.ts | 5 + packages/engine/test/FlowWire.test.ts | 278 ++++++++++++ packages/kernel/src/CapabilityEnvelope.ts | 129 ++++++ packages/kernel/src/index.ts | 8 + .../kernel/test/CapabilityEnvelope.test.ts | 105 +++++ packages/kernel/test/index.test.ts | 1 + 7 files changed, 943 insertions(+) create mode 100644 packages/engine/src/FlowWire.ts create mode 100644 packages/engine/test/FlowWire.test.ts create mode 100644 packages/kernel/src/CapabilityEnvelope.ts create mode 100644 packages/kernel/test/CapabilityEnvelope.test.ts diff --git a/packages/engine/src/FlowWire.ts b/packages/engine/src/FlowWire.ts new file mode 100644 index 00000000..974eef0f --- /dev/null +++ b/packages/engine/src/FlowWire.ts @@ -0,0 +1,417 @@ +/** + * The transport-neutral flow execution contract. + * + * `FlowProxy` projects flows onto Effect RPC groups and HTTP APIs. This module + * is the third projection, for runtimes where neither stack is practical — a + * browser Service Worker answering `postMessage`, an edge worker's `fetch` + * handler, a sandbox guest reading JSON off a socket. One serializable + * {@link Request} in, one serializable {@link Response} out, and the same + * `Flow` definitions and `FlowEngine` underneath as every other projection. + * + * The request may carry a capability envelope. The engine does not interpret + * envelopes itself — authority is the permission kernel's business — so hosts + * install an {@link EnvelopeInterpreter} that turns the opaque envelope value + * into ambient authority around the execution. Handling is fail-closed: a + * request that names an envelope is refused, not run wide, when no + * interpreter is installed. + * + * @since 0.1.0 + */ +import * as Cause from "effect/Cause" +import * as Context from "effect/Context" +import * as Effect from "effect/Effect" +import * as Exit from "effect/Exit" +import * as Layer from "effect/Layer" +import * as Option from "effect/Option" +import * as Schema from "effect/Schema" +import type * as Flow from "./Flow.ts" +import type { FlowEngine } from "./FlowEngine.ts" + +/** + * One serializable flow execution request. + * + * `payload` is the flow payload in its JSON encoding. `envelope` is an opaque + * serialized capability envelope; the engine hands it to the installed + * {@link EnvelopeInterpreter} without reading it. + * + * @category models + * @since 0.1.0 + */ +export class Request extends Schema.Class("@smithers/engine/FlowWire/Request")({ + flow: Schema.String, + payload: Schema.Json, + executionId: Schema.optional(Schema.String), + envelope: Schema.optional(Schema.Json) +}) {} + +/** + * The request named a flow the serving runtime does not register. + * + * @category errors + * @since 0.1.0 + */ +export class FlowNotFound extends Schema.TaggedErrorClass()( + "@smithers/engine/FlowWire/FlowNotFound", + { + flow: Schema.String + } +) {} + +/** + * The request's envelope could not become authority. + * + * `unsupported` — the serving runtime has no {@link EnvelopeInterpreter}. + * `uninterpretable` — the interpreter rejected the envelope value. + * Both refuse execution; an envelope is never silently ignored. + * + * @category errors + * @since 0.1.0 + */ +export class EnvelopeRejected extends Schema.TaggedErrorClass()( + "@smithers/engine/FlowWire/EnvelopeRejected", + { + code: Schema.Literals(["unsupported", "uninterpretable"]), + message: Schema.String + } +) {} + +/** + * The request value itself did not decode. + * + * @category errors + * @since 0.1.0 + */ +export class RequestInvalid extends Schema.TaggedErrorClass()( + "@smithers/engine/FlowWire/RequestInvalid", + { + message: Schema.String + } +) {} + +/** + * A request refused before the flow ran. No durable boundary was opened. + * + * @category models + * @since 0.1.0 + */ +export class Rejected extends Schema.Class("@smithers/engine/FlowWire/Rejected")({ + _tag: Schema.tag("Rejected"), + reason: Schema.Union([RequestInvalid, FlowNotFound, EnvelopeRejected]) +}) {} + +/** + * A request that reached the flow. `exit` is the flow's `Exit` in the JSON + * encoding of the flow's own success and error schemas. + * + * @category models + * @since 0.1.0 + */ +export class Completed extends Schema.Class("@smithers/engine/FlowWire/Completed")({ + _tag: Schema.tag("Completed"), + exit: Schema.Json +}) {} + +/** + * One serializable flow execution response. + * + * @category models + * @since 0.1.0 + */ +export const Response = Schema.Union([Completed, Rejected]) + +/** + * One serializable flow execution response. + * + * @category models + * @since 0.1.0 + */ +export type Response = typeof Response.Type + +/** + * Turns a serialized capability envelope into ambient authority around one + * execution. + * + * The engine defines only the seam. The canonical implementation decodes the + * value with `@smithers/kernel`'s `CapabilityEnvelope` and intersects the + * current `CapabilitySet`, so an envelope can only narrow what the executing + * host already allows. + * + * @category services + * @since 0.1.0 + */ +export class EnvelopeInterpreter extends Context.Service< + EnvelopeInterpreter, + { + readonly apply: ( + envelope: Schema.Json + ) => ( + effect: Effect.Effect + ) => Effect.Effect + } +>()("@smithers/engine/FlowWire/EnvelopeInterpreter") {} + +/** + * Builds an interpreter layer from an envelope decoder. + * + * The decoder validates the untrusted wire value and returns the pure + * attenuation combinator to wrap around the execution. Splitting decoding + * from wrapping is what keeps failures honest: a decoding failure becomes an + * `uninterpretable` refusal, while failures of the wrapped execution pass + * through untouched. + * + * @category layers + * @since 0.1.0 + */ +export const layerInterpreter = ( + decode: ( + envelope: Schema.Json + ) => Effect.Effect< + (effect: Effect.Effect) => Effect.Effect, + { readonly message: string } + > +): Layer.Layer => + Layer.succeed(EnvelopeInterpreter)({ + apply: (envelope) => (effect) => + decode(envelope).pipe( + Effect.mapError((failure) => new EnvelopeRejected({ code: "uninterpretable", message: failure.message })), + Effect.flatMap((attenuate) => attenuate(effect)) + ) + }) + +const exitSchema = (flow: Flow.AnyWithProps) => + Schema.toCodecJson( + Schema.Exit(flow.successSchema, flow.errorSchema, Schema.Defect()) + ) + +const decodeRequest = Schema.decodeUnknownEffect(Request) + +/** + * Builds the one serving function every placement shares. + * + * A Service Worker calls it from a message listener, an edge worker from its + * `fetch` handler, a local process from an HTTP route, a sandbox guest from + * its socket loop — the decoding, envelope handling, execution, and exit + * encoding are identical because they are this function. + * + * Envelope refusals raised by the interpreter come back as a {@link Rejected} + * response rather than an encoded flow failure, so a caller can always tell + * "the request was refused" apart from "the flow ran and failed". + * + * @category constructors + * @since 0.1.0 + */ +export const serve = >( + flows: Flows +): (input: unknown) => Effect.Effect> => { + const byTag = new Map() + for (const flow of flows) { + byTag.set(flow._tag, flow as Flow.AnyWithProps) + } + return (input) => + Effect.gen(function*() { + const decoded = yield* Effect.result(decodeRequest(input)) + if (decoded._tag === "Failure") { + return new Rejected({ + reason: new RequestInvalid({ + message: `The flow wire request did not decode: ${decoded.failure.message}` + }) + }) + } + const request = decoded.success + const flow = byTag.get(request.flow) + if (flow === undefined) { + return new Rejected({ reason: new FlowNotFound({ flow: request.flow }) }) + } + + const payload = yield* Effect.result( + Schema.decodeUnknownEffect(Schema.toCodecJson(flow.payloadSchema))(request.payload) + ) + if (payload._tag === "Failure") { + return new Rejected({ + reason: new RequestInvalid({ + message: `The payload for flow ${request.flow} did not decode: ${payload.failure.message}` + }) + }) + } + + let run: Effect.Effect = flow.execute(payload.success, { + executionId: request.executionId + }) as Effect.Effect + + if (request.envelope !== undefined) { + const interpreter = yield* Effect.serviceOption(EnvelopeInterpreter) + if (Option.isNone(interpreter)) { + return new Rejected({ + reason: new EnvelopeRejected({ + code: "unsupported", + message: "This runtime has no envelope interpreter installed; refusing to run with ambient authority" + }) + }) + } + run = interpreter.value.apply(request.envelope)(run) + } + + const exit = yield* Effect.exit(run) + const refusal = envelopeRefusal(exit) + if (refusal !== undefined) { + return new Rejected({ reason: refusal }) + } + const encoded = yield* Effect.orDie( + Schema.encodeEffect(exitSchema(flow))(exit as Exit.Exit) + ) + return new Completed({ exit: encoded }) + }) as Effect.Effect> +} + +/** + * One serialized HTTP projection of a wire response. + * + * @category models + * @since 0.1.0 + */ +export interface HttpResponse { + readonly status: 200 | 400 | 403 | 404 + readonly body: string +} + +const statusOf = (response: Response): HttpResponse["status"] => + response._tag === "Completed" + ? 200 + : response.reason instanceof FlowNotFound + ? 404 + : response.reason instanceof EnvelopeRejected + ? 403 + : 400 + +/** + * Builds the HTTP body-in, body-out projection of the serving function. + * + * An edge worker's `fetch` handler and a local process's HTTP route differ + * only in how they read the request body and write the response — the JSON + * decoding, envelope handling, execution, and status mapping are this + * function. Statuses: `200` the flow ran (the encoded exit carries success or + * typed failure), `400` invalid request, `403` refused envelope, `404` + * unknown flow. The body is always the encoded {@link Response}, so clients + * decode uniformly regardless of status. + * + * @category constructors + * @since 0.1.0 + */ +export const serveHttp = >( + flows: Flows +): (body: string) => Effect.Effect> => { + const handler = serve(flows) + return (body) => + Effect.gen(function*() { + const parsed = yield* Effect.result(Effect.try({ + try: () => JSON.parse(body) as unknown, + catch: (cause) => new RequestInvalid({ message: `The request body is not JSON: ${String(cause)}` }) + })) + const response = parsed._tag === "Failure" + ? new Rejected({ reason: parsed.failure }) + : yield* handler(parsed.success) + const encoded = yield* Effect.orDie(Schema.encodeEffect(Response)(response)) + return { status: statusOf(response), body: JSON.stringify(encoded) } + }) +} + +const envelopeRefusal = ( + exit: Exit.Exit +): EnvelopeRejected | undefined => { + if (!Exit.isFailure(exit)) { + return undefined + } + for (const reason of exit.cause.reasons) { + if (Cause.isFailReason(reason) && reason.error instanceof EnvelopeRejected) { + return reason.error + } + } + return undefined +} + +const decodeResponse = Schema.decodeUnknownEffect(Response) + +const describeReason = ( + reason: RequestInvalid | FlowNotFound | EnvelopeRejected +): string => + reason instanceof EnvelopeRejected + ? `${reason._tag} (${reason.code}): ${reason.message}` + : reason instanceof FlowNotFound + ? `${reason._tag}: ${reason.flow}` + : `${reason._tag}: ${reason.message}` + +/** + * A failure of the transport carrying a wire request, or a refusal from the + * serving side. + * + * @category errors + * @since 0.1.0 + */ +export class WireError extends Schema.TaggedErrorClass()( + "@smithers/engine/FlowWire/WireError", + { + message: Schema.String, + reason: Schema.optional(Schema.Union([RequestInvalid, FlowNotFound, EnvelopeRejected])), + cause: Schema.optional(Schema.Unknown) + } +) {} + +/** + * Builds a typed client over any request-in, response-out transport. + * + * The transport is one function — post JSON, receive JSON. Placements differ + * only in what that function does: call the serving function in-process, + * `postMessage` to a Service Worker, `fetch` an edge or sandbox endpoint. + * + * @category constructors + * @since 0.1.0 + */ +export const client = ( + post: (request: unknown) => Effect.Effect +) => +< + Tag extends string, + Payload extends Flow.AnyStructSchema, + Success extends Schema.Top, + Error extends Schema.Top +>( + flow: Flow.Flow, + payload: Payload["Type"], + options?: { + readonly executionId?: string | undefined + readonly envelope?: Schema.Json | undefined + } +): Effect.Effect => + Effect.gen(function*() { + const flowProps = flow as unknown as Flow.AnyWithProps + const encodedPayload = yield* Effect.orDie( + Schema.encodeEffect(Schema.toCodecJson(flowProps.payloadSchema))(payload) + ) + const request: Record = { + flow: flow._tag, + payload: encodedPayload + } + if (options?.executionId !== undefined) { + request.executionId = options.executionId + } + if (options?.envelope !== undefined) { + request.envelope = options.envelope + } + const raw = yield* post(request) + const response = yield* Effect.orDie(decodeResponse(raw)) + if (response._tag === "Rejected") { + return yield* Effect.fail( + new WireError({ + message: `The flow wire request was rejected: ${describeReason(response.reason)}`, + reason: response.reason + }) + ) + } + const exit = (yield* Effect.orDie( + Schema.decodeUnknownEffect(exitSchema(flowProps))(response.exit) + )) as Exit.Exit + if (Exit.isSuccess(exit)) { + return exit.value + } + return yield* Effect.failCause(exit.cause) + }) as Effect.Effect diff --git a/packages/engine/src/index.ts b/packages/engine/src/index.ts index 9e818a36..416f6ad7 100644 --- a/packages/engine/src/index.ts +++ b/packages/engine/src/index.ts @@ -44,6 +44,11 @@ export * as FlowProxy from "./FlowProxy.ts" */ export * as FlowProxyServer from "./FlowProxyServer.ts" +/** + * @since 0.1.0 + */ +export * as FlowWire from "./FlowWire.ts" + /** * @since 4.0.0 */ diff --git a/packages/engine/test/FlowWire.test.ts b/packages/engine/test/FlowWire.test.ts new file mode 100644 index 00000000..4d2299f8 --- /dev/null +++ b/packages/engine/test/FlowWire.test.ts @@ -0,0 +1,278 @@ +import { Context, Effect, Layer, Schema } from "effect" +import { describe, expect, it } from "vitest" +import { Flow, FlowEngine, FlowWire } from "../src/index.ts" + +const effect = (name: string, body: () => Effect.Effect) => + it(name, () => Effect.runPromise(body() as Effect.Effect)) + +/** + * A stand-in for kernel ambient authority: a fiber `Context.Reference` the + * interpreter narrows, exactly the shape `CapabilitySet.attenuate` has. The + * engine test must not depend on `@smithers/kernel`, so the reference is + * local. + */ +const Authority = Context.Reference>( + "test/FlowWire/Authority", + { defaultValue: () => ["*"] } +) + +const Echo = Flow.make("Wire/Echo", { + payload: { value: Schema.Number }, + success: Schema.Number, + error: Schema.Literal("negative"), + idempotencyKey: ({ value }) => String(value) +}) + +const Observe = Flow.make("Wire/Observe", { + payload: { probe: Schema.String }, + success: Schema.Struct({ granted: Schema.Boolean }), + idempotencyKey: ({ probe }) => probe +}) + +const flows = [Echo, Observe] as const + +const handlers = Layer.mergeAll( + Echo.toLayer(({ value }) => value < 0 ? Effect.fail("negative" as const) : Effect.succeed(value + 1)), + Observe.toLayer(({ probe }) => + Effect.gen(function*() { + const authority = yield* Authority + return { granted: authority.includes("*") || authority.includes(probe) } + }) + ) +).pipe(Layer.provideMerge(FlowEngine.layerMemory)) + +/** + * The test interpreter reads `{ allow: [...] }` envelopes and replaces the + * wildcard authority with the envelope's list — monotone in spirit, minimal + * in mechanics. + */ +const EnvelopeSchema = Schema.Struct({ allow: Schema.Array(Schema.String) }) + +const interpreter = FlowWire.layerInterpreter((envelope) => + Schema.decodeUnknownEffect(EnvelopeSchema)(envelope).pipe( + Effect.mapError((error) => ({ message: error.message })), + Effect.map((decoded) => (wrapped: Effect.Effect): Effect.Effect => + Effect.updateService(wrapped, Authority, (current) => + current.filter((entry) => + entry !== "*" + ).concat(decoded.allow)) + ) + ) +) + +// The wire is JSON: every request/response crosses a stringify/parse +// boundary, exactly like postMessage or fetch. +const overTheWire = ( + handler: (input: unknown) => Effect.Effect +) => +(request: unknown): Effect.Effect => + Effect.gen(function*() { + const serialized = JSON.parse(JSON.stringify(request)) + const response = yield* handler(serialized) + const encoded = yield* Effect.orDie(Schema.encodeEffect(FlowWire.Response)(response)) + return JSON.parse(JSON.stringify(encoded)) + }) as unknown as Effect.Effect + +describe("FlowWire", () => { + effect("serves an execute request end to end over JSON", () => { + const call = FlowWire.client(overTheWire(FlowWire.serve(flows))) + return Effect.gen(function*() { + const result = yield* call(Echo, { value: 41 }, { executionId: "wire-echo" }) + expect(result).toBe(42) + // Without options the execution id derives from the idempotency key. + const optionless = yield* call(Echo, { value: 10 }) + expect(optionless).toBe(11) + }).pipe(Effect.provide(interpreter), Effect.provide(handlers)) + }) + + effect("returns the flow's typed error through the wire", () => { + const call = FlowWire.client(overTheWire(FlowWire.serve(flows))) + return Effect.gen(function*() { + const exit = yield* Effect.exit(call(Echo, { value: -1 }, { executionId: "wire-negative" })) + expect(exit._tag).toBe("Failure") + expect(String(exit)).toContain("negative") + }).pipe(Effect.provide(interpreter), Effect.provide(handlers)) + }) + + effect("rejects a request for an unregistered flow", () => { + const serveOnlyEcho = FlowWire.serve([Echo] as const) + return Effect.gen(function*() { + const response = yield* serveOnlyEcho({ + flow: "Wire/Observe", + payload: { probe: "net" }, + executionId: "missing" + }) + expect(response._tag).toBe("Rejected") + if (response._tag === "Rejected") { + expect(response.reason._tag).toBe("@smithers/engine/FlowWire/FlowNotFound") + } + }).pipe(Effect.provide(handlers)) + }) + + effect("rejects an undecodable request", () => { + const handler = FlowWire.serve(flows) + return Effect.gen(function*() { + const response = yield* handler({ nonsense: true }) + expect(response._tag).toBe("Rejected") + if (response._tag === "Rejected") { + expect(response.reason._tag).toBe("@smithers/engine/FlowWire/RequestInvalid") + } + }).pipe(Effect.provide(handlers)) + }) + + effect("rejects an undecodable payload", () => { + const handler = FlowWire.serve(flows) + return Effect.gen(function*() { + const response = yield* handler({ + flow: "Wire/Echo", + payload: { value: "not a number" }, + executionId: "bad-payload" + }) + expect(response._tag).toBe("Rejected") + if (response._tag === "Rejected") { + expect(response.reason._tag).toBe("@smithers/engine/FlowWire/RequestInvalid") + } + }).pipe(Effect.provide(handlers)) + }) + + effect("an envelope narrows the authority the handler observes", () => { + const call = FlowWire.client(overTheWire(FlowWire.serve(flows))) + return Effect.gen(function*() { + const granted = yield* call(Observe, { probe: "net" }, { + executionId: "enveloped-allow", + envelope: { allow: ["net"] } + }) + expect(granted).toEqual({ granted: true }) + const denied = yield* call(Observe, { probe: "fs" }, { + executionId: "enveloped-deny", + envelope: { allow: ["net"] } + }) + expect(denied).toEqual({ granted: false }) + // Without an envelope the ambient wildcard authority applies. + const ambient = yield* call(Observe, { probe: "fs" }, { executionId: "ambient" }) + expect(ambient).toEqual({ granted: true }) + }).pipe(Effect.provide(interpreter), Effect.provide(handlers)) + }) + + effect("fails closed when an envelope arrives and no interpreter is installed", () => { + const call = FlowWire.client(overTheWire(FlowWire.serve(flows))) + return Effect.gen(function*() { + const exit = yield* Effect.exit(call(Observe, { probe: "net" }, { + executionId: "no-interpreter", + envelope: { allow: ["net"] } + })) + expect(exit._tag).toBe("Failure") + expect(String(exit)).toContain("unsupported") + }).pipe(Effect.provide(handlers)) + }) + + effect("fails closed when the interpreter cannot read the envelope", () => { + const call = FlowWire.client(overTheWire(FlowWire.serve(flows))) + return Effect.gen(function*() { + const exit = yield* Effect.exit(call(Observe, { probe: "net" }, { + executionId: "bad-envelope", + envelope: { deny: true } + })) + expect(exit._tag).toBe("Failure") + expect(String(exit)).toContain("uninterpretable") + }).pipe(Effect.provide(interpreter), Effect.provide(handlers)) + }) + + effect("a client call for an unregistered flow surfaces the flow name", () => { + const call = FlowWire.client(overTheWire(FlowWire.serve([Echo] as const))) + return Effect.gen(function*() { + const exit = yield* Effect.exit(call(Observe, { probe: "net" }, { executionId: "client-missing" })) + expect(exit._tag).toBe("Failure") + expect(String(exit)).toContain("FlowNotFound") + expect(String(exit)).toContain("Wire/Observe") + }).pipe(Effect.provide(handlers)) + }) + + effect("schema drift between client and server is a refusal, not a crash", () => { + // The server's "Wire/Echo" takes a string payload; the client still holds + // the numeric definition. The mismatch must come back as a refusal. + const DriftedEcho = Flow.make("Wire/Echo", { + payload: { value: Schema.String }, + success: Schema.String, + idempotencyKey: ({ value }) => value + }) + const call = FlowWire.client(overTheWire(FlowWire.serve([DriftedEcho] as const))) + return Effect.gen(function*() { + const exit = yield* Effect.exit(call(Echo, { value: 7 }, { executionId: "drift" })) + expect(exit._tag).toBe("Failure") + expect(String(exit)).toContain("RequestInvalid") + expect(String(exit)).toContain("did not decode") + }).pipe(Effect.provide(handlers)) + }) + + effect("the HTTP projection maps refusals onto statuses and always encodes a response body", () => { + const handler = FlowWire.serveHttp(flows) + const decodeBody = (body: string) => Effect.orDie(Schema.decodeUnknownEffect(FlowWire.Response)(JSON.parse(body))) + return Effect.gen(function*() { + const completed = yield* handler(JSON.stringify({ + flow: "Wire/Echo", + payload: { value: 41 }, + executionId: "http-echo" + })) + expect(completed.status).toBe(200) + expect((yield* decodeBody(completed.body))._tag).toBe("Completed") + + // A flow that ran and failed is still a 200: the exit carries the error. + const failed = yield* handler(JSON.stringify({ + flow: "Wire/Echo", + payload: { value: -1 }, + executionId: "http-negative" + })) + expect(failed.status).toBe(200) + + const notJson = yield* handler("{not json") + expect(notJson.status).toBe(400) + const notJsonBody = yield* decodeBody(notJson.body) + expect(notJsonBody._tag).toBe("Rejected") + + const badPayload = yield* handler(JSON.stringify({ + flow: "Wire/Echo", + payload: { value: "not a number" }, + executionId: "http-bad" + })) + expect(badPayload.status).toBe(400) + + const missing = yield* handler(JSON.stringify({ + flow: "Wire/Missing", + payload: {}, + executionId: "http-missing" + })) + expect(missing.status).toBe(404) + + const enveloped = yield* handler(JSON.stringify({ + flow: "Wire/Echo", + payload: { value: 1 }, + executionId: "http-enveloped", + envelope: { allow: ["net"] } + })) + expect(enveloped.status).toBe(403) + }).pipe(Effect.provide(handlers)) + }) + + effect("repeated wire requests for one execution id deduplicate", () => { + let calls = 0 + const Counted = Flow.make("Wire/Counted", { + payload: { value: Schema.Number }, + success: Schema.Number, + idempotencyKey: ({ value }) => String(value) + }) + const layer = Counted.toLayer(({ value }) => + Effect.sync(() => { + calls++ + return value * 2 + }) + ).pipe(Layer.provideMerge(FlowEngine.layerMemory)) + const call = FlowWire.client(overTheWire(FlowWire.serve([Counted] as const))) + return Effect.gen(function*() { + const first = yield* call(Counted, { value: 4 }, { executionId: "dedupe" }) + const second = yield* call(Counted, { value: 4 }, { executionId: "dedupe" }) + expect([first, second]).toEqual([8, 8]) + expect(calls).toBe(1) + }).pipe(Effect.provide(layer)) + }) +}) diff --git a/packages/kernel/src/CapabilityEnvelope.ts b/packages/kernel/src/CapabilityEnvelope.ts new file mode 100644 index 00000000..8de7763c --- /dev/null +++ b/packages/kernel/src/CapabilityEnvelope.ts @@ -0,0 +1,129 @@ +/** + * The serializable capability envelope. + * + * One flow execution request names its authority once, as data, and every + * placement — a browser Service Worker, a local Bun process, an edge worker, + * a cloud sandbox — turns that same data into ambient authority the same way: + * by intersecting it with whatever the executing host already allows. An + * envelope can therefore only narrow; a request cannot mint capability its + * host never had, no matter which runtime executes it. + * + * Governing design: + * `docs/specs/Concepts/Permission Kernel.md` and + * `docs/specs/Concepts/Effect Taxonomy.md`. + * + * @since 0.1.0 + */ +import { Effect, Schema } from "effect" +import { CapabilityPattern } from "./Capability.ts" +import * as CapabilitySet from "./CapabilitySet.ts" + +/** + * The wire version this module reads and writes. Interpreters fail closed on + * any other value rather than guessing at future semantics. + * + * @category models + * @since 0.1.0 + */ +export const version = "flows/capability-envelope/v1" as const + +/** + * A versioned, JSON-serializable any-of group of capability patterns. + * + * The empty pattern list is the empty envelope: it denies every capability, + * because an intersected empty any-of group can never match. Unrestricted + * authority is expressed by *omitting* the envelope, never by an envelope + * value. + * + * @category models + * @since 0.1.0 + */ +export class CapabilityEnvelope extends Schema.Class( + "@smithers/kernel/CapabilityEnvelope" +)({ + version: Schema.Literal(version), + patterns: Schema.Array(CapabilityPattern) +}) {} + +/** + * Constructs an envelope from capability patterns. + * + * @category constructors + * @since 0.1.0 + */ +export const make = ( + patterns: ReadonlyArray +): CapabilityEnvelope => new CapabilityEnvelope({ version, patterns }) + +/** + * An envelope that could not be decoded. Interpretation is fail-closed: an + * uninterpretable envelope refuses execution instead of running with the + * host's ambient authority. + * + * @category errors + * @since 0.1.0 + */ +export class CapabilityEnvelopeError extends Schema.TaggedErrorClass()( + "@smithers/kernel/CapabilityEnvelopeError", + { + message: Schema.String + } +) {} + +const decodeEnvelope = Schema.decodeUnknownEffect(CapabilityEnvelope) + +/** + * Decodes an untrusted wire value into an envelope, failing closed. + * + * @category parsing + * @since 0.1.0 + */ +export const decode = ( + input: unknown +): Effect.Effect => + decodeEnvelope(input).pipe( + Effect.mapError((issue) => + new CapabilityEnvelopeError({ + message: `The capability envelope is not a valid ${version} value: ${issue.message}` + }) + ) + ) + +const encodeEnvelope = Schema.encodeEffect(Schema.toCodecJson(CapabilityEnvelope)) + +/** + * Encodes an envelope for the wire. + * + * @category encoding + * @since 0.1.0 + */ +export const encode = ( + envelope: CapabilityEnvelope +): Effect.Effect => Effect.orDie(encodeEnvelope(envelope)) + +/** + * Runs an effect with authority intersected with the envelope's patterns. + * + * This is deliberately the only conversion from envelope to authority, and it + * is monotone: `CapabilitySet.attenuate` intersects, so the executing host's + * own policy always still applies. + * + * @category combinators + * @since 0.1.0 + */ +export const apply = ( + envelope: CapabilityEnvelope +): (effect: Effect.Effect) => Effect.Effect => CapabilitySet.attenuate(envelope.patterns) + +/** + * Decodes an untrusted wire value and applies it in one step — the shape a + * host needs to interpret envelopes arriving with remote execution requests. + * + * @category combinators + * @since 0.1.0 + */ +export const interpret = (input: unknown) => +( + effect: Effect.Effect +): Effect.Effect => + Effect.flatMap(decode(input), (envelope) => apply(envelope)(effect)) diff --git a/packages/kernel/src/index.ts b/packages/kernel/src/index.ts index a0eadac4..a7cb47c4 100644 --- a/packages/kernel/src/index.ts +++ b/packages/kernel/src/index.ts @@ -12,6 +12,14 @@ */ export * as Capability from "./Capability.ts" +/** + * The serializable capability envelope carried by flow execution requests. + * + * @category namespace exports + * @since 0.1.0 + */ +export * as CapabilityEnvelope from "./CapabilityEnvelope.ts" + /** * Monotone ambient authority and intersection. * diff --git a/packages/kernel/test/CapabilityEnvelope.test.ts b/packages/kernel/test/CapabilityEnvelope.test.ts new file mode 100644 index 00000000..c27dd68a --- /dev/null +++ b/packages/kernel/test/CapabilityEnvelope.test.ts @@ -0,0 +1,105 @@ +import { Effect } from "effect" +import { describe, expect, it } from "vitest" +import { Capability, CapabilityPattern } from "../src/Capability.ts" +import * as CapabilityEnvelope from "../src/CapabilityEnvelope.ts" +import * as CapabilitySet from "../src/CapabilitySet.ts" + +const run = (effect: Effect.Effect) => Effect.runPromise(effect) + +const pattern = (action: CapabilityPattern["action"], resource: string) => new CapabilityPattern({ action, resource }) + +const netGet = new Capability({ action: "net:get", resource: "https://example.com/data" }) +const fsWrite = new Capability({ action: "fs:write", resource: "/workspace/out.txt" }) + +describe("CapabilityEnvelope", () => { + it("round-trips through its JSON encoding", () => + run(Effect.gen(function*() { + const envelope = CapabilityEnvelope.make([pattern("net:*", "**"), pattern("fs:read", "/workspace/**")]) + const encoded = yield* CapabilityEnvelope.encode(envelope) + // the wire value is plain JSON: survives stringify/parse untouched + const wire: unknown = JSON.parse(JSON.stringify(encoded)) + const decoded = yield* CapabilityEnvelope.decode(wire) + expect(decoded.version).toBe(CapabilityEnvelope.version) + expect(decoded.patterns).toHaveLength(2) + expect(decoded.patterns[0]?.action).toBe("net:*") + }))) + + it("fails closed on an unknown version", async () => { + const exit = await Effect.runPromiseExit( + CapabilityEnvelope.decode({ version: "flows/capability-envelope/v2", patterns: [] }) + ) + expect(exit._tag).toBe("Failure") + }) + + it("fails closed on a malformed value", async () => { + const exit = await Effect.runPromiseExit(CapabilityEnvelope.decode("not an envelope")) + expect(exit._tag).toBe("Failure") + }) + + it("apply intersects: the envelope narrows ambient authority", () => + run( + Effect.gen(function*() { + const set = yield* CapabilitySet.current + expect(CapabilitySet.allows(set, netGet)).toBe(true) + expect(CapabilitySet.allows(set, fsWrite)).toBe(true) + }).pipe( + CapabilityEnvelope.apply( + CapabilityEnvelope.make([pattern("net:*", "**"), pattern("fs:write", "/workspace/**")]) + ) + ) + )) + + it("apply denies capabilities outside the envelope", () => + run( + Effect.gen(function*() { + const set = yield* CapabilitySet.current + expect(CapabilitySet.allows(set, netGet)).toBe(false) + expect(CapabilitySet.allows(set, fsWrite)).toBe(true) + }).pipe( + CapabilityEnvelope.apply(CapabilityEnvelope.make([pattern("fs:*", "**")])) + ) + )) + + it("the empty envelope denies everything", () => + run( + Effect.gen(function*() { + const set = yield* CapabilitySet.current + expect(CapabilitySet.allows(set, netGet)).toBe(false) + expect(CapabilitySet.allows(set, fsWrite)).toBe(false) + }).pipe(CapabilityEnvelope.apply(CapabilityEnvelope.make([]))) + )) + + it("apply cannot widen an already-attenuated fiber", () => + run( + Effect.gen(function*() { + const set = yield* CapabilitySet.current + // The outer attenuation to fs-only still applies even though the + // envelope would allow the network. + expect(CapabilitySet.allows(set, netGet)).toBe(false) + expect(CapabilitySet.allows(set, fsWrite)).toBe(true) + }).pipe( + CapabilityEnvelope.apply(CapabilityEnvelope.make([pattern("*", "**")])), + CapabilitySet.attenuate([pattern("fs:*", "**")]) + ) + )) + + it("interpret decodes then applies in one step", () => + run(Effect.gen(function*() { + const wire = yield* CapabilityEnvelope.encode( + CapabilityEnvelope.make([pattern("net:get", "https://example.com/**")]) + ) + const set = yield* CapabilityEnvelope.interpret(wire)(CapabilitySet.current) + expect(CapabilitySet.allows(set, netGet)).toBe(true) + expect(CapabilitySet.allows(set, fsWrite)).toBe(false) + }))) + + it("interpret fails with CapabilityEnvelopeError on garbage", async () => { + const exit = await Effect.runPromiseExit( + CapabilityEnvelope.interpret({ nonsense: true })(Effect.void) + ) + expect(exit._tag).toBe("Failure") + if (exit._tag === "Failure") { + expect(String(exit.cause)).toContain("CapabilityEnvelopeError") + } + }) +}) diff --git a/packages/kernel/test/index.test.ts b/packages/kernel/test/index.test.ts index 8bd426f3..03266701 100644 --- a/packages/kernel/test/index.test.ts +++ b/packages/kernel/test/index.test.ts @@ -5,6 +5,7 @@ describe("kernel package barrel", () => { it("exports every public namespace", () => { expect(Object.keys(Kernel).sort()).toEqual([ "Capability", + "CapabilityEnvelope", "CapabilitySet", "FileSystem", "GrantEvent",