runtime: drive stores from coalesced scheduling events

docs/architecture.md

@@ -265,21 +265,20 @@ failed through its task/instance ownership channel. Cleanup and terminal
 notification are both attempted, and the first poison cause is retained even
 when cleanup fails.
 
-**Overlapping drivers.** Concurrent exports may run overlapping
-`driveAsync` loops on one store. The invariant is that an activation
-consumes a settlement at most once and never resumes from an obsolete
-settlement. `runtime/src/exec/boundary.ts` enforces this with synchronous
-awaiting-membership removal, memoized Promise tags, Promise-identity
-checks, and per-store pending-resumption bookkeeping.
-
-The asynchronous host-activity and host-settlement pumps are fallback
-drivers: they stand down cooperatively when another driver is active.
-This is not a ban on synchronous pump participation: `HostActivity.pump()`
-services settled activations and ticks ready threads before its async
-fallback checks driver depth, including while an export driver is live.
-Arrival notifications wake parked drivers so they can yield or reconsider their
-waits. New host-call registrations also wake incumbent drivers rather
-than leaving them parked on an obsolete snapshot of pending work.
+**Event-driven store service.** Each store has one coalescing runnable-work
+coordinator. Host settlements and scheduler state transitions record their
+result first, then request service; the coordinator stops when no work is
+runnable and never remains parked racing outstanding host Promises. Host stream
+and future retention remains liveness evidence, not a second pump.
+
+Engine-only promising-entry hops and pending resumption claims are mandatory
+continuations of the current canonical transfer, not extra guest scheduling
+points. Autonomous service therefore waits while an unsettled entry hop exists.
+A hop's own queued settlement may still dispatch, and registration of a genuine
+`SuspensionPoint` park requests service because it removes that barrier. Direct
+call, cancellation, and synchronous entry paths remain distinct from ordinary
+store draining. A bounded work quantum yields to platform timers and I/O only
+between complete canonical steps.
 
 **Between-calls progress.** A host import settling can resume background
 guest work even with no export call in flight. A task waiting for the

harness/src/wasmtime-expectations.ts

@@ -180,7 +180,7 @@ export const WASMTIME_EXPECTATION_GROUPS: readonly WasmtimeExpectationGroup[] =
           rows: [{
             lines: [65],
             cause:
-              'Error: expected trap "wasm trap: cannot block a synchronous task before returning", got "wasm trap: deadlock detected: event loop cannot make further progress (export \'run\': every suspended activation is waiting on a suspension only this scheduler could resume, and none is ready)"',
+              'Error: expected trap "wasm trap: cannot block a synchronous task before returning", got "wasm trap: deadlock detected: event loop cannot make further progress (export \'run\': no runnable work or host call is outstanding)"',
             status: "failed",
           }],
         },

harness/tests/wasmtime_subprocess_test.ts

@@ -26,6 +26,67 @@ Deno.test({
   },
 });
 
+Deno.test({
+  name: "worker-style final output is complete before prompt explicit exit",
+  ignore: !canRun,
+  fn: async () => {
+    const expected = {
+      source: "x.wast",
+      results: [{ line: 1, type: "module", status: "passed" }],
+    };
+    const script = `
+      const bytes = new TextEncoder().encode(JSON.stringify(${
+      JSON.stringify(expected)
+    }) + "\\n");
+      let written = 0;
+      while (written < bytes.length) written += Deno.stdout.writeSync(bytes.subarray(written));
+      setInterval(() => {}, 10);
+      Deno.exit(0);
+    `;
+    const started = performance.now();
+    const outcome = await runChild(
+      new Deno.Command(Deno.execPath(), {
+        args: ["eval", script],
+        stdout: "piped",
+        stderr: "piped",
+      }),
+      1_000,
+    );
+    if (outcome === "timeout" || !outcome.success) {
+      throw new Error("explicit worker exit did not complete");
+    }
+    if (performance.now() - started > 900) {
+      throw new Error("worker did not exit promptly");
+    }
+    const parsed = JSON.parse(new TextDecoder().decode(outcome.stdout));
+    const malformed = validateWorkerResult(
+      { source_filename: "x.wast", commands: [{ line: 1, type: "module" }] },
+      parsed,
+    );
+    if (malformed !== undefined) throw new Error(malformed);
+  },
+});
+
+Deno.test({
+  name: "worker failure before final output remains non-green",
+  ignore: !canRun,
+  fn: async () => {
+    const outcome = await runChild(
+      new Deno.Command(Deno.execPath(), {
+        args: ["eval", 'throw new Error("worker failed before result")'],
+        stdout: "piped",
+        stderr: "piped",
+      }),
+      1_000,
+    );
+    if (outcome === "timeout") throw new Error("failed worker timed out");
+    if (outcome.success) throw new Error("failed worker exited successfully");
+    if (outcome.stdout.length !== 0) {
+      throw new Error("failed worker wrote a partial result");
+    }
+  },
+});
+
 Deno.test("malformed worker result is rejected", () => {
   const doc = {
     source_filename: "x.wast",

runtime/src/embedder/streams.ts

@@ -102,12 +102,13 @@ function reportProducerFailure(
     ? cause
     : new StreamProducerError(where, cause);
   const shared = host.value as unknown as {
-    boundStore?: { hostFailure?: unknown } | null;
+    boundStore?: { hostFailure?: unknown; requestService?: () => void } | null;
   };
   producerFailures.set(host.value as object, err);
   const store = shared.boundStore;
   if (store != null && typeof store === "object") {
     if (store.hostFailure === undefined) store.hostFailure = err;
+    store.requestService?.();
   }
   return err;
 }