#24: exact ambient attribution for JSPI continuation chunks

docs/consumers.md

@@ -94,7 +94,7 @@ Reference implementations developed here, pending upstreaming
 | `exams/iroh-endpoint` | the endpoint exit exam | 5/5: bind+identity, relay echo, WebRTC upgrade, jco#11/#13 assertions, teardown |
 | `ct-runner` | L3 runner for the polymorph-test L1 contract | golden-tested L4 JSONL; drives the websocket suite |
 | `tools/smoke-c0` | C0 smoke legs + report | legs 1–4 (`REPORT.md`) |
-| `tools/smoke-tls` | polymorph-tls conformance under deltic ([#18](https://github.com/lann/deltic/issues/18)) | translate 8/8; suites 6/6 applicable green per target (named xfails: tag-gating [#25](https://github.com/lann/deltic/issues/25), callback-null-context [#24](https://github.com/lann/deltic/issues/24)) |
+| `tools/smoke-tls` | polymorph-tls conformance under deltic ([#18](https://github.com/lann/deltic/issues/18)) | translate 8/8; suites: all applicable cases green on every composition (sole named xfail class: tag-gating [#25](https://github.com/lann/deltic/issues/25); the callback-null-context defect it found, [#24](https://github.com/lann/deltic/issues/24), is fixed — attribution sentinels, `runtime/src/jspi/bridge.ts`) |
 
 Deferred consumer surfaces: experiment-mosh deep E2E
 ([#2](https://github.com/lann/deltic/issues/2)), webcrypto family completion

runtime/src/intrinsics/context.ts

@@ -12,6 +12,7 @@
 
 import { assert_, trapIf } from "../cabi/trap.ts";
 import { currentThread } from "../task/mod.ts";
+import { ambientDebug, dbgId } from "../task/scheduler.ts";
 import type { CurrentThreadLike } from "../task/mod.ts";
 import type { CoreFn } from "../exec/boundary.ts";
 import { UnsupportedFeatureError } from "./errors.ts";
@@ -37,13 +38,37 @@ export function canonContextGet(i: number): number {
   assert_(i < NUM_CONTEXT_SLOTS, `context.get slot ${i} out of range`);
   const result = thread.storage[i];
   assert_(result < 2 ** 32, "context.get value out of i32 range");
+  if (CTX_TRACE) trace(`get[${i}] -> ${result}`, thread);
   return result >>> 0;
 }
 
+// Standing probe (CE_CTX_TRACE=1): per-call context-slot traffic with the
+// full ambient state — the instrument that isolated issue #24. Cheap and
+// env-gated; keep.
+const CTX_TRACE = (() => {
+  try {
+    return Deno.env.get("CE_CTX_TRACE") === "1";
+  } catch {
+    return false;
+  }
+})();
+export function ctxThreadId(t: unknown): string {
+  return dbgId(t);
+}
+function trace(msg: string, thread: unknown): void {
+  const a = ambientDebug();
+  console.error(`[ctx] ${ctxThreadId(thread)} ${msg} storage=${
+    JSON.stringify((thread as CurrentThreadLike).storage)
+  } | stack=[${a.stack.map(ctxThreadId).join(",")}] claims=[${
+    a.claims.map(ctxThreadId).join(",")
+  }] resuming=${a.resuming === null ? "-" : ctxThreadId(a.resuming)}`);
+}
+
 /** definitions.py `canon_context_set` (line 2358). */
 export function canonContextSet(i: number, v: number): void {
   const thread = currentThread<CurrentThreadLike>();
   assert_(i < NUM_CONTEXT_SLOTS, `context.set slot ${i} out of range`);
+  if (CTX_TRACE) trace(`set[${i}] = ${v >>> 0}`, thread);
   thread.storage[i] = v >>> 0;
 }
 

runtime/src/jspi/bridge.ts

@@ -46,6 +46,7 @@ import { isSupported, makePromising, makeSuspending } from "./mechanics.ts";
 import {
   withActivation,
   claimActivationAmbient,
+  dbgId,
   consumeClaimIfRunning,
   maybeCurrentThread,
   releaseActivationAmbient,
@@ -227,6 +228,75 @@ export function enterWasm<T extends (...a: never[]) => unknown>(
  * exactly, and `blockCurrentActivation` has just released this activation's
  * claim on the way in — re-adding it here would strand it.
  */
+// ---------------------------------------------------------------------------
+// Continuation-chunk attribution sentinels (issue #24)
+// ---------------------------------------------------------------------------
+//
+// PROBLEM. Engine continuation chunks — the segments of a promising wasm
+// activation between suspension/hop points — begin as promise REACTIONS,
+// with no synchronous signal to this runtime. When several activations have
+// pending continuations (a settled real suspension racing a fast-path hop,
+// or two fast-path hops from nested entries), the chunks interleave at an
+// empty bracket stack, and every ambient read in a later chunk — a hop's
+// `owner` capture at `claimingFn` entry, or an unsafe intrinsic like
+// `context.set`, which has no hop at all — inherits whatever claim the
+// previous chunk left on top. Claim-stack ordering alone cannot repair
+// this: the release edges are themselves promise reactions. Measured
+// consequence (issue #24): wit-bindgen's callback epilogue restored one
+// task's state pointer into another thread's context slots, and the next
+// invocation of the starved thread's callback hit
+// `assert!(!state.is_null())` (async_support.rs:578) -> unreachable.
+// Reachable only with enough concurrently-suspended sibling activations
+// (first corpus: polymorph-tls' webcrypto-composed suite, three async
+// wit-bindgen components deep).
+//
+// FIX. Exploit the one ordering guarantee the platform does give us:
+// microtasks run FIFO, and between our code queueing a microtask and the
+// engine queueing the continuation reaction there is only synchronous
+// engine-internal promise machinery. So at EVERY point where an engine
+// continuation is about to be queued, queue a SENTINEL first that claims
+// the chunk's owner (move-to-top):
+//
+//   * fast-path hop: sentinel queued synchronously in `claimingFn` before
+//     returning the plain value — the engine queues the hop reaction while
+//     processing that return, so the queue reads [sentinel, chunk].
+//   * genuine suspension: the wrapper attached to the import's thenable
+//     queues the sentinel inside the settle reaction, before returning the
+//     value — the engine (attached to the WRAPPED promise) queues the
+//     resumption when that wrapper returns, so again [sentinel, chunk].
+//     This holds even when several promises settle in one drain: each
+//     pair is queued contiguously from within its own settle reaction.
+//
+// Nothing is delayed or reordered — unlike a serializing gate, which
+// measurably shifted the deterministic-profile backpressure-admission
+// order (async-calls-sync.wast caught it). This is the JSPI substitute for
+// what fibers give wasmtime for free: identity travels with the
+// resumption, here as a claim planted one microtask ahead of it.
+
+function sentinelFor(owner: unknown): void {
+  if (owner === null || owner === undefined) return;
+  // `Promise.resolve().then`, not `queueMicrotask`: identical FIFO
+  // placement, but the latter does not exist in bare engine shells
+  // (SpiderMonkey jsshell; sm-pinned lane caught it).
+  SENTINEL_TICK.then(() => claimActivationAmbient(owner));
+}
+const SENTINEL_TICK = Promise.resolve();
+
+/** Wrap a suspending import's thenable so the eventual resumption chunk is
+ * preceded contiguously by its attribution sentinel. */
+function attributeContinuation<T>(owner: unknown, r: PromiseLike<T>): Promise<T> {
+  return Promise.resolve(r).then(
+    (v) => {
+      sentinelFor(owner);
+      return v;
+    },
+    (e) => {
+      sentinelFor(owner);
+      throw e;
+    },
+  );
+}
+
 export function suspendingImport<T extends (...a: never[]) => unknown>(
   fn: T,
   mode: SuspensionMode,
@@ -255,9 +325,21 @@ export function suspendingImport<T extends (...a: never[]) => unknown>(
       throw e;
     }
     if (r === null || typeof (r as { then?: unknown })?.then !== "function") {
+      // Fast path (jspi pin (j)): the value still returns to wasm through an
+      // engine microtask hop, so the rest of the caller's frame is an engine
+      // continuation chunk like any other. The synchronous claim covers any
+      // reads before the hop; the sentinel re-claims contiguously ahead of
+      // the hop reaction (see the header above — issue #24's second shape
+      // was exactly a fast-path hop chunk misattributed after a sibling's
+      // claim intervened).
       claimActivationAmbient(owner);
+      sentinelFor(owner);
+      return r;
+    }
+    if (SP_TRACE) {
+      console.error(`[sp] hop-suspend owner=${dbgId(owner)} promise=${dbgId(r)}`);
     }
-    return r;
+    return attributeContinuation(owner, r as PromiseLike<unknown>);
   };
   return makeSuspending(claimingFn);
 }
@@ -313,6 +395,14 @@ export function assertModeConsistent(
  * produced at that block point (an event triple's code, a subtask state, a
  * packed copy result).
  */
+const SP_TRACE = (() => {
+  try {
+    return Deno.env.get("CE_SP_TRACE") === "1";
+  } catch {
+    return false;
+  }
+})();
+
 export class SuspensionPoint<T = unknown> implements SchedulableThread {
   readonly promise: Promise<T>;
   #settle!: (v: T) => void;
@@ -349,6 +439,9 @@ export class SuspensionPoint<T = unknown> implements SchedulableThread {
   ) {
     this.#store = store;
     this.owner = owner ?? maybeCurrentThread() ?? task?.implicitThread ?? null;
+    if (SP_TRACE) {
+      console.error(`[sp] mint ${dbgId(this)} owner=${dbgId(this.owner)} task=${dbgId(this.task)}\n${(new Error().stack ?? "").split("\n").slice(2, 5).join("\n")}`);
+    }
     this.promise = new Promise<T>((res, rej) => {
       this.#settle = res;
       this.#fail = rej;
@@ -367,6 +460,9 @@ export class SuspensionPoint<T = unknown> implements SchedulableThread {
   /** Settle the import's Promise; the engine resumes the wasm activation. */
   resume(cancelled: Cancelled = false): void {
     assert_(!this.#done, "resume of an already-resumed suspension point");
+    if (SP_TRACE) {
+      console.error(`[sp] resume ${dbgId(this)} owner=${dbgId(this.owner)}\n${(new Error().stack ?? "").split("\n").slice(2, 5).join("\n")}`);
+    }
     this.#done = true;
     this.#store.stopWaiting(this);
     let value: T;

runtime/src/task/mod.ts

@@ -21,6 +21,7 @@ import {
   CANCELLED_TRUE,
   chooseCandidate,
   Store,
+  dbgId,
 } from "./scheduler.ts";
 import { Thread } from "./thread.ts";
 import { Waitable, WaitableSet } from "./waitable.ts";
@@ -198,6 +199,14 @@ export function liftOptionsEqual(
  * One export activation (definitions.py `class Task`, line 444). Also the
  * task-side borrow scope: `numBorrows` satisfies cabi's `TaskBorrowScope`.
  */
+const ADMIT_TRACE = (() => {
+  try {
+    return Deno.env.get("CE_SP_TRACE") === "1";
+  } catch {
+    return false;
+  }
+})();
+
 export class Task {
   state: TaskState = "initial";
   /** TaskBorrowScope (cabi/context.ts): live borrows lowered into this task. */
@@ -298,6 +307,9 @@ export class Task {
         this.inst.exclusiveThread = thread;
       }
     }
+    if (ADMIT_TRACE) {
+      console.error(`[admit] task=${dbgId(this)} thread=${dbgId(thread)}`);
+    }
     this.registerThread(thread);
     return true;
   }

runtime/src/task/scheduler.ts

@@ -328,17 +328,58 @@ const activationClaims: any[] = [];
 /**
  * Record that the engine will run `t`'s wasm outside our frames.
  *
- * Idempotent: an activation that makes ten non-suspending `Suspending` calls
- * in a row claims once. A null/undefined activation is "no claim" — the
- * instantiation-time shape that has no thread at all.
+ * Idempotent in MEMBERSHIP but not in POSITION: re-claiming MOVES an
+ * existing claim to the top. The stack's contract is "top = the innermost
+ * activation the engine is running outside our frames", and a re-claim is
+ * direct evidence that `t` is running RIGHT NOW (its Suspending import just
+ * returned into its wasm). The previous early-return kept stale order: a
+ * nested callee's claim whose release edge is a promise reaction
+ * (`Store.noteAwaiting` -> `releaseClaimOf`) outlives the callee by a
+ * microtask, and an outer activation's continuation chunk that resumed in
+ * that window re-claimed itself as a NOOP — leaving the finished callee on
+ * top, so every ambient read in the rest of the chunk (the next hop's
+ * `owner` capture, and any unsafe intrinsic like `context.set`, which has
+ * no hop to re-anchor on) answered the wrong thread. Found as issue #24:
+ * wit-bindgen's callback epilogue restored its task pointer into another
+ * thread's context slots, and the next disciplined callback invocation
+ * panicked on a null slot (async_support.rs:578).
+ *
+ * A null/undefined activation is "no claim" — the instantiation-time shape
+ * that has no thread at all.
  */
 // deno-lint-ignore no-explicit-any
 export function claimActivationAmbient(t: any): void {
   if (t === null || t === undefined) return;
-  if (activationClaims.includes(t)) return;
+  if (AMBIENT_TRACE) traceAmbient("claim", t);
+  const i = activationClaims.indexOf(t);
+  if (i === activationClaims.length - 1 && i !== -1) return; // already top
+  if (i !== -1) activationClaims.splice(i, 1);
   activationClaims.push(t);
 }
 
+// #24 probe.
+// deno-lint-ignore no-explicit-any
+function traceAmbient(what: string, t: any): void {
+  // Lazy import avoidance: reuse context.ts's ids via a local map.
+  console.error(
+    `[amb] ${what} ${dbgId(t)} | stack=[${threadStack.map(dbgId).join(",")}] ` +
+      `claims=[${activationClaims.map(dbgId).join(",")}] resuming=${
+        resumingThread === null ? "-" : dbgId(resumingThread)
+      }\n${(new Error().stack ?? "").split("\n").slice(2, 6).join("\n")}`,
+  );
+}
+const dbgIds = new WeakMap<object, number>();
+let nextDbgId = 1;
+export function dbgId(t: unknown): string {
+  if (t === null || t === undefined || typeof t !== "object") return String(t);
+  let id = dbgIds.get(t);
+  if (id === undefined) {
+    id = nextDbgId++;
+    dbgIds.set(t, id);
+  }
+  return `T${id}`;
+}
+
 /**
  * Drop `t`'s activation-ambient claim, if it holds one.
  *
@@ -351,6 +392,7 @@ export function claimActivationAmbient(t: any): void {
 // deno-lint-ignore no-explicit-any
 export function releaseActivationAmbient(t: any): void {
   if (t === null || t === undefined) return;
+  if (AMBIENT_TRACE) traceAmbient("release", t);
   let i = activationClaims.indexOf(t);
   if (i === -1) {
     const implicit = (t as { task?: { implicitThread?: unknown } })?.task
@@ -384,6 +426,7 @@ let resumingThread: any = null;
 /** Claim the ambient for `t` across an engine-driven resumption. */
 // deno-lint-ignore no-explicit-any
 export function setResumingThread(t: any): void {
+  if (AMBIENT_TRACE) traceAmbient("set-resuming", t);
   assert_(
     resumingThread === null || resumingThread === t,
     "two activations claim the resumed ambient at once — the " +
@@ -453,6 +496,19 @@ const AMBIENT_TRACE = (() => {
   }
 })();
 
+/** Diagnostic (#24 probe): the full ambient state, for tracing. */
+export function ambientDebug(): {
+  stack: unknown[];
+  claims: unknown[];
+  resuming: unknown;
+} {
+  return {
+    stack: [...threadStack],
+    claims: [...activationClaims],
+    resuming: resumingThread,
+  };
+}
+
 /** Diagnostic: module-scope state that must NOT survive a completed call. */
 export function ambientResidue(): { stack: number; claim: boolean } {
   return {