From a8a3fbec49007e3222e413be03cdf06cbf9f5857 Mon Sep 17 00:00:00 2001 From: Alex Langenfeld Date: Fri, 4 Sep 2026 14:09:19 -0500 Subject: [PATCH 1/3] Add Run#getWritable() for appending to another run's stream A holder or watchdog run can own a session's stream while short-lived turn runs contribute to it, but until now a contributor could only get a writable if the owner handed one over through start(). getWritable() reconstructs it from the owner's run ID alone. The handle targets the owner's (runId, name) and seals to the owner's public key where it has one, so it grants append access without read capability. It carries the existing forwarding symbols, so passing it through start() and into a step keeps the owner's identity and stays on the zero-lookup path. Signed-off-by: Alex Langenfeld --- .changeset/run-get-writable.md | 6 + .../v4/api-reference/workflow-api/get-run.mdx | 39 ++ .../content/docs/v4/foundations/streaming.mdx | 88 +++++ .../v5/api-reference/workflow-api/get-run.mdx | 39 ++ .../content/docs/v5/foundations/streaming.mdx | 88 +++++ packages/core/src/runtime.ts | 1 + .../core/src/runtime/run-get-writable.test.ts | 369 ++++++++++++++++++ packages/core/src/runtime/run.ts | 142 ++++++- packages/core/src/serialization.ts | 155 +++++--- packages/workflow/src/api-workflow.ts | 1 + packages/workflow/src/api.ts | 1 + 11 files changed, 873 insertions(+), 56 deletions(-) create mode 100644 .changeset/run-get-writable.md create mode 100644 packages/core/src/runtime/run-get-writable.test.ts diff --git a/.changeset/run-get-writable.md b/.changeset/run-get-writable.md new file mode 100644 index 0000000000..ee2b7735d2 --- /dev/null +++ b/.changeset/run-get-writable.md @@ -0,0 +1,6 @@ +--- +'@workflow/core': minor +'workflow': minor +--- + +Add `Run#getWritable()`, which opens a writable onto an existing run's stream from its run ID so one run can append to a stream another run owns. diff --git a/docs/content/docs/v4/api-reference/workflow-api/get-run.mdx b/docs/content/docs/v4/api-reference/workflow-api/get-run.mdx index 27764ab352..60e81eabe7 100644 --- a/docs/content/docs/v4/api-reference/workflow-api/get-run.mdx +++ b/docs/content/docs/v4/api-reference/workflow-api/get-run.mdx @@ -67,6 +67,20 @@ import type { WorkflowReadableStreamOptions } from "workflow/api"; export default WorkflowReadableStreamOptions;`} /> +#### WorkflowRunWritableStreamOptions + +`run.getWritable()` returns a `WritableStream` that appends to the run's stream, letting one run contribute to a stream another run owns. See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream). + + + + +`getWritable()` grants append access, not read access: where the owning run publishes an encryption public key, writes are sealed to it and cannot be read back by the writer. Closing the returned writable closes the **shared** stream for every writer, so contributors should `releaseLock()` instead. The run must already exist; an unknown run rejects with `WorkflowRunNotFoundError`. + + #### StopSleepOptions (); // [!code highlight] + const writer = writable.getWriter(); + + await writer.write({ turn, text: "done" }); + + // Release rather than close: the stream belongs to the holder run. // [!code highlight] + writer.releaseLock(); // [!code highlight] +} +``` + +Pass `{ namespace: "name" }` to target one of the owning run's namespaced streams. See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream) for the full pattern and its caveats. + ## Related functions - [`start()`](/docs/api-reference/workflow-api/start): Start a new workflow and get its run ID. diff --git a/docs/content/docs/v4/foundations/streaming.mdx b/docs/content/docs/v4/foundations/streaming.mdx index cb1c9b2970..65a9d3816b 100644 --- a/docs/content/docs/v4/foundations/streaming.mdx +++ b/docs/content/docs/v4/foundations/streaming.mdx @@ -313,6 +313,94 @@ export async function POST(request: Request) { } ``` +## Writing to another run's stream + +`getWritable()` gives a run a writable onto its own stream. `getRun(runId).getWritable()` gives you a writable onto a *different* run's stream, reconstructed from that run's ID alone. + +This exists for the holder pattern: one long-lived run owns a session's stream, and short-lived runs contribute to it. Clients read one stream for the whole session, while the work that fills it is split across runs that start and finish independently. Because the writable is derived from the run ID, the holder never has to hand a handle to each contributor. + +```typescript title="workflows/session-holder.ts" lineNumbers +import { sleep } from "workflow"; + +// This run exists to own the session's stream. Its ID is the session's +// identity: clients read from it, and every turn writes to it. +export async function sessionHolder(sessionId: string) { + "use workflow"; + + await sleep("24h"); + await archiveSession(sessionId); +} + +async function archiveSession(sessionId: string) { + "use step"; + // Clean up when the session's lifetime is over. +} +``` + +Each turn is its own workflow. It knows only the holder's run ID: + +```typescript title="workflows/turn.ts" lineNumbers +import { getRun } from "workflow/api"; + +type SessionEvent = { turn: number; text: string }; + +export async function turnWorkflow(holderRunId: string, turn: number) { + "use workflow"; + + await runTurn(holderRunId, turn); +} + +async function runTurn(holderRunId: string, turn: number) { + "use step"; + + const writable = await getRun(holderRunId).getWritable(); // [!code highlight] + const writer = writable.getWriter(); + + await writer.write({ turn, text: "thinking" }); + await writer.write({ turn, text: "done" }); + + // Release, don't close: the stream outlives this turn. // [!code highlight] + writer.releaseLock(); // [!code highlight] +} +``` + +The client reads the holder's stream once and sees every turn: + +```typescript title="app/api/session/[id]/route.ts" lineNumbers +import { getRun } from "workflow/api"; + +type SessionEvent = { turn: number; text: string }; + +export async function GET( + request: Request, + { params }: { params: Promise<{ id: string }> } +) { + const { id } = await params; + const readable = getRun(id).getReadable(); + + return new Response(readable, { + headers: { "Content-Type": "text/event-stream" } + }); +} +``` + +Pass `{ namespace: 'name' }` to target one of the holder's [namespaced streams](#namespaced-streams), exactly as with `getReadable()`. + +### What this does and does not grant + + +**`writer.close()` closes the shared stream.** + +The handle is an ordinary `WritableStream`, so closing it ends the stream for *every* writer, and a closed stream cannot be reopened. Contributors should `releaseLock()` when they are done. Releasing the lock flushes pending writes, so there is never a need to close a stream just to get writes out. Leave closing to the run that owns the stream. + + +- **Append access, not read access.** Where the owning run publishes an [encryption](/docs/how-it-works/encryption) public key, contributions are sealed to it. The writing run cannot decrypt the stream, including its own writes. Read the stream with `getReadable()` from somewhere that holds the owning run's key. +- **No authorization of its own.** Reaching the run through your World is the authorization. This inherits whatever access the caller already had, so treat a run ID reaching untrusted code the same way you would treat any other capability. +- **The owner keeps the lifecycle.** The stream's region, retention, and terminal state follow the owning run. When that run's stream is closed or expires, contributors' writes fail; nothing about holding a writable extends it. +- **The run must already exist.** `getWritable()` rejects with `WorkflowRunNotFoundError` for an unknown run rather than creating one. When a holder and its first turn start together, start the holder first, or handle that rejection. + +The returned writable can be passed to `start()` and forwarded into steps like any other stream, and it keeps pointing at the owning run's stream wherever it travels. + ## Common patterns ### Progress updates for long-running tasks diff --git a/docs/content/docs/v5/api-reference/workflow-api/get-run.mdx b/docs/content/docs/v5/api-reference/workflow-api/get-run.mdx index 88bba5eb20..8e32099baf 100644 --- a/docs/content/docs/v5/api-reference/workflow-api/get-run.mdx +++ b/docs/content/docs/v5/api-reference/workflow-api/get-run.mdx @@ -65,6 +65,20 @@ import type { WorkflowReadableStreamOptions } from "workflow/api"; export default WorkflowReadableStreamOptions;`} /> +#### WorkflowRunWritableStreamOptions + +`run.getWritable()` returns a `WritableStream` that appends to the run's stream, letting one run contribute to a stream another run owns. See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream). + + + + +`getWritable()` grants append access, not read access: where the owning run publishes an encryption public key, writes are sealed to it and cannot be read back by the writer. Closing the returned writable closes the **shared** stream for every writer, so contributors should `releaseLock()` instead. The run must already exist; an unknown run rejects with `WorkflowRunNotFoundError`. + + #### StopSleepOptions (); // [!code highlight] + const writer = writable.getWriter(); + + await writer.write({ turn, text: "done" }); + + // Release rather than close: the stream belongs to the holder run. // [!code highlight] + writer.releaseLock(); // [!code highlight] +} +``` + +Pass `{ namespace: "name" }` to target one of the owning run's namespaced streams. See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream) for the full pattern and its caveats. + ## Related functions - [`start()`](/docs/api-reference/workflow-api/start): Start a new workflow and get its run ID. diff --git a/docs/content/docs/v5/foundations/streaming.mdx b/docs/content/docs/v5/foundations/streaming.mdx index 7e2477a437..79ac5caea2 100644 --- a/docs/content/docs/v5/foundations/streaming.mdx +++ b/docs/content/docs/v5/foundations/streaming.mdx @@ -314,6 +314,94 @@ export async function POST(request: Request) { } ``` +## Writing to another run's stream + +`getWritable()` gives a run a writable onto its own stream. `getRun(runId).getWritable()` gives you a writable onto a *different* run's stream, reconstructed from that run's ID alone. + +This exists for the holder pattern: one long-lived run owns a session's stream, and short-lived runs contribute to it. Clients read one stream for the whole session, while the work that fills it is split across runs that start and finish independently. Because the writable is derived from the run ID, the holder never has to hand a handle to each contributor. + +```typescript title="workflows/session-holder.ts" lineNumbers +import { sleep } from "workflow"; + +// This run exists to own the session's stream. Its ID is the session's +// identity: clients read from it, and every turn writes to it. +export async function sessionHolder(sessionId: string) { + "use workflow"; + + await sleep("24h"); + await archiveSession(sessionId); +} + +async function archiveSession(sessionId: string) { + "use step"; + // Clean up when the session's lifetime is over. +} +``` + +Each turn is its own workflow. It knows only the holder's run ID: + +```typescript title="workflows/turn.ts" lineNumbers +import { getRun } from "workflow/api"; + +type SessionEvent = { turn: number; text: string }; + +export async function turnWorkflow(holderRunId: string, turn: number) { + "use workflow"; + + await runTurn(holderRunId, turn); +} + +async function runTurn(holderRunId: string, turn: number) { + "use step"; + + const writable = await getRun(holderRunId).getWritable(); // [!code highlight] + const writer = writable.getWriter(); + + await writer.write({ turn, text: "thinking" }); + await writer.write({ turn, text: "done" }); + + // Release, don't close: the stream outlives this turn. // [!code highlight] + writer.releaseLock(); // [!code highlight] +} +``` + +The client reads the holder's stream once and sees every turn: + +```typescript title="app/api/session/[id]/route.ts" lineNumbers +import { getRun } from "workflow/api"; + +type SessionEvent = { turn: number; text: string }; + +export async function GET( + request: Request, + { params }: { params: Promise<{ id: string }> } +) { + const { id } = await params; + const readable = getRun(id).getReadable(); + + return new Response(readable, { + headers: { "Content-Type": "text/event-stream" } + }); +} +``` + +Pass `{ namespace: 'name' }` to target one of the holder's [namespaced streams](#namespaced-streams), exactly as with `getReadable()`. + +### What this does and does not grant + + +**`writer.close()` closes the shared stream.** + +The handle is an ordinary `WritableStream`, so closing it ends the stream for *every* writer, and a closed stream cannot be reopened. Contributors should `releaseLock()` when they are done. Releasing the lock flushes pending writes, so there is never a need to close a stream just to get writes out. Leave closing to the run that owns the stream. + + +- **Append access, not read access.** Where the owning run publishes an [encryption](/docs/how-it-works/encryption) public key, contributions are sealed to it. The writing run cannot decrypt the stream, including its own writes. Read the stream with `getReadable()` from somewhere that holds the owning run's key. +- **No authorization of its own.** Reaching the run through your World is the authorization. This inherits whatever access the caller already had, so treat a run ID reaching untrusted code the same way you would treat any other capability. +- **The owner keeps the lifecycle.** The stream's region, retention, and terminal state follow the owning run. When that run's stream is closed or expires, contributors' writes fail; nothing about holding a writable extends it. +- **The run must already exist.** `getWritable()` rejects with `WorkflowRunNotFoundError` for an unknown run rather than creating one. When a holder and its first turn start together, start the holder first, or handle that rejection. + +The returned writable can be passed to `start()` and forwarded into steps like any other stream, and it keeps pointing at the owning run's stream wherever it travels. + ## Common patterns ### Progress updates for long-running tasks diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index d9699fad35..c298b7dca4 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -161,6 +161,7 @@ export { Run, type WorkflowReadableStream, type WorkflowReadableStreamOptions, + type WorkflowRunWritableStreamOptions, } from './runtime/run.js'; export { type CancelRunOptions, diff --git a/packages/core/src/runtime/run-get-writable.test.ts b/packages/core/src/runtime/run-get-writable.test.ts new file mode 100644 index 0000000000..0ae36ba0d9 --- /dev/null +++ b/packages/core/src/runtime/run-get-writable.test.ts @@ -0,0 +1,369 @@ +/** + * `Run#getWritable()` — append access to a stream another run owns, + * reconstructed from the run ID alone. + * + * What these tests pin down is that the handle it returns is indistinguishable + * from one the owner handed over itself: same `(runId, name)` target, same + * forwarding symbols, same sealed-frame encoding. The two are produced by + * different code paths (owner metadata here, a wire descriptor in the + * reviver), and a divergence between them surfaces as a frame the owner cannot + * decrypt rather than as a type error, so the encoding is asserted at the byte + * level rather than by trusting the shape. + */ +import { WorkflowRunNotFoundError } from '@workflow/errors'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { importKey } from '../encryption.js'; +import { bytesToBase64, deriveRunKeyPair } from '../sealed-box.js'; +import { + dehydrateStepArguments, + dehydrateWorkflowArguments, + hydrateStepArguments, + hydrateWorkflowArguments, +} from '../serialization.js'; +import { + decrypt as decryptEnvelope, + runPayloadKeys, +} from '../serialization/encryption.js'; +import { hydrateData } from '../serialization-format.js'; +import { + STREAM_NAME_SYMBOL, + STREAM_SERVER_DEPLOYMENT_ID_SYMBOL, + STREAM_SERVER_PUBLIC_KEY_SYMBOL, + STREAM_SERVER_RUN_ID_SYMBOL, +} from '../symbols.js'; +import { getWorldLazy } from './get-world-lazy.js'; +import { getRun, Run } from './run.js'; + +vi.mock('../version.js', () => ({ version: '0.0.0-test' })); +vi.mock('./get-world-lazy.js', () => ({ getWorldLazy: vi.fn() })); + +const OWNER_RUN_ID = 'wrun_ownerrun'; +const OWNER_STREAM = 'strm_ownerrun_user'; +const OWNER_MATERIAL = new Uint8Array(32).fill(0x2b); + +/** + * The subset of a run record `getWritable()` reads. `resolveData: 'none'` + * keeps `deploymentId` and `encryptionPublicKey`, which is the whole reason it + * can skip resolving payload refs. + */ +function ownerRun(overrides: Record = {}) { + return { + runId: OWNER_RUN_ID, + status: 'running', + deploymentId: 'dpl_owner', + ...overrides, + }; +} + +function mockWorld(overrides: Record = {}) { + const world = { + runs: { get: vi.fn().mockResolvedValue(ownerRun()) }, + streams: { + write: vi.fn().mockResolvedValue(undefined), + close: vi.fn().mockResolvedValue(undefined), + getInfo: vi.fn().mockResolvedValue({ tailIndex: -1, done: false }), + }, + ...overrides, + } as any; + vi.mocked(getWorldLazy).mockReturnValue(world); + return world; +} + +/** Frames the sink actually received, in order. */ +function framesWritten(world: any): Uint8Array[] { + return world.streams.write.mock.calls + .map((c: any[]) => c[2]) + .filter((c: unknown) => c instanceof Uint8Array); +} + +afterEach(() => { + vi.clearAllMocks(); + vi.useRealTimers(); +}); + +describe('Run#getWritable', () => { + it('targets the owner run default stream', async () => { + const world = mockWorld(); + + const ops: Promise[] = []; + const writable = await getRun(OWNER_RUN_ID).getWritable({ ops }); + const writer = writable.getWriter(); + await writer.write('from-a-contributor'); + writer.releaseLock(); + await Promise.all(ops); + + expect(world.runs.get).toHaveBeenCalledWith(OWNER_RUN_ID, { + resolveData: 'none', + }); + expect(world.streams.write).toHaveBeenCalled(); + for (const call of world.streams.write.mock.calls) { + expect(call[0]).toBe(OWNER_RUN_ID); + expect(call[1]).toBe(OWNER_STREAM); + } + }); + + it('targets the owner run namespaced stream', async () => { + const world = mockWorld(); + + const ops: Promise[] = []; + const writable = await getRun(OWNER_RUN_ID).getWritable({ + namespace: 'session-events', + ops, + }); + const writer = writable.getWriter(); + await writer.write('namespaced'); + writer.releaseLock(); + await Promise.all(ops); + + // base64url('session-events'), appended to the run's default stream name. + const expected = `${OWNER_STREAM}_c2Vzc2lvbi1ldmVudHM`; + expect(world.streams.write.mock.calls[0][1]).toBe(expected); + expect((writable as any)[STREAM_NAME_SYMBOL]).toBe(expected); + }); + + it('rejects for a run that does not exist', async () => { + // The API reconstructs a handle onto an existing stream; it must never + // bring a run or a stream into being as a side effect of asking for one. + const world = mockWorld({ + runs: { + get: vi + .fn() + .mockRejectedValue(new WorkflowRunNotFoundError('wrun_ghost')), + }, + }); + + await expect(getRun('wrun_ghost').getWritable()).rejects.toThrow( + WorkflowRunNotFoundError + ); + expect(world.streams.write).not.toHaveBeenCalled(); + // Fail fast: one look, no backoff, for an ID the caller supplied. + expect(world.runs.get).toHaveBeenCalledTimes(1); + }); + + it('seals to the owner public key without resolving a symmetric key', async () => { + // The owner published its X25519 public key, so the contributor can seal + // with no key-API round trip — and, more importantly, without ever holding + // a key that could read the stream back. + const ownerKeyPair = await deriveRunKeyPair(OWNER_MATERIAL); + const getEncryptionKeyForRun = vi.fn(); + const world = mockWorld({ + getEncryptionKeyForRun, + runs: { + get: vi.fn().mockResolvedValue( + ownerRun({ + encryptionPublicKey: bytesToBase64(ownerKeyPair.publicKey), + }) + ), + }, + }); + + const ops: Promise[] = []; + const writable = await getRun(OWNER_RUN_ID).getWritable({ ops }); + const writer = writable.getWriter(); + await writer.write('sealed-to-owner'); + writer.releaseLock(); + await Promise.all(ops); + + expect(getEncryptionKeyForRun).not.toHaveBeenCalled(); + + const frames = framesWritten(world); + expect(frames.length).toBeGreaterThan(0); + // [4-byte length][encp][sealed...] + expect(new TextDecoder().decode(frames[0].subarray(4, 8))).toBe('encp'); + + // And the owner really can open it. + const ownerKeys = runPayloadKeys( + await importKey(OWNER_MATERIAL), + ownerKeyPair + ); + const opened = await decryptEnvelope(frames[0].subarray(4), ownerKeys); + expect(hydrateData(opened as Uint8Array, {})).toBe('sealed-to-owner'); + }); + + it('falls back to the owner deployment key when it published no public key', async () => { + // Runs created by older SDKs carry no public key. They still have to be + // writable, via the symmetric key imported encrypt-only. + const getEncryptionKeyForRun = vi.fn().mockResolvedValue(OWNER_MATERIAL); + const world = mockWorld({ getEncryptionKeyForRun }); + + const ops: Promise[] = []; + const writable = await getRun(OWNER_RUN_ID).getWritable({ ops }); + const writer = writable.getWriter(); + await writer.write('legacy-owner'); + writer.releaseLock(); + await Promise.all(ops); + + // Resolved from the deployment on the run record, so the tier that has to + // load the owning run first never runs. + expect(getEncryptionKeyForRun).toHaveBeenCalledWith(OWNER_RUN_ID, { + deploymentId: 'dpl_owner', + }); + expect(world.runs.get).toHaveBeenCalledTimes(1); + + const frames = framesWritten(world); + expect(new TextDecoder().decode(frames[0].subarray(4, 8))).toBe('encr'); + }); + + it('carries every forwarding symbol on the returned handle', async () => { + const ownerKeyPair = await deriveRunKeyPair(OWNER_MATERIAL); + const ownerPublicKey = bytesToBase64(ownerKeyPair.publicKey); + mockWorld({ + runs: { + get: vi + .fn() + .mockResolvedValue(ownerRun({ encryptionPublicKey: ownerPublicKey })), + }, + }); + + const writable = await getRun(OWNER_RUN_ID).getWritable(); + + expect((writable as any)[STREAM_NAME_SYMBOL]).toBe(OWNER_STREAM); + expect((writable as any)[STREAM_SERVER_RUN_ID_SYMBOL]).toBe(OWNER_RUN_ID); + expect((writable as any)[STREAM_SERVER_DEPLOYMENT_ID_SYMBOL]).toBe( + 'dpl_owner' + ); + expect((writable as any)[STREAM_SERVER_PUBLIC_KEY_SYMBOL]).toBe( + ownerPublicKey + ); + }); + + it('does not advertise a public key the owner never published', async () => { + // Stamping one here would send later contributors to seal to an address + // the owner cannot open. + mockWorld({ getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined) }); + + const writable = await getRun(OWNER_RUN_ID).getWritable(); + + expect((writable as any)[STREAM_SERVER_PUBLIC_KEY_SYMBOL]).toBeUndefined(); + }); + + it('keeps the owner identity through start() and into a step', async () => { + // The motivating shape: a turn workflow holding only the holder run's ID + // opens a writable, hands it to the workflow it starts, and that workflow + // hands it to the step that writes. Two serialization hops, and the owner + // must survive both or the writes land on the wrong stream. + const ownerKeyPair = await deriveRunKeyPair(OWNER_MATERIAL); + const ownerPublicKey = bytesToBase64(ownerKeyPair.publicKey); + const world = mockWorld({ + runs: { + get: vi + .fn() + .mockResolvedValue(ownerRun({ encryptionPublicKey: ownerPublicKey })), + }, + getEncryptionKeyForRun: vi.fn(), + }); + + const writable = await getRun(OWNER_RUN_ID).getWritable(); + + // Hop 1: start() dehydrates into the turn workflow's arguments. + const forwarded = await dehydrateWorkflowArguments( + writable, + 'wrun_turn', + undefined + ); + const inWorkflow = (await hydrateWorkflowArguments( + forwarded, + 'wrun_turn', + undefined + )) as WritableStream; + expect((inWorkflow as any)[STREAM_SERVER_RUN_ID_SYMBOL]).toBe(OWNER_RUN_ID); + expect((inWorkflow as any)[STREAM_SERVER_PUBLIC_KEY_SYMBOL]).toBe( + ownerPublicKey + ); + + // Hop 2: the workflow passes it into the step that writes. + const toStep = await dehydrateStepArguments( + inWorkflow, + 'wrun_turn', + undefined + ); + const ops: Promise[] = []; + const inStep = (await hydrateStepArguments( + toStep, + 'wrun_turn', + undefined, + ops, + globalThis, + {}, + 'dpl_turn' + )) as WritableStream; + + const writer = inStep.getWriter(); + await writer.write('written-two-hops-away'); + writer.releaseLock(); + await Promise.all(ops); + + // Still the owner's stream, still sealed, still no key lookup. + expect(world.getEncryptionKeyForRun).not.toHaveBeenCalled(); + const ownerFrames = world.streams.write.mock.calls.filter( + (c: any[]) => c[0] === OWNER_RUN_ID && c[1] === OWNER_STREAM + ); + expect(ownerFrames.length).toBeGreaterThan(0); + + const ownerKeys = runPayloadKeys( + await importKey(OWNER_MATERIAL), + ownerKeyPair + ); + const frame = ownerFrames[0][2] as Uint8Array; + const opened = await decryptEnvelope(frame.subarray(4), ownerKeys); + expect(hydrateData(opened as Uint8Array, {})).toBe('written-two-hops-away'); + }); + + it('drains on lock release without closing the shared stream', async () => { + // The contract that makes this safe for per-turn contributors: releasing + // the writer settles the flush, so nobody has to close a stream they do + // not own just to get their writes out. + const world = mockWorld(); + + const ops: Promise[] = []; + const writable = await getRun(OWNER_RUN_ID).getWritable({ ops }); + const writer = writable.getWriter(); + await writer.write('flushed-without-close'); + writer.releaseLock(); + + await Promise.all(ops); + + expect(world.streams.write).toHaveBeenCalled(); + expect(world.streams.close).not.toHaveBeenCalled(); + }); + + it('closes the owner stream when the caller closes the handle', async () => { + // Documented consequence rather than a feature: this handle is a normal + // WritableStream, so close() ends the shared stream for every writer. + const world = mockWorld(); + + const ops: Promise[] = []; + const writable = await getRun(OWNER_RUN_ID).getWritable({ ops }); + const writer = writable.getWriter(); + await writer.write('last-chunk'); + await writer.close(); + await Promise.all(ops); + + expect(world.streams.close).toHaveBeenCalledWith( + OWNER_RUN_ID, + OWNER_STREAM + ); + }); + + it('retries a missing run only for a resiliently started run', async () => { + // A Run handed back by an optimistic start() may legitimately not exist + // yet. Same bounded budget the return-value poll uses; no open-ended + // polling, and getRun() never opts into it. + vi.useFakeTimers(); + const runsGet = vi + .fn() + .mockRejectedValueOnce(new WorkflowRunNotFoundError(OWNER_RUN_ID)) + .mockResolvedValue(ownerRun()); + mockWorld({ runs: { get: runsGet } }); + + const pending = new Run(OWNER_RUN_ID, { + resilientStart: true, + }).getWritable(); + + await vi.advanceTimersByTimeAsync(1_000); + const writable = await pending; + + expect(runsGet).toHaveBeenCalledTimes(2); + expect((writable as any)[STREAM_SERVER_RUN_ID_SYMBOL]).toBe(OWNER_RUN_ID); + }); +}); diff --git a/packages/core/src/runtime/run.ts b/packages/core/src/runtime/run.ts index 2f116612fe..3f5068b2b7 100644 --- a/packages/core/src/runtime/run.ts +++ b/packages/core/src/runtime/run.ts @@ -13,6 +13,8 @@ import { type PayloadKey, } from '../serialization/encryption.js'; import { + createForwardedWritable, + getForwardedWritableEncryptionKey, getRunReadableStream, hydrateRunError, hydrateWorkflowReturnValue, @@ -32,6 +34,18 @@ const PAYLOAD_TERMINAL_RUN_STATUSES = new Set([ 'failed', ]); +/** + * Backoff for the "run does not exist yet" retry, used only by runs whose + * `run_created` event failed and that the resilient start path will create via + * `run_started`. 1s + 3s + 6s gives the queue 10s to deliver before the caller + * sees the 404. + * + * A `Run` from `getRun()` never uses this: for an ID a caller supplied, a + * missing run is a real error and reporting it immediately beats a ten-second + * pause on the way to the same answer. + */ +const NOT_FOUND_RETRY_DELAYS = [1_000, 3_000, 6_000]; + /** @internal */ export function getReturnValuePollIntervalMs(): number { return envNumber( @@ -154,6 +168,33 @@ export interface WorkflowReadableStreamOptions { global?: Record; } +/** + * Options for configuring a writable onto a workflow run's stream. + */ +export interface WorkflowRunWritableStreamOptions { + /** + * An optional namespace to distinguish between multiple streams associated + * with the same workflow run. + */ + namespace?: string; + /** + * Any asynchronous operations to complete before pausing or terminating the + * execution environment + * (i.e. using [`waitUntil()`](https://developer.mozilla.org/docs/Web/API/ExtendableEvent/waitUntil) or similar). + * + * Writes are acknowledged on buffer entry, so a caller that needs them + * durable before the environment goes away should pass an array here and + * await it after releasing the writer lock. + */ + ops?: Promise[]; + /** + * The global object to use for reducing types from the global scope. + * + * Defaults to {@link [`globalThis`](https://developer.mozilla.org/docs/Web/JavaScript/Reference/Global_Objects/globalThis)}. + */ + global?: Record; +} + /** * A handler class for a workflow run. */ @@ -394,6 +435,100 @@ export class Run { }); } + /** + * Retrieves a writable onto this run's stream, the one + * {@link Run.getReadable | getReadable()} reads and the run's own + * `getWritable()` writes. + * + * This is append access to a stream another run owns, reconstructed from the + * run ID alone. It exists for the holder pattern: a long-lived run owns a + * session's stream, and short-lived runs that know only its ID contribute to + * it without the owner having to hand a writable to each one. The handle can + * be forwarded onward through `start()` and into steps like any other. + * + * The run must already exist. A missing run rejects with + * {@link WorkflowRunNotFoundError} rather than creating anything, except on a + * `Run` from a resilient `start()`, which retries briefly while the run is + * still being created. + * + * Two things this deliberately does not give you: + * + * - **No read access.** Frames are sealed to the owner's public key where it + * has one, so the writer cannot decrypt this stream, including its own + * writes. Read it with `getReadable()` from somewhere holding the run's key. + * - **No authorization of its own.** Reaching the run through the World is + * the authorization; this inherits whatever the caller already had. + * + * The owner keeps the stream's lifecycle: its region, retention, and terminal + * state follow the owning run, not the caller. + * + * @remarks + * `writer.close()` closes the **shared** stream for everyone, not just this + * handle, and a closed stream cannot be reopened. A contributor should write + * and then `releaseLock()`; releasing the lock drains pending writes, so + * closing is never needed merely to flush. Leave closing to the run that owns + * the stream. + * + * @param options - The options for the writable stream. + * @returns A `WritableStream` targeting this run's stream. + * @throws WorkflowRunNotFoundError if the run does not exist. + */ + async getWritable( + options: WorkflowRunWritableStreamOptions = {} + ): Promise> { + 'use step'; + const { ops = [], global = globalThis, namespace } = options; + const run = await this.#getMetadata(); + const name = getWorkflowRunStreamId(this.runId, namespace); + + // Resolve before returning rather than handing the promise to the + // serializer: the sealed path costs no I/O, and on the fallback path a key + // that cannot be resolved is better reported here than on a write the + // caller has already been told was accepted. + const key = await getForwardedWritableEncryptionKey( + this.runId, + run.deploymentId, + run.encryptionPublicKey + ); + + return createForwardedWritable({ + global, + ops, + runId: this.runId, + name, + key, + deploymentId: run.deploymentId, + encryptionPublicKey: run.encryptionPublicKey, + }); + } + + /** + * Reads this run's metadata, absorbing the window in which a resiliently + * started run does not exist yet. + * + * `resolveData: 'none'` is deliberate: `deploymentId` and + * `encryptionPublicKey` both survive it, and resolving input/output payload + * refs to reach them would be work thrown away. + * @internal + */ + async #getMetadata() { + const world = await this.#lazyWorldPromise; + const maxRetries = this.#resilientStart ? NOT_FOUND_RETRY_DELAYS.length : 0; + let attempt = 0; + while (true) { + try { + return await world.runs.get(this.runId, { resolveData: 'none' }); + } catch (error) { + if (!WorkflowRunNotFoundError.is(error) || attempt >= maxRetries) { + throw error; + } + const delay = NOT_FOUND_RETRY_DELAYS[attempt]!; + attempt++; + await new Promise((resolve) => setTimeout(resolve, delay)); + } + } + } + /** @internal */ async #resolveTerminalReturnValue(run: WorkflowRun): Promise { if (run.status === 'completed') { @@ -451,8 +586,9 @@ export class Run { // and the runtime to create the run via run_started. // When resilientStart is false, 404 is a real error: fail fast. let notFoundRetries = 0; - const NOT_FOUND_MAX_RETRIES = this.#resilientStart ? 3 : 0; - const NOT_FOUND_DELAYS = [1_000, 3_000, 6_000]; + const NOT_FOUND_MAX_RETRIES = this.#resilientStart + ? NOT_FOUND_RETRY_DELAYS.length + : 0; // Prefer the World's long poll: one read that the backend holds open // until the run finishes, instead of asking again every second and @@ -531,7 +667,7 @@ export class Run { WorkflowRunNotFoundError.is(error) && notFoundRetries < NOT_FOUND_MAX_RETRIES ) { - const delay = NOT_FOUND_DELAYS[notFoundRetries]!; + const delay = NOT_FOUND_RETRY_DELAYS[notFoundRetries]!; notFoundRetries++; await new Promise((resolve) => setTimeout(resolve, delay)); continue; diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index 542d672f63..8957491cfd 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -2769,8 +2769,10 @@ export function getCommonRevivers(global: Record = globalThis) { * Tiers 2 and 3 import the key encrypt-only, which is an honor-system * restriction: the same bytes could decrypt. Tier 1 makes it a cryptographic * guarantee: a public key cannot read anything. + * + * @internal */ -async function getForwardedWritableEncryptionKey( +export async function getForwardedWritableEncryptionKey( runId: string, deploymentId: string | undefined, encryptionPublicKey: string | undefined @@ -2787,6 +2789,97 @@ async function getForwardedWritableEncryptionKey( return rawKey ? await importKey(rawKey, ['encrypt']) : undefined; } +/** + * Open a writable against a run-scoped server stream and tag the handle so it + * can be forwarded again without losing the target. + * + * Two callers reach the same shape from opposite directions: the external + * reviver, hydrating a descriptor another run put on the wire, and + * `Run#getWritable()`, building the first handle from owner metadata. Both need + * the serialize transform, the group-commit sink, the flushable wiring, and the + * four forwarding symbols in exactly the same arrangement, and a drift between + * them shows up as an unreadable frame rather than a type error. + * + * `runId`/`name` address the *owner's* stream, and `key` must already be the + * write-only key for that owner (see {@link getForwardedWritableEncryptionKey}), + * not the calling run's. `encryptionPublicKey` is stamped only when the owner + * published one: advertising a key the owner cannot open would send every + * downstream contributor to an address nobody reads. + * + * @internal + */ +export function createForwardedWritable({ + global, + ops, + runId, + name, + key, + deploymentId, + encryptionPublicKey, + runReadyBarrier, +}: { + global: Record; + ops: Promise[]; + runId: string; + name: string; + key: EncryptionKeyParam; + deploymentId?: string; + encryptionPublicKey?: string; + runReadyBarrier?: Promise; +}): WritableStream { + const serialize = getSerializeStream( + getExternalReducers(global, ops, runId, key), + key + ); + const serverWritable = new WorkflowServerWritableStream( + runId, + name, + runReadyBarrier + ); + + // Flushable rather than a bare pipeTo: the ops promise has to settle when the + // caller releases its writer lock, not only when the stream is closed. A + // contributor to a shared stream never closes it, so a pipeTo here would keep + // the function alive until its timeout. + const state = createFlushableState(); + ops.push(state.promise); + + flushablePipe(serialize.readable, serverWritable, state).catch(() => { + // Errors are handled via state.reject + }); + + pollWritableLock(serialize.writable, state); + + Object.defineProperty(serialize.writable, STREAM_NAME_SYMBOL, { + value: name, + writable: false, + }); + Object.defineProperty(serialize.writable, STREAM_SERVER_RUN_ID_SYMBOL, { + value: runId, + writable: false, + }); + if (typeof deploymentId === 'string') { + Object.defineProperty( + serialize.writable, + STREAM_SERVER_DEPLOYMENT_ID_SYMBOL, + { + value: deploymentId, + writable: false, + } + ); + } + // Keep the owner's public key on the handle so a further forward stays on + // the zero-lookup sealed path. + if (typeof encryptionPublicKey === 'string') { + Object.defineProperty(serialize.writable, STREAM_SERVER_PUBLIC_KEY_SYMBOL, { + value: encryptionPublicKey, + writable: false, + }); + } + + return serialize.writable as WritableStream; +} + /** * Create a run's object readable without dispatching its stream GET or * encryption-key lookup until the caller reads it. The external reviver starts @@ -3014,59 +3107,15 @@ export function getExternalRevivers( value.encryptionPublicKey ); - const serialize = getSerializeStream( - getExternalReducers(global, ops, targetRunId, targetKey), - targetKey - ); - const serverWritable = new WorkflowServerWritableStream( - targetRunId, - value.name - ); - - // Create flushable state for this stream - const state = createFlushableState(); - ops.push(state.promise); - - // Start the flushable pipe in the background - flushablePipe(serialize.readable, serverWritable, state).catch(() => { - // Errors are handled via state.reject - }); - - // Start polling to detect when user releases lock - pollWritableLock(serialize.writable, state); - - Object.defineProperty(serialize.writable, STREAM_NAME_SYMBOL, { - value: value.name, - writable: false, - }); - Object.defineProperty(serialize.writable, STREAM_SERVER_RUN_ID_SYMBOL, { - value: targetRunId, - writable: false, + return createForwardedWritable({ + global, + ops, + runId: targetRunId, + name: value.name, + key: targetKey, + deploymentId: value.deploymentId, + encryptionPublicKey: value.encryptionPublicKey, }); - if (typeof value.deploymentId === 'string') { - Object.defineProperty( - serialize.writable, - STREAM_SERVER_DEPLOYMENT_ID_SYMBOL, - { - value: value.deploymentId, - writable: false, - } - ); - } - // Keep the owner's public key on the handle so a further forward stays on - // the zero-lookup sealed path. - if (typeof value.encryptionPublicKey === 'string') { - Object.defineProperty( - serialize.writable, - STREAM_SERVER_PUBLIC_KEY_SYMBOL, - { - value: value.encryptionPublicKey, - writable: false, - } - ); - } - - return serialize.writable; }, AbortController: (value) => reviveAbortController(value, ops, runId), diff --git a/packages/workflow/src/api-workflow.ts b/packages/workflow/src/api-workflow.ts index a489fd929d..c8ec15a9e6 100644 --- a/packages/workflow/src/api-workflow.ts +++ b/packages/workflow/src/api-workflow.ts @@ -6,6 +6,7 @@ export type { StopSleepResult, WorkflowReadableStreamOptions, WorkflowRun, + WorkflowRunWritableStreamOptions, } from '@workflow/core/runtime'; export { Run } from '@workflow/core/runtime/run'; diff --git a/packages/workflow/src/api.ts b/packages/workflow/src/api.ts index 31ca6ed146..f16eedadce 100644 --- a/packages/workflow/src/api.ts +++ b/packages/workflow/src/api.ts @@ -25,6 +25,7 @@ export { Run, type WorkflowReadableStream, type WorkflowReadableStreamOptions, + type WorkflowRunWritableStreamOptions, } from '@workflow/core/runtime/run'; export { type StartOptions, From 7982ef0ac6cc57ea45b37d5796a9f943cfad33d7 Mon Sep 17 00:00:00 2001 From: Alex Langenfeld Date: Fri, 4 Sep 2026 14:15:34 -0500 Subject: [PATCH 2/3] Sort test imports to satisfy biome check Signed-off-by: Alex Langenfeld --- packages/core/src/runtime/run-get-writable.test.ts | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/core/src/runtime/run-get-writable.test.ts b/packages/core/src/runtime/run-get-writable.test.ts index 0ae36ba0d9..eb3b9f283f 100644 --- a/packages/core/src/runtime/run-get-writable.test.ts +++ b/packages/core/src/runtime/run-get-writable.test.ts @@ -14,16 +14,16 @@ import { WorkflowRunNotFoundError } from '@workflow/errors'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { importKey } from '../encryption.js'; import { bytesToBase64, deriveRunKeyPair } from '../sealed-box.js'; +import { + decrypt as decryptEnvelope, + runPayloadKeys, +} from '../serialization/encryption.js'; import { dehydrateStepArguments, dehydrateWorkflowArguments, hydrateStepArguments, hydrateWorkflowArguments, } from '../serialization.js'; -import { - decrypt as decryptEnvelope, - runPayloadKeys, -} from '../serialization/encryption.js'; import { hydrateData } from '../serialization-format.js'; import { STREAM_NAME_SYMBOL, From 716fce7dc795f50974984d8c533b0b466cf00ba9 Mon Sep 17 00:00:00 2001 From: Alex Langenfeld Date: Fri, 4 Sep 2026 14:37:26 -0500 Subject: [PATCH 3/3] docs: tighten run writable guidance Signed-off-by: Alex Langenfeld --- .changeset/run-get-writable.md | 2 +- .../v4/api-reference/workflow-api/get-run.mdx | 31 +-------- .../content/docs/v4/foundations/streaming.mdx | 68 ++----------------- .../v5/api-reference/workflow-api/get-run.mdx | 31 +-------- .../content/docs/v5/foundations/streaming.mdx | 68 ++----------------- .../core/src/runtime/run-get-writable.test.ts | 44 ------------ packages/core/src/runtime/run.ts | 63 +++-------------- packages/core/src/serialization.ts | 44 ++---------- 8 files changed, 24 insertions(+), 327 deletions(-) diff --git a/.changeset/run-get-writable.md b/.changeset/run-get-writable.md index ee2b7735d2..bb7e768ba9 100644 --- a/.changeset/run-get-writable.md +++ b/.changeset/run-get-writable.md @@ -3,4 +3,4 @@ 'workflow': minor --- -Add `Run#getWritable()`, which opens a writable onto an existing run's stream from its run ID so one run can append to a stream another run owns. +Add `Run#getWritable()` to append to an existing run's stream. diff --git a/docs/content/docs/v4/api-reference/workflow-api/get-run.mdx b/docs/content/docs/v4/api-reference/workflow-api/get-run.mdx index 60e81eabe7..484c4fdd93 100644 --- a/docs/content/docs/v4/api-reference/workflow-api/get-run.mdx +++ b/docs/content/docs/v4/api-reference/workflow-api/get-run.mdx @@ -69,17 +69,13 @@ export default WorkflowReadableStreamOptions;`} #### WorkflowRunWritableStreamOptions -`run.getWritable()` returns a `WritableStream` that appends to the run's stream, letting one run contribute to a stream another run owns. See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream). - - -`getWritable()` grants append access, not read access: where the owning run publishes an encryption public key, writes are sealed to it and cannot be read back by the writer. Closing the returned writable closes the **shared** stream for every writer, so contributors should `releaseLock()` instead. The run must already exist; an unknown run rejects with `WorkflowRunNotFoundError`. - +See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream) for usage and lifecycle details. #### StopSleepOptions @@ -186,31 +182,6 @@ const { stoppedCount } = await run.wakeUp({ }); ``` -### Write to a stream another run owns - -A holder run owns a session's stream for its whole lifetime, and short-lived -runs append to it knowing only its ID: - -```typescript lineNumbers -import { getRun } from "workflow/api"; - -type SessionEvent = { turn: number; text: string }; - -async function contributeToSession(holderRunId: string, turn: number) { - "use step"; - - const writable = await getRun(holderRunId).getWritable(); // [!code highlight] - const writer = writable.getWriter(); - - await writer.write({ turn, text: "done" }); - - // Release rather than close: the stream belongs to the holder run. // [!code highlight] - writer.releaseLock(); // [!code highlight] -} -``` - -Pass `{ namespace: "name" }` to target one of the owning run's namespaced streams. See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream) for the full pattern and its caveats. - ## Related functions - [`start()`](/docs/api-reference/workflow-api/start): Start a new workflow and get its run ID. diff --git a/docs/content/docs/v4/foundations/streaming.mdx b/docs/content/docs/v4/foundations/streaming.mdx index 65a9d3816b..851750f66e 100644 --- a/docs/content/docs/v4/foundations/streaming.mdx +++ b/docs/content/docs/v4/foundations/streaming.mdx @@ -315,91 +315,31 @@ export async function POST(request: Request) { ## Writing to another run's stream -`getWritable()` gives a run a writable onto its own stream. `getRun(runId).getWritable()` gives you a writable onto a *different* run's stream, reconstructed from that run's ID alone. - -This exists for the holder pattern: one long-lived run owns a session's stream, and short-lived runs contribute to it. Clients read one stream for the whole session, while the work that fills it is split across runs that start and finish independently. Because the writable is derived from the run ID, the holder never has to hand a handle to each contributor. - -```typescript title="workflows/session-holder.ts" lineNumbers -import { sleep } from "workflow"; - -// This run exists to own the session's stream. Its ID is the session's -// identity: clients read from it, and every turn writes to it. -export async function sessionHolder(sessionId: string) { - "use workflow"; - - await sleep("24h"); - await archiveSession(sessionId); -} - -async function archiveSession(sessionId: string) { - "use step"; - // Clean up when the session's lifetime is over. -} -``` - -Each turn is its own workflow. It knows only the holder's run ID: +`getRun(runId).getWritable()` appends to a stream owned by another run. This lets short-lived runs contribute to a long-lived holder run's stream using only its ID. ```typescript title="workflows/turn.ts" lineNumbers import { getRun } from "workflow/api"; type SessionEvent = { turn: number; text: string }; -export async function turnWorkflow(holderRunId: string, turn: number) { - "use workflow"; - - await runTurn(holderRunId, turn); -} - async function runTurn(holderRunId: string, turn: number) { "use step"; const writable = await getRun(holderRunId).getWritable(); // [!code highlight] const writer = writable.getWriter(); - await writer.write({ turn, text: "thinking" }); await writer.write({ turn, text: "done" }); - - // Release, don't close: the stream outlives this turn. // [!code highlight] writer.releaseLock(); // [!code highlight] } ``` -The client reads the holder's stream once and sees every turn: - -```typescript title="app/api/session/[id]/route.ts" lineNumbers -import { getRun } from "workflow/api"; - -type SessionEvent = { turn: number; text: string }; - -export async function GET( - request: Request, - { params }: { params: Promise<{ id: string }> } -) { - const { id } = await params; - const readable = getRun(id).getReadable(); - - return new Response(readable, { - headers: { "Content-Type": "text/event-stream" } - }); -} -``` - -Pass `{ namespace: 'name' }` to target one of the holder's [namespaced streams](#namespaced-streams), exactly as with `getReadable()`. - -### What this does and does not grant +Pass `{ namespace: "name" }` to target a [namespaced stream](#namespaced-streams). The writable can also be forwarded through `start()` and into steps. -**`writer.close()` closes the shared stream.** - -The handle is an ordinary `WritableStream`, so closing it ends the stream for *every* writer, and a closed stream cannot be reopened. Contributors should `releaseLock()` when they are done. Releasing the lock flushes pending writes, so there is never a need to close a stream just to get writes out. Leave closing to the run that owns the stream. +Contributors should call `releaseLock()`, which flushes pending writes. Calling `close()` closes the shared stream for every writer. -- **Append access, not read access.** Where the owning run publishes an [encryption](/docs/how-it-works/encryption) public key, contributions are sealed to it. The writing run cannot decrypt the stream, including its own writes. Read the stream with `getReadable()` from somewhere that holds the owning run's key. -- **No authorization of its own.** Reaching the run through your World is the authorization. This inherits whatever access the caller already had, so treat a run ID reaching untrusted code the same way you would treat any other capability. -- **The owner keeps the lifecycle.** The stream's region, retention, and terminal state follow the owning run. When that run's stream is closed or expires, contributors' writes fail; nothing about holding a writable extends it. -- **The run must already exist.** `getWritable()` rejects with `WorkflowRunNotFoundError` for an unknown run rather than creating one. When a holder and its first turn start together, start the holder first, or handle that rejection. - -The returned writable can be passed to `start()` and forwarded into steps like any other stream, and it keeps pointing at the owning run's stream wherever it travels. +The API grants append access, not read access or additional authorization. The owning run controls the stream's lifecycle, and an unknown run rejects with `WorkflowRunNotFoundError`. ## Common patterns diff --git a/docs/content/docs/v5/api-reference/workflow-api/get-run.mdx b/docs/content/docs/v5/api-reference/workflow-api/get-run.mdx index 8e32099baf..808b1290c6 100644 --- a/docs/content/docs/v5/api-reference/workflow-api/get-run.mdx +++ b/docs/content/docs/v5/api-reference/workflow-api/get-run.mdx @@ -67,17 +67,13 @@ export default WorkflowReadableStreamOptions;`} #### WorkflowRunWritableStreamOptions -`run.getWritable()` returns a `WritableStream` that appends to the run's stream, letting one run contribute to a stream another run owns. See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream). - - -`getWritable()` grants append access, not read access: where the owning run publishes an encryption public key, writes are sealed to it and cannot be read back by the writer. Closing the returned writable closes the **shared** stream for every writer, so contributors should `releaseLock()` instead. The run must already exist; an unknown run rejects with `WorkflowRunNotFoundError`. - +See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream) for usage and lifecycle details. #### StopSleepOptions @@ -203,31 +199,6 @@ export async function POST(req: Request) { The options object is optional: `await run.cancel()` cancels the run without recording a reason. -### Write to a stream another run owns - -A holder run owns a session's stream for its whole lifetime, and short-lived -runs append to it knowing only its ID: - -```typescript lineNumbers -import { getRun } from "workflow/api"; - -type SessionEvent = { turn: number; text: string }; - -async function contributeToSession(holderRunId: string, turn: number) { - "use step"; - - const writable = await getRun(holderRunId).getWritable(); // [!code highlight] - const writer = writable.getWriter(); - - await writer.write({ turn, text: "done" }); - - // Release rather than close: the stream belongs to the holder run. // [!code highlight] - writer.releaseLock(); // [!code highlight] -} -``` - -Pass `{ namespace: "name" }` to target one of the owning run's namespaced streams. See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream) for the full pattern and its caveats. - ## Related functions - [`start()`](/docs/api-reference/workflow-api/start): Start a new workflow and get its run ID. diff --git a/docs/content/docs/v5/foundations/streaming.mdx b/docs/content/docs/v5/foundations/streaming.mdx index 79ac5caea2..cbcd4209a2 100644 --- a/docs/content/docs/v5/foundations/streaming.mdx +++ b/docs/content/docs/v5/foundations/streaming.mdx @@ -316,91 +316,31 @@ export async function POST(request: Request) { ## Writing to another run's stream -`getWritable()` gives a run a writable onto its own stream. `getRun(runId).getWritable()` gives you a writable onto a *different* run's stream, reconstructed from that run's ID alone. - -This exists for the holder pattern: one long-lived run owns a session's stream, and short-lived runs contribute to it. Clients read one stream for the whole session, while the work that fills it is split across runs that start and finish independently. Because the writable is derived from the run ID, the holder never has to hand a handle to each contributor. - -```typescript title="workflows/session-holder.ts" lineNumbers -import { sleep } from "workflow"; - -// This run exists to own the session's stream. Its ID is the session's -// identity: clients read from it, and every turn writes to it. -export async function sessionHolder(sessionId: string) { - "use workflow"; - - await sleep("24h"); - await archiveSession(sessionId); -} - -async function archiveSession(sessionId: string) { - "use step"; - // Clean up when the session's lifetime is over. -} -``` - -Each turn is its own workflow. It knows only the holder's run ID: +`getRun(runId).getWritable()` appends to a stream owned by another run. This lets short-lived runs contribute to a long-lived holder run's stream using only its ID. ```typescript title="workflows/turn.ts" lineNumbers import { getRun } from "workflow/api"; type SessionEvent = { turn: number; text: string }; -export async function turnWorkflow(holderRunId: string, turn: number) { - "use workflow"; - - await runTurn(holderRunId, turn); -} - async function runTurn(holderRunId: string, turn: number) { "use step"; const writable = await getRun(holderRunId).getWritable(); // [!code highlight] const writer = writable.getWriter(); - await writer.write({ turn, text: "thinking" }); await writer.write({ turn, text: "done" }); - - // Release, don't close: the stream outlives this turn. // [!code highlight] writer.releaseLock(); // [!code highlight] } ``` -The client reads the holder's stream once and sees every turn: - -```typescript title="app/api/session/[id]/route.ts" lineNumbers -import { getRun } from "workflow/api"; - -type SessionEvent = { turn: number; text: string }; - -export async function GET( - request: Request, - { params }: { params: Promise<{ id: string }> } -) { - const { id } = await params; - const readable = getRun(id).getReadable(); - - return new Response(readable, { - headers: { "Content-Type": "text/event-stream" } - }); -} -``` - -Pass `{ namespace: 'name' }` to target one of the holder's [namespaced streams](#namespaced-streams), exactly as with `getReadable()`. - -### What this does and does not grant +Pass `{ namespace: "name" }` to target a [namespaced stream](#namespaced-streams). The writable can also be forwarded through `start()` and into steps. -**`writer.close()` closes the shared stream.** - -The handle is an ordinary `WritableStream`, so closing it ends the stream for *every* writer, and a closed stream cannot be reopened. Contributors should `releaseLock()` when they are done. Releasing the lock flushes pending writes, so there is never a need to close a stream just to get writes out. Leave closing to the run that owns the stream. +Contributors should call `releaseLock()`, which flushes pending writes. Calling `close()` closes the shared stream for every writer. -- **Append access, not read access.** Where the owning run publishes an [encryption](/docs/how-it-works/encryption) public key, contributions are sealed to it. The writing run cannot decrypt the stream, including its own writes. Read the stream with `getReadable()` from somewhere that holds the owning run's key. -- **No authorization of its own.** Reaching the run through your World is the authorization. This inherits whatever access the caller already had, so treat a run ID reaching untrusted code the same way you would treat any other capability. -- **The owner keeps the lifecycle.** The stream's region, retention, and terminal state follow the owning run. When that run's stream is closed or expires, contributors' writes fail; nothing about holding a writable extends it. -- **The run must already exist.** `getWritable()` rejects with `WorkflowRunNotFoundError` for an unknown run rather than creating one. When a holder and its first turn start together, start the holder first, or handle that rejection. - -The returned writable can be passed to `start()` and forwarded into steps like any other stream, and it keeps pointing at the owning run's stream wherever it travels. +The API grants append access, not read access or additional authorization. The owning run controls the stream's lifecycle, and an unknown run rejects with `WorkflowRunNotFoundError`. ## Common patterns diff --git a/packages/core/src/runtime/run-get-writable.test.ts b/packages/core/src/runtime/run-get-writable.test.ts index eb3b9f283f..371c9b9cec 100644 --- a/packages/core/src/runtime/run-get-writable.test.ts +++ b/packages/core/src/runtime/run-get-writable.test.ts @@ -1,15 +1,3 @@ -/** - * `Run#getWritable()` — append access to a stream another run owns, - * reconstructed from the run ID alone. - * - * What these tests pin down is that the handle it returns is indistinguishable - * from one the owner handed over itself: same `(runId, name)` target, same - * forwarding symbols, same sealed-frame encoding. The two are produced by - * different code paths (owner metadata here, a wire descriptor in the - * reviver), and a divergence between them surfaces as a frame the owner cannot - * decrypt rather than as a type error, so the encoding is asserted at the byte - * level rather than by trusting the shape. - */ import { WorkflowRunNotFoundError } from '@workflow/errors'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { importKey } from '../encryption.js'; @@ -41,11 +29,6 @@ const OWNER_RUN_ID = 'wrun_ownerrun'; const OWNER_STREAM = 'strm_ownerrun_user'; const OWNER_MATERIAL = new Uint8Array(32).fill(0x2b); -/** - * The subset of a run record `getWritable()` reads. `resolveData: 'none'` - * keeps `deploymentId` and `encryptionPublicKey`, which is the whole reason it - * can skip resolving payload refs. - */ function ownerRun(overrides: Record = {}) { return { runId: OWNER_RUN_ID, @@ -69,7 +52,6 @@ function mockWorld(overrides: Record = {}) { return world; } -/** Frames the sink actually received, in order. */ function framesWritten(world: any): Uint8Array[] { return world.streams.write.mock.calls .map((c: any[]) => c[2]) @@ -122,8 +104,6 @@ describe('Run#getWritable', () => { }); it('rejects for a run that does not exist', async () => { - // The API reconstructs a handle onto an existing stream; it must never - // bring a run or a stream into being as a side effect of asking for one. const world = mockWorld({ runs: { get: vi @@ -136,14 +116,10 @@ describe('Run#getWritable', () => { WorkflowRunNotFoundError ); expect(world.streams.write).not.toHaveBeenCalled(); - // Fail fast: one look, no backoff, for an ID the caller supplied. expect(world.runs.get).toHaveBeenCalledTimes(1); }); it('seals to the owner public key without resolving a symmetric key', async () => { - // The owner published its X25519 public key, so the contributor can seal - // with no key-API round trip — and, more importantly, without ever holding - // a key that could read the stream back. const ownerKeyPair = await deriveRunKeyPair(OWNER_MATERIAL); const getEncryptionKeyForRun = vi.fn(); const world = mockWorld({ @@ -171,7 +147,6 @@ describe('Run#getWritable', () => { // [4-byte length][encp][sealed...] expect(new TextDecoder().decode(frames[0].subarray(4, 8))).toBe('encp'); - // And the owner really can open it. const ownerKeys = runPayloadKeys( await importKey(OWNER_MATERIAL), ownerKeyPair @@ -181,8 +156,6 @@ describe('Run#getWritable', () => { }); it('falls back to the owner deployment key when it published no public key', async () => { - // Runs created by older SDKs carry no public key. They still have to be - // writable, via the symmetric key imported encrypt-only. const getEncryptionKeyForRun = vi.fn().mockResolvedValue(OWNER_MATERIAL); const world = mockWorld({ getEncryptionKeyForRun }); @@ -193,8 +166,6 @@ describe('Run#getWritable', () => { writer.releaseLock(); await Promise.all(ops); - // Resolved from the deployment on the run record, so the tier that has to - // load the owning run first never runs. expect(getEncryptionKeyForRun).toHaveBeenCalledWith(OWNER_RUN_ID, { deploymentId: 'dpl_owner', }); @@ -228,8 +199,6 @@ describe('Run#getWritable', () => { }); it('does not advertise a public key the owner never published', async () => { - // Stamping one here would send later contributors to seal to an address - // the owner cannot open. mockWorld({ getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined) }); const writable = await getRun(OWNER_RUN_ID).getWritable(); @@ -238,10 +207,6 @@ describe('Run#getWritable', () => { }); it('keeps the owner identity through start() and into a step', async () => { - // The motivating shape: a turn workflow holding only the holder run's ID - // opens a writable, hands it to the workflow it starts, and that workflow - // hands it to the step that writes. Two serialization hops, and the owner - // must survive both or the writes land on the wrong stream. const ownerKeyPair = await deriveRunKeyPair(OWNER_MATERIAL); const ownerPublicKey = bytesToBase64(ownerKeyPair.publicKey); const world = mockWorld({ @@ -293,7 +258,6 @@ describe('Run#getWritable', () => { writer.releaseLock(); await Promise.all(ops); - // Still the owner's stream, still sealed, still no key lookup. expect(world.getEncryptionKeyForRun).not.toHaveBeenCalled(); const ownerFrames = world.streams.write.mock.calls.filter( (c: any[]) => c[0] === OWNER_RUN_ID && c[1] === OWNER_STREAM @@ -310,9 +274,6 @@ describe('Run#getWritable', () => { }); it('drains on lock release without closing the shared stream', async () => { - // The contract that makes this safe for per-turn contributors: releasing - // the writer settles the flush, so nobody has to close a stream they do - // not own just to get their writes out. const world = mockWorld(); const ops: Promise[] = []; @@ -328,8 +289,6 @@ describe('Run#getWritable', () => { }); it('closes the owner stream when the caller closes the handle', async () => { - // Documented consequence rather than a feature: this handle is a normal - // WritableStream, so close() ends the shared stream for every writer. const world = mockWorld(); const ops: Promise[] = []; @@ -346,9 +305,6 @@ describe('Run#getWritable', () => { }); it('retries a missing run only for a resiliently started run', async () => { - // A Run handed back by an optimistic start() may legitimately not exist - // yet. Same bounded budget the return-value poll uses; no open-ended - // polling, and getRun() never opts into it. vi.useFakeTimers(); const runsGet = vi .fn() diff --git a/packages/core/src/runtime/run.ts b/packages/core/src/runtime/run.ts index 3f5068b2b7..2eddef3b98 100644 --- a/packages/core/src/runtime/run.ts +++ b/packages/core/src/runtime/run.ts @@ -34,16 +34,7 @@ const PAYLOAD_TERMINAL_RUN_STATUSES = new Set([ 'failed', ]); -/** - * Backoff for the "run does not exist yet" retry, used only by runs whose - * `run_created` event failed and that the resilient start path will create via - * `run_started`. 1s + 3s + 6s gives the queue 10s to deliver before the caller - * sees the 404. - * - * A `Run` from `getRun()` never uses this: for an ID a caller supplied, a - * missing run is a real error and reporting it immediately beats a ten-second - * pause on the way to the same answer. - */ +// Give resilient starts up to 10 seconds to create the run. const NOT_FOUND_RETRY_DELAYS = [1_000, 3_000, 6_000]; /** @internal */ @@ -436,41 +427,16 @@ export class Run { } /** - * Retrieves a writable onto this run's stream, the one - * {@link Run.getReadable | getReadable()} reads and the run's own - * `getWritable()` writes. - * - * This is append access to a stream another run owns, reconstructed from the - * run ID alone. It exists for the holder pattern: a long-lived run owns a - * session's stream, and short-lived runs that know only its ID contribute to - * it without the owner having to hand a writable to each one. The handle can - * be forwarded onward through `start()` and into steps like any other. - * - * The run must already exist. A missing run rejects with - * {@link WorkflowRunNotFoundError} rather than creating anything, except on a - * `Run` from a resilient `start()`, which retries briefly while the run is - * still being created. - * - * Two things this deliberately does not give you: + * Returns a writable that appends to this run's stream. * - * - **No read access.** Frames are sealed to the owner's public key where it - * has one, so the writer cannot decrypt this stream, including its own - * writes. Read it with `getReadable()` from somewhere holding the run's key. - * - **No authorization of its own.** Reaching the run through the World is - * the authorization; this inherits whatever the caller already had. - * - * The owner keeps the stream's lifecycle: its region, retention, and terminal - * state follow the owning run, not the caller. + * The run must already exist. The writable may be forwarded through `start()` + * and into steps, but grants no read access or additional authorization. * * @remarks - * `writer.close()` closes the **shared** stream for everyone, not just this - * handle, and a closed stream cannot be reopened. A contributor should write - * and then `releaseLock()`; releasing the lock drains pending writes, so - * closing is never needed merely to flush. Leave closing to the run that owns - * the stream. + * `writer.close()` closes the shared stream. Contributors should call + * `releaseLock()`, which also drains pending writes. * - * @param options - The options for the writable stream. - * @returns A `WritableStream` targeting this run's stream. + * @param options - The writable stream options. * @throws WorkflowRunNotFoundError if the run does not exist. */ async getWritable( @@ -481,10 +447,7 @@ export class Run { const run = await this.#getMetadata(); const name = getWorkflowRunStreamId(this.runId, namespace); - // Resolve before returning rather than handing the promise to the - // serializer: the sealed path costs no I/O, and on the fallback path a key - // that cannot be resolved is better reported here than on a write the - // caller has already been told was accepted. + // Resolve before returning so key lookup failures precede accepted writes. const key = await getForwardedWritableEncryptionKey( this.runId, run.deploymentId, @@ -502,15 +465,7 @@ export class Run { }); } - /** - * Reads this run's metadata, absorbing the window in which a resiliently - * started run does not exist yet. - * - * `resolveData: 'none'` is deliberate: `deploymentId` and - * `encryptionPublicKey` both survive it, and resolving input/output payload - * refs to reach them would be work thrown away. - * @internal - */ + /** Reads metadata, briefly retrying resilient starts. @internal */ async #getMetadata() { const world = await this.#lazyWorldPromise; const maxRetries = this.#resilientStart ? NOT_FOUND_RETRY_DELAYS.length : 0; diff --git a/packages/core/src/serialization.ts b/packages/core/src/serialization.ts index 8957491cfd..828fe4ce1c 100644 --- a/packages/core/src/serialization.ts +++ b/packages/core/src/serialization.ts @@ -2753,23 +2753,8 @@ export function getCommonRevivers(global: Record = globalThis) { } /** - * Resolve the write-only key a run needs when writing into another run's - * forwarded stream. - * - * Three tiers, cheapest first: - * - * 1. The descriptor carries the owner's X25519 public key: seal to it with - * no I/O whatsoever. The owner published the key when it created the - * stream, so this is the zero-round-trip path. - * 2. The descriptor carries the owner's deployment ID: resolve the owner's - * symmetric key, which cross-deployment means a key-API round trip. - * 3. Neither (descriptors written by older SDKs): load the owning run first, - * then resolve its symmetric key. - * - * Tiers 2 and 3 import the key encrypt-only, which is an honor-system - * restriction: the same bytes could decrypt. Tier 1 makes it a cryptographic - * guarantee: a public key cannot read anything. - * + * Resolves the owner's encryption key, preferring its public key, then its + * deployment, and finally its run record. * @internal */ export async function getForwardedWritableEncryptionKey( @@ -2789,25 +2774,7 @@ export async function getForwardedWritableEncryptionKey( return rawKey ? await importKey(rawKey, ['encrypt']) : undefined; } -/** - * Open a writable against a run-scoped server stream and tag the handle so it - * can be forwarded again without losing the target. - * - * Two callers reach the same shape from opposite directions: the external - * reviver, hydrating a descriptor another run put on the wire, and - * `Run#getWritable()`, building the first handle from owner metadata. Both need - * the serialize transform, the group-commit sink, the flushable wiring, and the - * four forwarding symbols in exactly the same arrangement, and a drift between - * them shows up as an unreadable frame rather than a type error. - * - * `runId`/`name` address the *owner's* stream, and `key` must already be the - * write-only key for that owner (see {@link getForwardedWritableEncryptionKey}), - * not the calling run's. `encryptionPublicKey` is stamped only when the owner - * published one: advertising a key the owner cannot open would send every - * downstream contributor to an address nobody reads. - * - * @internal - */ +/** Creates and tags a forwarded writable targeting the owner run. @internal */ export function createForwardedWritable({ global, ops, @@ -2837,10 +2804,7 @@ export function createForwardedWritable({ runReadyBarrier ); - // Flushable rather than a bare pipeTo: the ops promise has to settle when the - // caller releases its writer lock, not only when the stream is closed. A - // contributor to a shared stream never closes it, so a pipeTo here would keep - // the function alive until its timeout. + // Lock release must settle the flush promise; contributors do not close. const state = createFlushableState(); ops.push(state.promise);