🧭 Add durable Workspace coordination seams

architecture.md

@@ -380,6 +380,26 @@ historical execution that reached it. Replay never asks current state to prove a
 past effect: a file written and later deleted is absent at the frontier, while
 both completed effects still restore in order.
 
+Structured durable operations accept an explicit provider-neutral live
+coordinator. The default coordinator executes once, converts execution success
+or failure to the existing durable protocol `Result`, calls the existing Yield
+publication continuation once, and returns that same result only after
+publication completes. The continuation publishes through the durable stream's
+ordered append fence. A backing append failure activates fail-stop state and
+raises `DurablePersistenceError` with the adapter error as its cause; it writes
+no compensating `Close`, and catching it cannot admit later durable work. A
+marked pre-persistence policy rejection remains an ordinary policy failure.
+There is no generic durable validation hook: validation owned by a caller or
+provider occurs before it constructs the durable effect. Replay, including the
+replayed prefix of a partial run, bypasses the coordinator, execution and live
+publication. Callback-based durable effects keep their existing path.
+
+The shared Workspace durable-operation wrapper selects a contextual Workspace
+coordinator explicitly. Its default fails before execution or publication, so
+installing no provider cannot leak a mutation outside its transaction. Selecting
+Workspace coordination for one operation does not enlist unrelated durable
+operations in the same scope.
+
 Every Workspace-local expansion publishes one effect through one effect
 transaction:
 
@@ -472,8 +492,22 @@ after SQLite has restored the prior frontier.
 Retained roots, manifests and blobs remain indefinitely. Cloudflare garbage
 collection is not in the production closure and is never invoked. The provider
 exposes no public Workspace mutation effect, history selection or fork
-operation at this layer. Provider-neutral durable coordination, filtered
-journal routing, and atomic Workspace effect publication are also absent.
+operation at this layer. Provider-neutral durable coordination and explicit
+filtered-journal routing are present, while the Deno Workspace coordinator that
+combines mutation, root publication and journal publication atomically is
+absent.
+
+The Deno journal adapter routes an append ordinarily when no destination is
+bound. A publication may instead bind one exact transaction destination for its
+own lexical scope after the existing secret gate. The route validates the
+database lease, connection generation, transaction identity, token and open
+state before delegating to the existing `transaction.journal`; it contains no
+insertion SQL of its own. A nested route for another run delegates past itself,
+and `readAll()` always follows ordinary replay without routing or secret
+filtering. The provider-owned ordinary destination and each publication-local
+routed destination are terminal `{ at: "min" }` handlers, so an enclosing
+loaded copy with the same stable contextual name cannot suppress an append or
+run ahead of exact-token validation.
 
 The initial topology requires neither writable FUSE nor native subprocess
 access and does not bundle `workerd`. A Cloudflare-hosted or workerd-backed
@@ -793,6 +827,9 @@ Status is measured against main.
 | `Git.revParse()` | verifies and resolves one Git revision expression contextually | built on main |
 | workflow run storage | creates or compatibly finds one run by public run ID, retains its identity, state, document executions and filtered journal, and validates immutable Workspace roots through one provider-owned connection entry | built on the #365 stack; Workspace effect publication is unbuilt |
 | caller-owned storage transaction | publishes several changes, including journal events, in one transaction nothing else enlists in | built on main |
+| live durable-operation coordinator | explicitly coordinates structured live execution with existing Yield publication while leaving replay and callback effects unchanged | built on the #365 stack |
+| Workspace coordination API | fails closed by default and lets a Workspace operation explicitly select provider coordination | built on the #365 stack; the atomic Deno Workspace handler is unbuilt |
+| explicit WorkflowRun journal route | binds one already-filtered publication to one exact active transaction and otherwise uses ordinary serialized journal storage | built on the #365 stack |
 | `API.Service` / `startService()` | creates an authenticated, supervised loopback service attachment through a provider-neutral operation | built on main |
 | `service=<binding>` | publishes the attachment's endpoint into the live binding overlay for its invocation | built on main |
 | `ephemeral eval` | reconstructs live middleware and bindings without a journal entry | built on main |

packages/durable-streams/README.md

@@ -286,6 +286,24 @@ gate's ordinary failure and may produce a separately admitted `Close(err)`.
 
 Violating this invariant (advancing the generator before the write) creates an unrecoverable gap: the journal would be missing an entry, and replay would feed the wrong result to a subsequent effect.
 
+### Live operation coordinators
+
+`createDurableOperation` accepts an optional
+`LiveDurableOperationCoordinator`. The default coordinator executes the live
+operation once, converts its success or failure to the existing protocol
+`Result`, invokes the Yield publication continuation once, and returns that
+same result after publication completes. Replay bypasses the coordinator,
+executor, continuation, and live append; a partially replayed run coordinates
+only its live suffix.
+
+The publication continuation uses the ordered append fence described above. A
+backing append failure therefore activates the same fail-stop state and raises
+`DurablePersistenceError` with the adapter error as its cause. A marked
+pre-persistence policy rejection remains an ordinary policy failure. There is no
+generic validation option on durable operations: a caller or provider validates
+before constructing the durable effect. Callback-based durable effects retain
+their existing execution path.
+
 ---
 
 ## Divergence detection

packages/durable-streams/effect.ts

@@ -29,6 +29,10 @@ import {
   rememberDurabilityFailure,
 } from "./durability.ts";
 import { StaleInputError } from "./errors.ts";
+import {
+  defaultLiveDurableOperationCoordinator,
+  type LiveDurableOperationCoordinator,
+} from "./live-coordinator.ts";
 import { ReplayGuard } from "./replay-guard.ts";
 import { protocolToEffection, serializeError } from "./serialize.ts";
 import type {
@@ -306,10 +310,15 @@ export function createDurableEffect<T>(
  *
  * @param desc Structured description for the journal and divergence detection
  * @param execute Returns an Operation to run during live execution
+ * @param options.coordinator Selects the live execution/publication boundary;
+ * replay never invokes it
  */
 export function createDurableOperation<T extends Json>(
   desc: EffectDescription,
   execute: () => Operation<T>,
+  options: {
+    coordinator?: LiveDurableOperationCoordinator;
+  } = {},
 ): DurableEffect<T> {
   return {
     description: `${desc.type}(${desc.name})`,
@@ -340,24 +349,17 @@ export function createDurableOperation<T extends Json>(
           return;
         }
 
-        let result: Result;
         try {
-          const value = yield* execute();
-          result = { status: "ok", value: value as Json };
-        } catch (e) {
-          const error = e instanceof Error ? e : new Error(String(e));
-          result = { status: "err", error: serializeError(error) };
-        }
-
-        const event: Yield = {
-          type: "yield",
-          coroutineId: ctx.coroutineId,
-          description: desc,
-          result,
-        };
-
-        try {
-          yield* appendDurableEvent(ctx, event);
+          const coordinator = options.coordinator ?? defaultLiveDurableOperationCoordinator;
+          const result = yield* coordinator.run(execute, function* (published) {
+            const event: Yield = {
+              type: "yield",
+              coroutineId: ctx.coroutineId,
+              description: desc,
+              result: published,
+            };
+            yield* appendDurableEvent(ctx, event);
+          });
           resolve(protocolToEffection<T>(result));
         } catch (err) {
           resolve({

packages/durable-streams/live-coordinator.ts

@@ -0,0 +1,30 @@
+import type { Operation } from "effection";
+import { serializeError } from "./serialize.ts";
+import type { Json, Result } from "./types.ts";
+
+/** Coordinates one live structured durable operation with its publication. */
+export interface LiveDurableOperationCoordinator {
+  run<T extends Json>(
+    execute: () => Operation<T>,
+    publish: (result: Result) => Operation<void>,
+  ): Operation<Result>;
+}
+
+/** The ordinary live path: execute once, publish once, then return the same result. */
+export const defaultLiveDurableOperationCoordinator: LiveDurableOperationCoordinator = {
+  *run<T extends Json>(
+    execute: () => Operation<T>,
+    publish: (result: Result) => Operation<void>,
+  ): Operation<Result> {
+    let result: Result;
+    try {
+      result = { status: "ok", value: yield* execute() };
+    } catch (error) {
+      const failure = error instanceof Error ? error : new Error(String(error));
+      result = { status: "err", error: serializeError(failure) };
+    }
+
+    yield* publish(result);
+    return result;
+  },
+};

packages/durable-streams/mod.ts

@@ -78,6 +78,10 @@ export { parseDurableEvent } from "./parse.ts";
 export { createDurableEffect, createDurableOperation } from "./effect.ts";
 export type { Executor } from "./effect.ts";
 
+// Structured live-operation coordination
+export { defaultLiveDurableOperationCoordinator } from "./live-coordinator.ts";
+export type { LiveDurableOperationCoordinator } from "./live-coordinator.ts";
+
 // Workflow-enabled effects
 export { durableAction, durableCall, durableSleep, versionCheck } from "./operations.ts";