diff --git a/.changeset/run-get-writable.md b/.changeset/run-get-writable.md
new file mode 100644
index 0000000000..bb7e768ba9
--- /dev/null
+++ b/.changeset/run-get-writable.md
@@ -0,0 +1,6 @@
+---
+'@workflow/core': minor
+'workflow': minor
+---
+
+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 27764ab352..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
@@ -67,6 +67,16 @@ import type { WorkflowReadableStreamOptions } from "workflow/api";
 export default WorkflowReadableStreamOptions;`}
 />
 
+#### WorkflowRunWritableStreamOptions
+
+<TSDoc
+definition={`
+import type { WorkflowRunWritableStreamOptions } from "workflow/api";
+export default WorkflowRunWritableStreamOptions;`}
+/>
+
+See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream) for usage and lifecycle details.
+
 #### StopSleepOptions
 
 <TSDoc
diff --git a/docs/content/docs/v4/foundations/streaming.mdx b/docs/content/docs/v4/foundations/streaming.mdx
index cb1c9b2970..851750f66e 100644
--- a/docs/content/docs/v4/foundations/streaming.mdx
+++ b/docs/content/docs/v4/foundations/streaming.mdx
@@ -313,6 +313,34 @@ export async function POST(request: Request) {
 }
 ```
 
+## Writing to another run's stream
+
+`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 };
+
+async function runTurn(holderRunId: string, turn: number) {
+  "use step";
+
+  const writable = await getRun(holderRunId).getWritable<SessionEvent>(); // [!code highlight]
+  const writer = writable.getWriter();
+
+  await writer.write({ turn, text: "done" });
+  writer.releaseLock(); // [!code highlight]
+}
+```
+
+Pass `{ namespace: "name" }` to target a [namespaced stream](#namespaced-streams). The writable can also be forwarded through `start()` and into steps.
+
+<Callout type="warn">
+Contributors should call `releaseLock()`, which flushes pending writes. Calling `close()` closes the shared stream for every writer.
+</Callout>
+
+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
 
 ### 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..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
@@ -65,6 +65,16 @@ import type { WorkflowReadableStreamOptions } from "workflow/api";
 export default WorkflowReadableStreamOptions;`}
 />
 
+#### WorkflowRunWritableStreamOptions
+
+<TSDoc
+definition={`
+import type { WorkflowRunWritableStreamOptions } from "workflow/api";
+export default WorkflowRunWritableStreamOptions;`}
+/>
+
+See [Writing to another run's stream](/docs/foundations/streaming#writing-to-another-runs-stream) for usage and lifecycle details.
+
 #### StopSleepOptions
 
 <TSDoc
diff --git a/docs/content/docs/v5/foundations/streaming.mdx b/docs/content/docs/v5/foundations/streaming.mdx
index 7e2477a437..cbcd4209a2 100644
--- a/docs/content/docs/v5/foundations/streaming.mdx
+++ b/docs/content/docs/v5/foundations/streaming.mdx
@@ -314,6 +314,34 @@ export async function POST(request: Request) {
 }
 ```
 
+## Writing to another run's stream
+
+`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 };
+
+async function runTurn(holderRunId: string, turn: number) {
+  "use step";
+
+  const writable = await getRun(holderRunId).getWritable<SessionEvent>(); // [!code highlight]
+  const writer = writable.getWriter();
+
+  await writer.write({ turn, text: "done" });
+  writer.releaseLock(); // [!code highlight]
+}
+```
+
+Pass `{ namespace: "name" }` to target a [namespaced stream](#namespaced-streams). The writable can also be forwarded through `start()` and into steps.
+
+<Callout type="warn">
+Contributors should call `releaseLock()`, which flushes pending writes. Calling `close()` closes the shared stream for every writer.
+</Callout>
+
+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
 
 ### 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..371c9b9cec
--- /dev/null
+++ b/packages/core/src/runtime/run-get-writable.test.ts
@@ -0,0 +1,325 @@
+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 { 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);
+
+function ownerRun(overrides: Record<string, unknown> = {}) {
+  return {
+    runId: OWNER_RUN_ID,
+    status: 'running',
+    deploymentId: 'dpl_owner',
+    ...overrides,
+  };
+}
+
+function mockWorld(overrides: Record<string, unknown> = {}) {
+  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;
+}
+
+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<any>[] = [];
+    const writable = await getRun(OWNER_RUN_ID).getWritable<string>({ 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<any>[] = [];
+    const writable = await getRun(OWNER_RUN_ID).getWritable<string>({
+      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 () => {
+    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();
+    expect(world.runs.get).toHaveBeenCalledTimes(1);
+  });
+
+  it('seals to the owner public key without resolving a symmetric key', async () => {
+    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<any>[] = [];
+    const writable = await getRun(OWNER_RUN_ID).getWritable<string>({ 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');
+
+    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 () => {
+    const getEncryptionKeyForRun = vi.fn().mockResolvedValue(OWNER_MATERIAL);
+    const world = mockWorld({ getEncryptionKeyForRun });
+
+    const ops: Promise<any>[] = [];
+    const writable = await getRun(OWNER_RUN_ID).getWritable<string>({ ops });
+    const writer = writable.getWriter();
+    await writer.write('legacy-owner');
+    writer.releaseLock();
+    await Promise.all(ops);
+
+    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 () => {
+    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 () => {
+    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<string>();
+
+    // 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<string>;
+    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<any>[] = [];
+    const inStep = (await hydrateStepArguments(
+      toStep,
+      'wrun_turn',
+      undefined,
+      ops,
+      globalThis,
+      {},
+      'dpl_turn'
+    )) as WritableStream<string>;
+
+    const writer = inStep.getWriter();
+    await writer.write('written-two-hops-away');
+    writer.releaseLock();
+    await Promise.all(ops);
+
+    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 () => {
+    const world = mockWorld();
+
+    const ops: Promise<any>[] = [];
+    const writable = await getRun(OWNER_RUN_ID).getWritable<string>({ 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 () => {
+    const world = mockWorld();
+
+    const ops: Promise<any>[] = [];
+    const writable = await getRun(OWNER_RUN_ID).getWritable<string>({ 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 () => {
+    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..2eddef3b98 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,9 @@ const PAYLOAD_TERMINAL_RUN_STATUSES = new Set<WorkflowRunStatus>([
   'failed',
 ]);
 
+// Give resilient starts up to 10 seconds to create the run.
+const NOT_FOUND_RETRY_DELAYS = [1_000, 3_000, 6_000];
+
 /** @internal */
 export function getReturnValuePollIntervalMs(): number {
   return envNumber(
@@ -154,6 +159,33 @@ export interface WorkflowReadableStreamOptions {
   global?: Record<string, any>;
 }
 
+/**
+ * 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<any>[];
+  /**
+   * 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<string, any>;
+}
+
 /**
  * A handler class for a workflow run.
  */
@@ -394,6 +426,64 @@ export class Run<TResult> {
     });
   }
 
+  /**
+   * Returns a writable that appends to this run's stream.
+   *
+   * 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. Contributors should call
+   * `releaseLock()`, which also drains pending writes.
+   *
+   * @param options - The writable stream options.
+   * @throws WorkflowRunNotFoundError if the run does not exist.
+   */
+  async getWritable<W = any>(
+    options: WorkflowRunWritableStreamOptions = {}
+  ): Promise<WritableStream<W>> {
+    'use step';
+    const { ops = [], global = globalThis, namespace } = options;
+    const run = await this.#getMetadata();
+    const name = getWorkflowRunStreamId(this.runId, namespace);
+
+    // Resolve before returning so key lookup failures precede accepted writes.
+    const key = await getForwardedWritableEncryptionKey(
+      this.runId,
+      run.deploymentId,
+      run.encryptionPublicKey
+    );
+
+    return createForwardedWritable<W>({
+      global,
+      ops,
+      runId: this.runId,
+      name,
+      key,
+      deploymentId: run.deploymentId,
+      encryptionPublicKey: run.encryptionPublicKey,
+    });
+  }
+
+  /** Reads metadata, briefly retrying resilient starts. @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<TResult> {
     if (run.status === 'completed') {
@@ -451,8 +541,9 @@ export class Run<TResult> {
     // 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 +622,7 @@ export class Run<TResult> {
           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..828fe4ce1c 100644
--- a/packages/core/src/serialization.ts
+++ b/packages/core/src/serialization.ts
@@ -2753,24 +2753,11 @@ export function getCommonRevivers(global: Record<string, any> = 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
  */
-async function getForwardedWritableEncryptionKey(
+export async function getForwardedWritableEncryptionKey(
   runId: string,
   deploymentId: string | undefined,
   encryptionPublicKey: string | undefined
@@ -2787,6 +2774,76 @@ async function getForwardedWritableEncryptionKey(
   return rawKey ? await importKey(rawKey, ['encrypt']) : undefined;
 }
 
+/** Creates and tags a forwarded writable targeting the owner run. @internal */
+export function createForwardedWritable<W = any>({
+  global,
+  ops,
+  runId,
+  name,
+  key,
+  deploymentId,
+  encryptionPublicKey,
+  runReadyBarrier,
+}: {
+  global: Record<string, any>;
+  ops: Promise<any>[];
+  runId: string;
+  name: string;
+  key: EncryptionKeyParam;
+  deploymentId?: string;
+  encryptionPublicKey?: string;
+  runReadyBarrier?: Promise<unknown>;
+}): WritableStream<W> {
+  const serialize = getSerializeStream(
+    getExternalReducers(global, ops, runId, key),
+    key
+  );
+  const serverWritable = new WorkflowServerWritableStream(
+    runId,
+    name,
+    runReadyBarrier
+  );
+
+  // Lock release must settle the flush promise; contributors do not close.
+  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<W>;
+}
+
 /**
  * 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 +3071,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,