feat(phase-5.5): backend bridge — lazy catalog + walker + endpoints

engine/db/migrations_sql/003_add_uuid_column.sql

@@ -0,0 +1,28 @@
+-- 003_add_uuid_column.sql
+--
+-- Adds an explicit, indexed `uuid` column to the `modules` table.
+-- Backfills existing rows by extracting the trailing 8 hex chars of the
+-- generated id (`<prefix>_<slug>_<token_hex(4)>` per ModuleRepository._gen_id).
+--
+-- After this migration, server code that needs canonical UUID lookup
+-- (engine catalog, embed-bundle, drift hashes) does indexed equality
+-- queries against `uuid` instead of `LIKE '%_<8hex>'` string scans.
+--
+-- Partial-apply note: SQLite's Python binding implicitly commits DDL
+-- before the surrounding `with conn:` block can roll back. If the final
+-- CREATE UNIQUE INDEX fails because two pre-existing rows share the same
+-- trailing 8-hex chars (collision probability ~10^-6 for libraries with
+-- <1000 modules), the column is added but the migrations table is not
+-- bumped, and a retry would fail with `duplicate column name: uuid`.
+-- Recovery is manual: rename one of the colliding ids in the modules
+-- table, then re-run migrate(). This is acceptable for a forward-only
+-- one-shot migration on a per-user library DB. Collisions in practice
+-- are vanishingly rare and recoverable.
+
+ALTER TABLE modules ADD COLUMN uuid TEXT;
+
+UPDATE modules
+SET uuid = substr(id, -8)
+WHERE uuid IS NULL;
+
+CREATE UNIQUE INDEX IF NOT EXISTS idx_modules_uuid ON modules(uuid);

engine/db/repositories.py

@@ -13,6 +13,7 @@ import sqlite3
 from typing import Any
 
 from engine._utils import now_iso as _now
+from engine.modules.snapshot import payload_hash
 
 _TYPE_PREFIX = {
     "wildcard": "wc",
@@ -49,15 +50,18 @@ def _slug(name: str) -> str:
 
 
 def _row_to_module(row: sqlite3.Row) -> dict[str, Any]:
+    payload = json.loads(row["payload"])
     return {
         "id": row["id"],
+        "uuid": row["uuid"],
         "type": row["type"],
         "name": row["name"],
         "description": row["description"],
         "category_id": row["category_id"],
         "tags": json.loads(row["tags"]),
         "is_favorite": bool(row["is_favorite"]),
-        "payload": json.loads(row["payload"]),
+        "payload": payload,
+        "payload_hash": payload_hash(payload),
         "version": row["version"],
         "created_at": row["created_at"],
         "updated_at": row["updated_at"],
@@ -68,9 +72,11 @@ class ModuleRepository:
     def __init__(self, conn: sqlite3.Connection) -> None:
         self._conn = conn
 
-    def _gen_id(self, type: str, name: str) -> str:
+    def _gen_id(self, type: str, name: str) -> tuple[str, str]:
+        """Generate `(id, uuid)` for a new row. id = `<prefix>_<slug>_<uuid>`."""
         prefix = _TYPE_PREFIX.get(type, "mod")
-        return f"{prefix}_{_slug(name)}_{secrets.token_hex(4)}"
+        uuid = secrets.token_hex(4)
+        return f"{prefix}_{_slug(name)}_{uuid}", uuid
 
     def create(
         self,
@@ -87,16 +93,16 @@ class ModuleRepository:
             raise ValueError(
                 f"unknown module type {type!r}; expected one of {sorted(_VALID_TYPES)}"
             )
-        mid = self._gen_id(type, name)
+        mid, uuid = self._gen_id(type, name)
         now = _now()
         with self._conn:
             self._conn.execute(
                 "INSERT INTO modules("
-                "id, type, name, description, category_id, tags, "
+                "id, uuid, type, name, description, category_id, tags, "
                 "is_favorite, payload, version, created_at, updated_at"
-                ") VALUES(?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?);",
+                ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?, ?);",
                 (
-                    mid, type, name, description, category_id,
+                    mid, uuid, type, name, description, category_id,
                     json.dumps(tags), int(is_favorite),
                     json.dumps(payload), now, now,
                 ),
@@ -111,6 +117,40 @@ class ModuleRepository:
             raise ModuleNotFound(module_id)
         return _row_to_module(row)
 
+    def get_by_uuid(self, uuid: str) -> dict[str, Any]:
+        """Lookup by indexed uuid. Raises ModuleNotFound on miss."""
+        cur = self._conn.execute(
+            "SELECT * FROM modules WHERE uuid = ?;", (uuid,),
+        )
+        row = cur.fetchone()
+        if row is None:
+            raise ModuleNotFound(uuid)
+        return _row_to_module(row)
+
+    def get_by_uuids(self, uuids: list[str]) -> list[dict[str, Any]]:
+        """Bulk indexed lookup. Dedups input and returns rows in the same
+        order as the (deduped) input. Silently skips missing uuids —
+        callers that need explicit miss-detection should compare returned
+        uuids against the input set.
+
+        Note on input size: callers (Test Runner request walker, embed-
+        bundle endpoint) pass at most a few dozen uuids in practice, so
+        we do not chunk. SQLite's `SQLITE_LIMIT_VARIABLE_NUMBER` defaults
+        to 32766 on modern builds (3.32+) and 999 on older ones — well
+        above the expected upper bound."""
+        if not uuids:
+            return []
+        unique = list(dict.fromkeys(uuids))  # dedup, preserve input order
+        placeholders = ",".join("?" for _ in unique)
+        cur = self._conn.execute(
+            f"SELECT * FROM modules WHERE uuid IN ({placeholders});",
+            unique,
+        )
+        # SQLite makes no order guarantee for IN-predicate results, so we
+        # re-order against the input list to honor the docstring contract.
+        by_uuid = {r["uuid"]: r for r in cur.fetchall()}
+        return [_row_to_module(by_uuid[u]) for u in unique if u in by_uuid]
+
     def update(
         self,
         module_id: str,

engine/modules/__init__.py

@@ -30,6 +30,16 @@ class _RuntimeResolveContext:
         return None
 
     def get_module(self, uuid: str) -> dict[str, Any] | None:
+        """Catalog lookup. Pure O(1) dict get. Spec §2.5.
+
+        No DB fallback. No network. No side effects. If the catalog does
+        not contain `uuid`, returns None and lets the resolver decide
+        (lenient → warning, strict → UnknownRefError).
+
+        Engine isolation invariant — `engine/modules/__init__.py` MUST NOT
+        import from `engine.db`, `wp_api`, or `wp_nodes`. Pinned by
+        `tests/engine/modules/test_resolve_context_isolation.py`.
+        """
         return self._catalog.get(uuid)
 
 

engine/modules/snapshot.py

@@ -1,4 +1,5 @@
-"""Snapshot freezing, content hashing, and legacy-shape coercion.
+"""Snapshot freezing, content hashing, legacy-shape coercion, and transitive
+ref walking.
 
 A *snapshot* is the JSON shape embedded in a ComfyUI workflow's WP_Context
 node when a user picks a library row. The snapshot is decoupled from the
@@ -12,9 +13,42 @@ from __future__ import annotations
 
 import hashlib
 import json
-from typing import Any
+from collections.abc import Callable, Iterable
+from dataclasses import dataclass, field
+from typing import Any, Literal, TypedDict
 
 from engine._utils import now_iso as _now_iso
+from engine.syntax.tokenize import tokenize_text
+from engine.syntax.types import TokenKind
+
+
+class _SourceUser(TypedDict):
+    kind: Literal["user"]
+
+
+class _SourceDep(TypedDict):
+    kind: Literal["dep"]
+    parent_uuids: list[str]
+
+
+SnapshotSource = _SourceUser | _SourceDep
+
+
+class SnapshotEntry(TypedDict):
+    """Canonical shape stored in `__wp_catalog__` and embedded in
+    WP_Context workflow JSON. Spec §2.4 — pinned, do not rename fields.
+
+    `payload_hash` covers `payload` only (not name/description/tags) so
+    a rename does not flip drift state. `source` records why the entry
+    is in the embed: explicit user pick vs. transitive dep of one or
+    more picks (multi-parent supported)."""
+    snapshot_version: Literal[1]
+    uuid: str            # 8 hex chars
+    type: str            # ModuleType — wildcards-only in catalog (spec §2.7)
+    name: str            # display label
+    payload: dict[str, Any]
+    payload_hash: str    # SHA-256 of payload
+    source: SnapshotSource
 
 
 def _fresh_instance() -> dict[str, Any]:
@@ -84,3 +118,169 @@ def coerce_legacy_module(raw: dict[str, Any]) -> dict[str, Any]:
         "payload": payload,
         "instance": _fresh_instance(),
     }
+
+
+# ---------------------------------------------------------------------------
+# Transitive-ref walker (spec §2.7, §2.8, §2.10, §6)
+# ---------------------------------------------------------------------------
+
+
+class WalkOverflow(TypedDict):
+    """Walker-recorded anomaly. Returned alongside the partial snapshot
+    bundle; never aborts the walk. Spec §2.8."""
+    uuid: str
+    reason: Literal["max_depth", "cycle_detected", "missing_target"]
+
+
+@dataclass
+class WalkResult:
+    """Return value of `walk_transitive_refs`. Two fields, two concerns:
+    `snapshots` is the materialised catalog dict; `walk_overflow` lists
+    every uuid the walker could not include (cycle, depth, miss)."""
+    snapshots: dict[str, SnapshotEntry] = field(default_factory=dict)
+    walk_overflow: list[WalkOverflow] = field(default_factory=list)
+
+
+def _extract_at_uuid_refs(payload: dict[str, Any]) -> set[str]:
+    """Collect all ``@{8hex}`` refs referenced inside a payload's
+    string-typed leaves.  Uses ``tokenize_text`` exclusively — only
+    canonical 8-hex-char refs (spec §2.4) are followed.
+
+    Surface-agnostic: collects from any string in the dict/list tree;
+    the caller decides which module types have their payloads walked
+    (spec §2.7)."""
+    refs: set[str] = set()
+
+    def _scan(value: Any) -> None:
+        if isinstance(value, str):
+            for tok in tokenize_text(value):
+                if tok.kind == TokenKind.REF:
+                    uid = (tok.meta or {}).get("uuid")
+                    if isinstance(uid, str):
+                        refs.add(uid)
+        elif isinstance(value, dict):
+            for v in value.values():
+                _scan(v)
+        elif isinstance(value, list):
+            for item in value:
+                _scan(item)
+
+    _scan(payload)
+    return refs
+
+
+def _row_to_snapshot_entry(
+    row: dict[str, Any], source: SnapshotSource,
+) -> SnapshotEntry:
+    """Convert a repository row dict to a canonical SnapshotEntry."""
+    return {
+        "snapshot_version": 1,
+        "uuid": row["uuid"],
+        "type": row["type"],
+        "name": row["name"],
+        "payload": row["payload"],
+        "payload_hash": row.get("payload_hash") or payload_hash(row["payload"]),
+        "source": source,
+    }
+
+
+def walk_transitive_refs(
+    root_uuids: Iterable[str],
+    *,
+    fetch_module: Callable[[str], dict[str, Any] | None],
+    max_depth: int = 8,
+) -> WalkResult:
+    """Lazy transitive walker. Spec §2.10.
+
+    Starts from ``root_uuids``, follows ``@{8hex}`` refs in wildcard payloads
+    only (spec §2.7), and returns a ``WalkResult`` containing every reachable
+    SnapshotEntry plus a list of overflow records (cycle, depth, missing).
+
+    Cycle detection uses a per-path ancestor frozenset carried with each BFS
+    queue entry. When we pop ``(uuid, depth, parent, ancestors)`` and find
+    ``uuid`` already in ``ancestors``, we record ``cycle_detected`` and skip
+    without emitting the node. This means the node that *caused* the cycle
+    reference is already snapshotted (it was emitted when we first reached it
+    on a shorter path), so both sides of a two-node cycle appear in snapshots.
+
+    Engine isolation invariant: this function does NOT import from ``engine.db``
+    or ``wp_api``; the caller injects a ``fetch_module(uuid) -> row | None``
+    callable. WP_Context graph nodes never call this — they receive a
+    pre-walked dict embedded in workflow JSON.
+    """
+    result = WalkResult()
+    # parent_map: uuid -> list of direct parent uuids (accumulated across BFS)
+    parent_map: dict[str, list[str]] = {}
+    seen_overflow: set[tuple[str, str]] = set()
+
+    def _record_overflow(uuid: str, reason: str) -> None:
+        key = (uuid, reason)
+        if key in seen_overflow:
+            return
+        seen_overflow.add(key)
+        result.walk_overflow.append(
+            {"uuid": uuid, "reason": reason},  # type: ignore[arg-type]
+        )
+
+    # Queue items: (uuid, depth, parent_uuid_or_None, ancestor_frozenset)
+    # ancestor_frozenset contains all uuids on the path from a root to *before*
+    # this node — used to detect back-edges (cycles).
+    queue: list[tuple[str, int, str | None, frozenset[str]]] = []
+
+    for u in root_uuids:
+        queue.append((u, 0, None, frozenset()))
+
+    while queue:
+        uuid, depth, parent, ancestors = queue.pop(0)
+
+        # Track parentage before any short-circuit so multi-parent deps
+        # accumulate even if we have already snapshotted the child.
+        if parent is not None:
+            parent_map.setdefault(uuid, [])
+            if parent not in parent_map[uuid]:
+                parent_map[uuid].append(parent)
+
+        # Cycle: this uuid appears on the path from a root to here.
+        if uuid in ancestors:
+            _record_overflow(uuid, "cycle_detected")
+            continue
+
+        # Already snapshotted on a shorter BFS path — skip emission but
+        # parent tracking above already ran, so multi-parent deps accumulate.
+        if uuid in result.snapshots:
+            continue
+
+        if depth > max_depth:
+            _record_overflow(uuid, "max_depth")
+            continue
+
+        row = fetch_module(uuid)
+        if row is None:
+            _record_overflow(uuid, "missing_target")
+            continue
+
+        # Determine source based on whether this uuid has a parent.
+        if parent is None:
+            source: SnapshotSource = {"kind": "user"}
+        else:
+            source = {"kind": "dep", "parent_uuids": list(parent_map[uuid])}
+        result.snapshots[uuid] = _row_to_snapshot_entry(row, source)
+
+        # Only walk wildcard payloads — spec §2.7.
+        if row.get("type") != "wildcard":
+            continue
+
+        new_ancestors = ancestors | {uuid}
+        for child_uuid in _extract_at_uuid_refs(row["payload"]):
+            queue.append((child_uuid, depth + 1, uuid, new_ancestors))
+
+    # After walk: re-stamp every dep's parent list (BFS may have accumulated
+    # additional parents after first emission).
+    for uuid, parents in parent_map.items():
+        if uuid not in result.snapshots:
+            continue
+        existing = result.snapshots[uuid]
+        if existing["source"]["kind"] == "dep":
+            existing["source"]["parent_uuids"] = list(parents)
+
+    return result

engine/pipeline.py

@@ -37,10 +37,20 @@ class PipelineEngine:
         ctx.setdefault("__wp_internal_flags__", {})
 
         for index, module in enumerate(modules):
-            if not getattr(module, "enabled", True):
+            # Normalise id/type/enabled reads to work for both dicts and objects.
+            if isinstance(module, dict):
+                _module_id = module.get("id", "")
+                _module_type_raw = module.get("type", None) or ""
+                _module_enabled = module.get("enabled", True)
+            else:
+                _module_id = getattr(module, "id", "")
+                _module_type_raw = getattr(module, "type", None) or ""
+                _module_enabled = getattr(module, "enabled", True)
+
+            if not _module_enabled:
                 ctx["__wp_trace__"].append({
-                    "id": getattr(module, "id", ""),
-                    "type": getattr(module, "type", ""),
+                    "id": _module_id,
+                    "type": _module_type_raw,
                     "enabled": False,
                     "status": "skipped_disabled",
                     "writes": [],
@@ -48,7 +58,7 @@ class PipelineEngine:
                 })
                 continue
 
-            module_type = getattr(module, "type", None) or ""
+            module_type = _module_type_raw
             try:
                 # coerce_legacy_module expects a dict; convert dataclasses or
                 # duck-typed objects by reading known attrs via getattr.
@@ -75,7 +85,7 @@ class PipelineEngine:
                     "Failed to coerce module %r at index %s: %s", module, index, e
                 )
                 ctx["__wp_trace__"].append({
-                    "id": getattr(module, "id", ""),
+                    "id": _module_id,
                     "type": module_type,
                     "enabled": True,
                     "status": "failed",
@@ -93,7 +103,7 @@ class PipelineEngine:
                     index,
                 )
                 ctx["__wp_trace__"].append({
-                    "id": getattr(module, "id", ""),
+                    "id": _module_id,
                     "type": module_type,
                     "enabled": True,
                     "status": "skipped_unknown_type",
@@ -108,7 +118,7 @@ class PipelineEngine:
                 ctx["__wp_warnings__"].append({
                     "type": "handler_error",
                     "severity": "error",
-                    "module_id": getattr(module, "id", ""),
+                    "module_id": _module_id,
                     "source_field": "payload",
                     "position": 0,
                     "token_index": None,
@@ -122,7 +132,7 @@ class PipelineEngine:
                     ),
                 })
                 ctx["__wp_trace__"].append({
-                    "id": getattr(module, "id", ""),
+                    "id": _module_id,
                     "type": module_type,
                     "enabled": True,
                     "status": "failed",
@@ -144,7 +154,7 @@ class PipelineEngine:
                 })
 
             ctx["__wp_trace__"].append({
-                "id": getattr(module, "id", ""),
+                "id": _module_id,
                 "type": module_type,
                 "enabled": True,
                 "status": "ok",

engine/syntax/resolve.py

@@ -297,8 +297,14 @@ def _resolve_ref(
         )
         return ""
 
-    # Pick weighted option
-    options = module.get("options", [])
+    # Pick weighted option.
+    # Support both SnapshotEntry shape (payload.options, spec §2.4) and the
+    # legacy flat shape used by existing unit tests (options at top level).
+    payload_dict = module.get("payload")
+    if isinstance(payload_dict, dict):
+        options = payload_dict.get("options", [])
+    else:
+        options = module.get("options", [])
     chosen = _pick_weighted(options, ctx.rng)
     if chosen is None:
         return ""

src/manager/__tests__/ConstraintEditor.test.ts

@@ -30,6 +30,7 @@ function makeWildcardRow(id: string, name: string, opts: {
 } = {}): ModuleRow {
   return {
     id,
+    uuid: "aabbccdd",
     name,
     type: "wildcard",
     description: "",
@@ -40,6 +41,7 @@ function makeWildcardRow(id: string, name: string, opts: {
       options: (opts.values ?? []).map((v, i) => ({ id: `o${i}`, value: v, weight: 1 })),
       sub_categories: opts.subs ?? [],
     },
+    payload_hash: "0".repeat(64),
     version: 1,
     created_at: "",
     updated_at: "",

src/manager/__tests__/ImportExport.test.ts

@@ -29,6 +29,7 @@ const apiAny = api as unknown as {
 function mkModule(over: Partial<ModuleRow>): ModuleRow {
   return {
     id: "wc_aaaa1111",
+    uuid: "aabbccdd",
     type: "wildcard",
     name: "Sample",
     description: "",
@@ -36,6 +37,7 @@ function mkModule(over: Partial<ModuleRow>): ModuleRow {
     tags: [],
     is_favorite: false,
     payload: { options: [] },
+    payload_hash: "0".repeat(64),
     version: 1,
     created_at: "2024-01-01T00:00:00Z",
     updated_at: "2024-01-01T00:00:00Z",

src/manager/__tests__/PipelineSteps.test.ts

@@ -6,6 +6,7 @@ import type { ModuleRow, ModuleType, PipelineStep } from "../api/types";
 function makeModule(over: Partial<ModuleRow>): ModuleRow {
   return {
     id: "m_a",
+    uuid: "aabbccdd",
     type: "wildcard",
     name: "alpha",
     description: "",
@@ -13,6 +14,7 @@ function makeModule(over: Partial<ModuleRow>): ModuleRow {
     tags: [],
     is_favorite: false,
     payload: {},
+    payload_hash: "0".repeat(64),
     version: 1,
     created_at: "",
     updated_at: "",

src/manager/__tests__/TestRunner.test.ts

@@ -6,7 +6,7 @@ import type { ModuleRow } from "../api/types";
 vi.mock("../api/client", () => {
   const MOCK_MODULES: ModuleRow[] = [
     {
-      id: "wc_a", type: "wildcard", name: "Hair Color",
+      id: "wc_a", uuid: "aabbccdd", type: "wildcard", name: "Hair Color",
       description: "", category_id: null, tags: [], is_favorite: false,
       payload: {
         options: [
@@ -16,10 +16,11 @@ vi.mock("../api/client", () => {
         sub_categories: ["warm", "cool"],
         var_binding: "hair_color",
       },
+      payload_hash: "0".repeat(64),
       version: 1, created_at: "", updated_at: "",
     },
     {
-      id: "wc_o", type: "wildcard", name: "Outfit",
+      id: "wc_o", uuid: "bbbbbbbb", type: "wildcard", name: "Outfit",
       description: "", category_id: null, tags: [], is_favorite: false,
       payload: {
         options: [
@@ -27,26 +28,29 @@ vi.mock("../api/client", () => {
         ],
         sub_categories: [],
       },
+      payload_hash: "0".repeat(64),
       version: 1, created_at: "", updated_at: "",
     },
     {
-      id: "fv_a", type: "fixed_values", name: "Subject Profile",
+      id: "fv_a", uuid: "cccccccc", type: "fixed_values", name: "Subject Profile",
       description: "", category_id: null, tags: [], is_favorite: false,
       payload: { values: [{ var: "name", value: "Mira" }, { var: "age", value: "29" }] },
+      payload_hash: "0".repeat(64),
       version: 1, created_at: "", updated_at: "",
     },
     {
-      id: "cb_a", type: "combine", name: "Subject Phrase",
+      id: "cb_a", uuid: "dddddddd", type: "combine", name: "Subject Phrase",
       description: "", category_id: null, tags: [], is_favorite: false,
       payload: {
         template: "$name with $hair_color hair",
         output_var: "subject_phrase",
         input_vars: ["name", "hair_color"],
       },
+      payload_hash: "0".repeat(64),
       version: 1, created_at: "", updated_at: "",
     },
     {
-      id: "dv_a", type: "derivation", name: "Always Append",
+      id: "dv_a", uuid: "eeeeeeee", type: "derivation", name: "Always Append",
       description: "", category_id: null, tags: [], is_favorite: false,
       payload: {
         rules: [{
@@ -57,10 +61,11 @@ vi.mock("../api/client", () => {
           }],
         }],
       },
+      payload_hash: "0".repeat(64),
       version: 1, created_at: "", updated_at: "",
     },
     {
-      id: "cn_a", type: "constraint", name: "Hair × Outfit",
+      id: "cn_a", uuid: "ffffffff", type: "constraint", name: "Hair × Outfit",
       description: "", category_id: null, tags: [], is_favorite: false,
       payload: {
         source_wildcard_id: "wc_a",
@@ -68,15 +73,17 @@ vi.mock("../api/client", () => {
         matrix: { warm: {} },
         exceptions: [],
       },
+      payload_hash: "0".repeat(64),
       version: 1, created_at: "", updated_at: "",
     },
     {
-      id: "pl_a", type: "pipeline", name: "Quick Portrait",
+      id: "pl_a", uuid: "11111111", type: "pipeline", name: "Quick Portrait",
       description: "", category_id: null, tags: [], is_favorite: false,
       payload: { steps: [
         { id: "s1", module_id: "fv_a", enabled: true },
         { id: "s2", module_id: "wc_a", enabled: true },
       ] },
+      payload_hash: "0".repeat(64),
       version: 1, created_at: "", updated_at: "",
     },
   ];

src/manager/__tests__/api-types.test.ts

@@ -0,0 +1,83 @@
+/**
+ * Spec §4.3 — TS twin of canonical Python SnapshotEntry shape. Pinned
+ * here so a future refactor on either side surfaces a type error before
+ * runtime drift can happen.
+ */
+import { describe, expect, it } from "vitest";
+import type {
+  ModuleRow,
+  SnapshotEntry,
+  WalkOverflow,
+  EmbedBundle,
+} from "../api/types";
+import { api } from "../api/client";
+
+describe("API types", () => {
+  it("SnapshotEntry has the seven canonical fields", () => {
+    // Compile-time check: this object literal must satisfy the type.
+    const entry: SnapshotEntry = {
+      snapshot_version: 1,
+      uuid: "aabbccdd",
+      type: "wildcard",
+      name: "color",
+      payload: { options: [] },
+      payload_hash: "0".repeat(64),
+      source: { kind: "user" },
+    };
+    expect(entry.snapshot_version).toBe(1);
+  });
+
+  it("SnapshotEntry source supports kind=dep with parent_uuids", () => {
+    const entry: SnapshotEntry = {
+      snapshot_version: 1, uuid: "a1b2c3d4", type: "wildcard", name: "x",
+      payload: {}, payload_hash: "h".repeat(64),
+      source: { kind: "dep", parent_uuids: ["aabbccdd"] },
+    };
+    expect((entry.source as { kind: "dep"; parent_uuids: string[] }).parent_uuids)
+      .toEqual(["aabbccdd"]);
+  });
+
+  it("WalkOverflow shape", () => {
+    const overflows: WalkOverflow[] = [
+      { uuid: "x", reason: "max_depth" },
+      { uuid: "y", reason: "cycle_detected" },
+      { uuid: "z", reason: "missing_target" },
+    ];
+    expect(overflows).toHaveLength(3);
+  });
+
+  it("EmbedBundle has flat envelope shape", () => {
+    const bundle: EmbedBundle = {
+      modules: [{ type: "wildcard" }],
+      snapshots: {},
+      pickOrder: ["aabbccdd"],
+      walkOverflow: [],
+    };
+    expect(bundle.pickOrder).toEqual(["aabbccdd"]);
+  });
+
+  it("ModuleRow includes uuid and payload_hash", () => {
+    const row: ModuleRow = {
+      id: "wc_x_aabbccdd",
+      uuid: "aabbccdd",
+      type: "wildcard",
+      name: "x",
+      description: "",
+      category_id: null,
+      tags: [],
+      is_favorite: false,
+      payload: {},
+      payload_hash: "0".repeat(64),
+      version: 1,
+      created_at: "2026-04-28T00:00:00.000000Z",
+      updated_at: "2026-04-28T00:00:00.000000Z",
+    };
+    expect(row.uuid).toBe("aabbccdd");
+    expect(row.payload_hash).toBeDefined();
+  });
+
+  it("client exposes embedBundle and hashes methods", () => {
+    expect(typeof api.modules.embedBundle).toBe("function");
+    expect(typeof api.modules.hashes).toBe("function");
+  });
+});

src/manager/__tests__/bundleSelection.test.ts

@@ -19,6 +19,7 @@ import type { CategoryRow, ModuleRow } from "../api/types";
 function mkModule(over: Partial<ModuleRow>): ModuleRow {
   return {
     id: "wc_aaaa1111",
+    uuid: "aabbccdd",
     type: "wildcard",
     name: "Sample",
     description: "",
@@ -26,6 +27,7 @@ function mkModule(over: Partial<ModuleRow>): ModuleRow {
     tags: [],
     is_favorite: false,
     payload: {},
+    payload_hash: "0".repeat(64),
     version: 1,
     created_at: "2024-01-01T00:00:00Z",
     updated_at: "2024-01-01T00:00:00Z",

src/manager/__tests__/resolver.test.ts

@@ -24,6 +24,7 @@ function makeWildcard(
 ): ModuleRow {
   return {
     id: extras.id ?? `wc_${name.toLowerCase()}`,
+    uuid: "aabbccdd",
     type: "wildcard",
     name,
     description: "",
@@ -40,6 +41,7 @@ function makeWildcard(
       sub_categories: extras.sub_categories ?? [],
       var_binding: extras.var_binding,
     },
+    payload_hash: "0".repeat(64),
     version: 1,
     created_at: "",
     updated_at: "",
@@ -255,9 +257,10 @@ describe("runStep", () => {
 
   it("runs a fixed_values step writing every binding into ctx", () => {
     const f: ModuleRow = {
-      id: "fv1", type: "fixed_values", name: "Pack",
+      id: "fv1", uuid: "aabbccdd", type: "fixed_values", name: "Pack",
       description: "", category_id: null, tags: [], is_favorite: false,
       payload: { values: [{ var: "city", value: "Tokyo" }, { name: "$ward", value: "Shibuya" }] },
+      payload_hash: "0".repeat(64),
       version: 1, created_at: "", updated_at: "",
     };
     const step: PipelineStep = { id: "s1", module_id: f.id, enabled: true };

src/manager/__tests__/wildcardSyntax.test.ts

@@ -23,6 +23,7 @@ function wc(
   }
   return {
     id: partial.id ?? `wc_${partial.name}`,
+    uuid: partial.uuid ?? "aabbccdd",
     type: "wildcard",
     name: partial.name,
     description: partial.description ?? "",
@@ -30,6 +31,7 @@ function wc(
     tags: partial.tags ?? [],
     is_favorite: partial.is_favorite ?? false,
     payload,
+    payload_hash: partial.payload_hash ?? "0".repeat(64),
     version: partial.version ?? 1,
     created_at: partial.created_at ?? "",
     updated_at: partial.updated_at ?? "",

src/manager/api/client.ts

@@ -1,5 +1,6 @@
 import type {
   CategoryCreateInput, CategoryRow,
+  EmbedBundle,
   ImportBundle, ImportResult,
   MatchRequest, MatchResponse,
   ModuleCreateInput, ModuleListResponse, ModuleRow, ModuleUpdateInput,
@@ -91,6 +92,16 @@ export const api = {
         method: "POST", body: JSON.stringify(body),
       });
     },
+    embedBundle(uuids: string[]) {
+      return request<EmbedBundle>("/wp/api/modules/embed-bundle", {
+        method: "POST", body: JSON.stringify({ uuids }),
+      });
+    },
+    hashes() {
+      return request<{ hashes: Record<string, string> }>(
+        "/wp/api/modules/hashes", { method: "GET" },
+      );
+    },
   },
   categories: {
     list() {

src/manager/api/types.ts

@@ -107,6 +107,7 @@ export interface PipelinePayload {
 
 export interface ModuleRow {
   id: string;
+  uuid: string;             // spec §4.2
   type: ModuleType;
   name: string;
   description: string;
@@ -114,6 +115,7 @@ export interface ModuleRow {
   tags: string[];
   is_favorite: boolean;
   payload: Record<string, unknown>;
+  payload_hash: string;     // spec §4.2
   version: number;
   created_at: string;
   updated_at: string;
@@ -224,3 +226,53 @@ export interface ImportResult {
   categories_imported: number;
   skipped: string[];
 }
+
+/**
+ * Spec §2.4 — canonical snapshot entry. Mirrors the Python
+ * `engine.modules.snapshot.SnapshotEntry` TypedDict. Stored:
+ *   - in `__wp_catalog__` at runtime (memory only)
+ *   - in WP_Context workflow JSON (persisted, decoupled from DB)
+ *
+ * `payload_hash` covers `payload` only (not `name` / `tags` / etc.) so a
+ * rename does not flip drift state.
+ */
+export interface SnapshotEntry {
+  snapshot_version: 1;
+  uuid: string;             // 8 hex chars
+  type: ModuleType;         // wildcards-only in catalog (spec §2.7)
+  name: string;
+  payload: Record<string, unknown>;
+  payload_hash: string;     // SHA-256 hex, 64 chars
+  source:
+    | { kind: "user" }
+    | { kind: "dep"; parent_uuids: string[] };
+}
+
+/**
+ * Walker-recorded anomaly. Spec §2.8 — extensible discriminator. New
+ * reasons can be added without breaking existing consumers; they
+ * default-handle unknown variants.
+ */
+export interface WalkOverflow {
+  uuid: string;
+  reason: "max_depth" | "cycle_detected" | "missing_target";
+}
+
+/**
+ * Server response shape from `POST /wp/api/modules/embed-bundle`.
+ * Pre-split so the SPA picker doesn't filter wildcards out of a mixed
+ * bundle. Spec §4.2.
+ *
+ * - `modules`: full payloads of explicit picks, in input order. The
+ *   pipeline executes these.
+ * - `snapshots`: wildcard catalog (picks of type wildcard + transitive
+ *   wildcards). Becomes `ctx['__wp_catalog__']` at run time.
+ * - `pickOrder`: uuid order for UI display.
+ * - `walkOverflow`: cycles / depth caps / missing targets the walker hit.
+ */
+export interface EmbedBundle {
+  modules: Record<string, unknown>[];
+  snapshots: Record<string, SnapshotEntry>;
+  pickOrder: string[];
+  walkOverflow: WalkOverflow[];
+}

tests/engine/db/test_migrations.py

@@ -1,6 +1,8 @@
 """Tests for the versioned migration runner."""
+import sqlite3
+
 from engine.db.connection import get_connection
-from engine.db.migrations import current_version, migrate
+from engine.db.migrations import _discover, current_version, migrate
 
 
 def test_initial_version_is_zero(tmp_path):
@@ -29,7 +31,7 @@ def test_migrate_records_version(tmp_path):
     migrate(conn)
     # Keep this assertion in sync with the highest-numbered migration in
     # ``engine/db/migrations_sql``.
-    assert current_version(conn) == 2
+    assert current_version(conn) == 3
     conn.close()
 
 
@@ -49,6 +51,130 @@ def test_modules_table_has_expected_columns(tmp_path):
     assert cols == {
         "id", "type", "name", "description", "category_id",
         "tags", "is_favorite", "payload", "version",
-        "created_at", "updated_at",
+        "created_at", "updated_at", "uuid",
     }
     conn.close()
+
+
+def test_003_adds_uuid_column_with_backfill_and_index(tmp_path):
+    """003_add_uuid_column.sql adds uuid column, backfills from id suffix,
+    and creates a unique index. Verified end-to-end against a real
+    sqlite3 connection so we catch syntax errors the runner ignores."""
+    db = tmp_path / "lib.db"
+    conn = sqlite3.connect(str(db))
+    conn.row_factory = sqlite3.Row
+
+    # Migrate to v002 first (current head before this plan), seed a row
+    # with the legacy schema (no uuid column).
+    pairs = _discover()
+    for version, sql_file in pairs:
+        if version >= 3:
+            break
+        with conn:
+            conn.executescript(sql_file.read_text())
+            conn.execute(
+                "INSERT OR IGNORE INTO migrations(version, applied_at) "
+                "VALUES (?, datetime('now'));",
+                (version,),
+            )
+    conn.execute(
+        "INSERT INTO modules(id, type, name, description, category_id, "
+        "tags, is_favorite, payload, version, created_at, updated_at) "
+        "VALUES ('wc_color_aabbccdd', 'wildcard', 'color', '', NULL, "
+        "'[]', 0, '{}', 1, datetime('now'), datetime('now'));"
+    )
+    conn.commit()
+
+    # Run full migrate (will pick up 003)
+    migrate(conn)
+
+    # Column exists
+    cols = {r["name"] for r in conn.execute("PRAGMA table_info(modules);")}
+    assert "uuid" in cols
+
+    # Backfill correct
+    row = conn.execute(
+        "SELECT uuid FROM modules WHERE id = 'wc_color_aabbccdd';"
+    ).fetchone()
+    assert row["uuid"] == "aabbccdd"
+
+    # Unique index present
+    idx = {r["name"] for r in conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='index';"
+    )}
+    assert "idx_modules_uuid" in idx
+
+
+def test_003_backfills_multiple_rows_and_preserves_pre_set_uuid(tmp_path):
+    """Multi-row backfill: every NULL uuid gets the trailing 8 hex of its id.
+    Pre-set uuids are preserved (the WHERE uuid IS NULL guard).
+
+    This pins the multi-row UPDATE path that the single-row test in
+    `test_003_adds_uuid_column_with_backfill_and_index` does not exercise."""
+    db = tmp_path / "lib.db"
+    conn = sqlite3.connect(str(db))
+    conn.row_factory = sqlite3.Row
+
+    # Migrate to v002 first, seed three rows on the legacy schema.
+    pairs = _discover()
+    for version, sql_file in pairs:
+        if version >= 3:
+            break
+        with conn:
+            conn.executescript(sql_file.read_text())
+            conn.execute(
+                "INSERT OR IGNORE INTO migrations(version, applied_at) "
+                "VALUES (?, datetime('now'));",
+                (version,),
+            )
+    for id_ in ("wc_a_aaaaaaaa", "wc_b_bbbbbbbb", "fv_c_cccccccc"):
+        conn.execute(
+            "INSERT INTO modules(id, type, name, description, category_id, "
+            "tags, is_favorite, payload, version, created_at, updated_at) "
+            "VALUES (?, 'wildcard', 'x', '', NULL, '[]', 0, '{}', 1, "
+            "datetime('now'), datetime('now'));",
+            (id_,),
+        )
+    conn.commit()
+
+    migrate(conn)
+
+    rows = {
+        r["id"]: r["uuid"]
+        for r in conn.execute("SELECT id, uuid FROM modules ORDER BY id;")
+    }
+    assert rows == {
+        "wc_a_aaaaaaaa": "aaaaaaaa",
+        "wc_b_bbbbbbbb": "bbbbbbbb",
+        "fv_c_cccccccc": "cccccccc",
+    }
+
+
+def test_003_is_idempotent_when_run_against_already_migrated_db(tmp_path):
+    """Calling migrate() twice on a fresh DB does not error and leaves
+    every row's uuid intact. Backstops the `IF NOT EXISTS` guard on the
+    index and the no-op behavior of UPDATE WHERE uuid IS NULL when no
+    rows are NULL."""
+    db = tmp_path / "lib.db"
+    conn = sqlite3.connect(str(db))
+    conn.row_factory = sqlite3.Row
+
+    migrate(conn)
+    conn.execute(
+        "INSERT INTO modules(id, uuid, type, name, description, "
+        "category_id, tags, is_favorite, payload, version, created_at, "
+        "updated_at) VALUES ('wc_x_12345678', '12345678', 'wildcard', "
+        "'x', '', NULL, '[]', 0, '{}', 1, datetime('now'), datetime('now'));"
+    )
+    conn.commit()
+
+    # Re-running migrate is a no-op (version already == 3) — calling it
+    # twice in succession should NOT trip any "duplicate column" or
+    # uniqueness errors.
+    migrate(conn)
+    migrate(conn)
+
+    row = conn.execute(
+        "SELECT uuid FROM modules WHERE id = 'wc_x_12345678';"
+    ).fetchone()
+    assert row["uuid"] == "12345678"

tests/engine/db/test_module_repository.py

@@ -215,3 +215,90 @@ def test_create_rejects_unknown_type(wp_db):
             type="bogus", name="x", description="",
             category_id=None, tags=[], payload={},
         )
+
+
+def test_create_writes_uuid_column_matching_id_suffix(wp_db):
+    """create() must persist the uuid column. The value equals the trailing
+    8 hex chars of the generated id so existing rows (backfilled) and new
+    rows share one identity convention."""
+    repo = ModuleRepository(wp_db)
+    row = repo.create(
+        type="wildcard", name="color", description="",
+        category_id=None, tags=[], payload={"options": []},
+    )
+    assert "uuid" in row
+    assert len(row["uuid"]) == 8
+    assert all(c in "0123456789abcdef" for c in row["uuid"])
+    assert row["id"].endswith(f"_{row['uuid']}")
+
+
+def test_row_to_module_includes_payload_hash(wp_db):
+    """API responses (and drift detection) need payload_hash on every row.
+    Sourced from engine.modules.snapshot.payload_hash so drift comparison
+    is bit-identical to embedded snapshot hashes."""
+    repo = ModuleRepository(wp_db)
+    row = repo.create(
+        type="wildcard", name="x", description="",
+        category_id=None, tags=[], payload={"options": [{"value": "a"}]},
+    )
+    assert "payload_hash" in row
+    # 64-char SHA-256 hex
+    assert len(row["payload_hash"]) == 64
+    assert all(c in "0123456789abcdef" for c in row["payload_hash"])
+
+
+def test_get_by_uuid_returns_row(wp_db):
+    repo = ModuleRepository(wp_db)
+    created = repo.create(
+        type="wildcard", name="color", description="",
+        category_id=None, tags=[], payload={"options": []},
+    )
+    fetched = repo.get_by_uuid(created["uuid"])
+    assert fetched["id"] == created["id"]
+
+
+def test_get_by_uuid_raises_when_missing(wp_db):
+    repo = ModuleRepository(wp_db)
+    with pytest.raises(ModuleNotFound):
+        repo.get_by_uuid("00000000")
+
+
+def test_get_by_uuids_bulk_lookup_dedups_and_skips_missing(wp_db):
+    """Bulk lookup feeds the embed-bundle endpoint and the test-runner
+    lazy catalog. Returns rows in the same order callers passed uuids in;
+    missing uuids are silently skipped (caller decides how to surface)."""
+    repo = ModuleRepository(wp_db)
+    a = repo.create(type="wildcard", name="a", description="",
+                   category_id=None, tags=[], payload={"options": []})
+    b = repo.create(type="wildcard", name="b", description="",
+                   category_id=None, tags=[], payload={"options": []})
+    rows = repo.get_by_uuids([a["uuid"], "deadbeef", b["uuid"], a["uuid"]])
+    # Dedups (returns each uuid at most once), skips missing
+    assert len(rows) == 2
+    found_uuids = {r["uuid"] for r in rows}
+    assert found_uuids == {a["uuid"], b["uuid"]}
+
+
+def test_get_by_uuids_empty_input_returns_empty_list(wp_db):
+    repo = ModuleRepository(wp_db)
+    assert repo.get_by_uuids([]) == []
+
+
+def test_get_by_uuids_returns_rows_in_input_order(wp_db):
+    """The docstring promises input-order preservation. SQLite's IN
+    predicate makes no order guarantee, so the implementation must
+    re-order results against the (deduped) input list. Pin that
+    contract explicitly so a future contributor doesn't drop the
+    re-order step thinking it's redundant."""
+    repo = ModuleRepository(wp_db)
+    rows = []
+    for n in ("alpha", "bravo", "charlie", "delta"):
+        rows.append(repo.create(
+            type="wildcard", name=n, description="",
+            category_id=None, tags=[], payload={"options": []},
+        ))
+    # Request in reverse insertion order; result must come back in the
+    # SAME reverse order (not the natural SELECT order).
+    requested = [rows[3]["uuid"], rows[1]["uuid"], rows[2]["uuid"], rows[0]["uuid"]]
+    fetched = repo.get_by_uuids(requested)
+    assert [r["uuid"] for r in fetched] == requested

tests/engine/modules/test_resolve_context_isolation.py

@@ -0,0 +1,37 @@
+"""Pin the no-DB-fallback rule on get_module(uuid). Spec §2.5 / §2.10.
+
+If a future contributor adds `from engine.db import ...` here for an
+'ergonomic' DB fallback, this test trips and forces the discussion."""
+from __future__ import annotations
+
+import importlib
+import sys
+
+
+def test_engine_modules_init_does_not_import_db():
+    if "engine.modules" in sys.modules:
+        del sys.modules["engine.modules"]
+    importlib.import_module("engine.modules")
+    mod = sys.modules["engine.modules"]
+    forbidden = ("engine.db", "wp_api", "wp_nodes", "comfy_api")
+    for name in dir(mod):
+        attr = getattr(mod, name)
+        attr_module = getattr(attr, "__module__", "")
+        for prefix in forbidden:
+            assert not attr_module.startswith(prefix), (
+                f"engine.modules.__init__ leaks {prefix} via attribute "
+                f"{name!r} (came from {attr_module})"
+            )
+
+
+def test_get_module_returns_none_for_missing_uuid_without_db_call():
+    """get_module is a pure dict lookup. Never queries DB even when miss."""
+    from engine.modules import build_resolve_ctx
+    ctx_dict = {
+        "__wp_rng__": __import__("random").Random(0),
+        "__wp_warnings__": [],
+        "__wp_catalog__": {"aabbccdd": {"uuid": "aabbccdd"}},
+    }
+    ctx = build_resolve_ctx(ctx_dict, surface="wildcard")
+    assert ctx.get_module("aabbccdd") == {"uuid": "aabbccdd"}
+    assert ctx.get_module("nonexist") is None

tests/engine/modules/test_snapshot_walker.py

@@ -0,0 +1,181 @@
+"""Walker + SnapshotEntry tests. Mirrors spec §2.4–2.10 + §6 (5.5.0).
+
+The walker is the single shared lazy-walk implementation used by:
+- wp_api/test_runner.py:run_test (request-scoped roots)
+- wp_api/modules.py:embed_bundle (user pick roots)
+NEVER called from wp_nodes/context_node.py — graph runs use pre-walked
+snapshots embedded in workflow JSON."""
+from __future__ import annotations
+
+import importlib
+import sys
+
+
+def test_snapshot_entry_canonical_shape():
+    """SnapshotEntry has the 7 fields locked in spec §2.4. Other code
+    grep-asserts these names; never rename without updating callers."""
+    from engine.modules.snapshot import SnapshotEntry
+    # SnapshotEntry is a TypedDict — read its annotations
+    annotations = SnapshotEntry.__annotations__
+    assert set(annotations.keys()) == {
+        "snapshot_version", "uuid", "type", "name",
+        "payload", "payload_hash", "source",
+    }
+
+
+def test_engine_modules_snapshot_does_not_import_db_or_wp_api():
+    """Engine isolation: snapshot.py must be importable with no DB layer
+    in sys.modules. CLAUDE.md isolation rule — engine/ never knows about
+    wp_api/ or wp_nodes/. Any future contributor adding `from engine.db
+    import ...` here breaks the runtime separation."""
+    # Force fresh import to capture import-time side effects
+    if "engine.modules.snapshot" in sys.modules:
+        del sys.modules["engine.modules.snapshot"]
+    importlib.import_module("engine.modules.snapshot")
+    mod = sys.modules["engine.modules.snapshot"]
+    forbidden_prefixes = ("engine.db", "wp_api", "wp_nodes", "comfy_api")
+    for name in dir(mod):
+        attr = getattr(mod, name)
+        attr_module = getattr(attr, "__module__", "")
+        for prefix in forbidden_prefixes:
+            assert not attr_module.startswith(prefix), (
+                f"engine.modules.snapshot leaks {prefix} via attribute "
+                f"{name!r} (came from {attr_module})"
+            )
+
+
+# ---------------------------------------------------------------------------
+# Walker tests (Task 5)
+# ---------------------------------------------------------------------------
+
+def _make_module(uuid: str, *, type: str = "wildcard", name: str | None = None,
+                 payload: dict | None = None) -> dict:
+    """Test helper — builds a row matching what repo.get_by_uuid returns."""
+    p = payload or {"options": [{"value": "x", "weight": 1}]}
+    return {
+        "id": f"wc_{name or uuid}_{uuid}",
+        "uuid": uuid,
+        "type": type,
+        "name": name or uuid,
+        "payload": p,
+        "payload_hash": "deadbeef" * 8,  # actual hash not tested here
+    }
+
+
+def test_walk_single_module_no_refs_returns_one_entry():
+    from engine.modules.snapshot import walk_transitive_refs
+    catalog = {"a1111111": _make_module("a1111111")}
+    result = walk_transitive_refs(
+        ["a1111111"], fetch_module=catalog.get,
+    )
+    assert set(result.snapshots.keys()) == {"a1111111"}
+    assert result.snapshots["a1111111"]["source"] == {"kind": "user"}
+    assert result.walk_overflow == []
+
+
+def test_walk_follows_at_uuid_refs_marks_deps():
+    from engine.modules.snapshot import walk_transitive_refs
+    # outfit references @{color_uuid}
+    catalog = {
+        "aa111111": _make_module("aa111111", name="outfit", payload={
+            "options": [{"value": "@{bb222222} dress", "weight": 1}],
+        }),
+        "bb222222": _make_module("bb222222", name="color", payload={
+            "options": [{"value": "red", "weight": 1}],
+        }),
+    }
+    result = walk_transitive_refs(["aa111111"], fetch_module=catalog.get)
+    assert set(result.snapshots.keys()) == {"aa111111", "bb222222"}
+    assert result.snapshots["aa111111"]["source"] == {"kind": "user"}
+    assert result.snapshots["bb222222"]["source"] == {
+        "kind": "dep", "parent_uuids": ["aa111111"],
+    }
+
+
+def test_walk_records_missing_target_overflow():
+    from engine.modules.snapshot import walk_transitive_refs
+    catalog = {
+        "a1111111": _make_module("a1111111", payload={
+            "options": [{"value": "@{deadc0de}", "weight": 1}],
+        }),
+    }
+    result = walk_transitive_refs(["a1111111"], fetch_module=catalog.get)
+    # Walker returns the picked module but records the missing dep
+    assert "a1111111" in result.snapshots
+    assert "deadc0de" not in result.snapshots
+    assert {"uuid": "deadc0de", "reason": "missing_target"} in result.walk_overflow
+
+
+def test_walk_records_max_depth_overflow():
+    """Spec §2.10 + §6 — walker honors max_ref_depth=8 cap."""
+    from engine.modules.snapshot import walk_transitive_refs
+    # Build a 10-deep linear chain: a → b → c → ... → j
+    chain_uuids = [f"a{i:07x}" for i in range(10)]
+    catalog = {}
+    for i, u in enumerate(chain_uuids):
+        next_ref = f"@{{{chain_uuids[i+1]}}}" if i + 1 < len(chain_uuids) else "leaf"
+        catalog[u] = _make_module(u, payload={
+            "options": [{"value": next_ref, "weight": 1}],
+        })
+    result = walk_transitive_refs(
+        [chain_uuids[0]], fetch_module=catalog.get, max_depth=8,
+    )
+    # First 9 (depth 0..8 inclusive) are walked, 10th is overflow
+    assert len(result.snapshots) == 9
+    assert any(o["reason"] == "max_depth" for o in result.walk_overflow)
+
+
+def test_walk_handles_cycle_records_overflow():
+    """A → B → A. Walker terminates, records cycle, includes both."""
+    from engine.modules.snapshot import walk_transitive_refs
+    catalog = {
+        "a1111111": _make_module("a1111111", payload={
+            "options": [{"value": "@{b2222222}", "weight": 1}],
+        }),
+        "b2222222": _make_module("b2222222", payload={
+            "options": [{"value": "@{a1111111}", "weight": 1}],
+        }),
+    }
+    result = walk_transitive_refs(["a1111111"], fetch_module=catalog.get)
+    assert set(result.snapshots.keys()) == {"a1111111", "b2222222"}
+    assert any(o["reason"] == "cycle_detected" for o in result.walk_overflow)
+
+
+def test_walk_multi_parent_dep_lists_both_parents():
+    """A and C both ref B. B's source.parent_uuids contains both."""
+    from engine.modules.snapshot import walk_transitive_refs
+    catalog = {
+        "a1111111": _make_module("a1111111", payload={
+            "options": [{"value": "@{b2222222}", "weight": 1}],
+        }),
+        "c3333333": _make_module("c3333333", payload={
+            "options": [{"value": "@{b2222222}", "weight": 1}],
+        }),
+        "b2222222": _make_module("b2222222"),
+    }
+    result = walk_transitive_refs(
+        ["a1111111", "c3333333"], fetch_module=catalog.get,
+    )
+    assert result.snapshots["b2222222"]["source"]["kind"] == "dep"
+    assert sorted(
+        result.snapshots["b2222222"]["source"]["parent_uuids"]
+    ) == ["a1111111", "c3333333"]
+
+
+def test_walk_skips_non_wildcard_picks_at_root():
+    """Spec §2.7 — catalog only ever contains wildcards. If a non-wildcard
+    is in the roots list (e.g. user picked a Combine), walker emits it but
+    does NOT walk its payload for refs (combines aren't @{}-targets)."""
+    from engine.modules.snapshot import walk_transitive_refs
+    catalog = {
+        "dd111111": _make_module("dd111111", type="combine", payload={
+            "template": "$alpha and @{cc333333}",  # ref ignored
+        }),
+        "cc333333": _make_module("cc333333"),
+    }
+    result = walk_transitive_refs(["dd111111"], fetch_module=catalog.get)
+    # Combine present at root level (the picker still wants to embed it),
+    # but its @{} refs are NOT followed because non-wildcards do not
+    # participate in the catalog.
+    assert "dd111111" in result.snapshots
+    assert "cc333333" not in result.snapshots

tests/test_nodes_context.py

@@ -2,8 +2,6 @@
 
 import json
 
-import pytest
-
 from wp_nodes.context_node import WPContext
 from wp_nodes.types import ContextPayload
 
@@ -88,9 +86,13 @@ class TestWPContextExecute:
         assert result.context == {}
         assert result.debug["trace"] == []
 
-    def test_malformed_modules_raises(self):
-        with pytest.raises(ValueError):
-            WPContext.execute(seed=0, modules="{not json", upstream=None)
+    def test_malformed_modules_runs_empty(self):
+        # deserialize_node_input is robust: malformed JSON returns ([], {}, [])
+        # so the pipeline runs with an empty module list rather than crashing.
+        out = WPContext.execute(seed=0, modules="{not json", upstream=None)
+        result = out.values[0]
+        assert result.context == {}
+        assert result.debug["trace"] == []
 
     def test_downstream_overwrite_flows(self):
         first = WPContext.execute(

tests/wp_api/test_helpers.py

@@ -0,0 +1,36 @@
+"""extract_referenced_uuids — used by Test Runner to collect lazy-walk
+roots from incoming request bodies. Walks the whole nested structure
+because requests can put @{uuid} refs inside payload, instance overrides,
+or any other text-bearing field."""
+from __future__ import annotations
+
+from wp_api._helpers import extract_referenced_uuids
+
+
+def test_finds_refs_in_nested_string_values():
+    body = {
+        "type": "wildcard",
+        "payload": {
+            "options": [
+                {"value": "red @{aabbccdd} thing", "weight": 1},
+                {"value": "no ref here", "weight": 1},
+            ],
+        },
+        "instance": {"variable_binding": "color"},
+    }
+    assert extract_referenced_uuids(body) == {"aabbccdd"}
+
+
+def test_finds_multiple_refs_dedups():
+    body = {"v": "@{aaaaaaaa} and @{bbbbbbbb} and @{aaaaaaaa} again"}
+    assert extract_referenced_uuids(body) == {"aaaaaaaa", "bbbbbbbb"}
+
+
+def test_ignores_text_that_is_not_a_valid_ref():
+    body = {"v": "@{not-hex} @{toolong12345} @{TOOSHORT}"}
+    assert extract_referenced_uuids(body) == set()
+
+
+def test_empty_or_missing_payload_returns_empty_set():
+    assert extract_referenced_uuids({}) == set()
+    assert extract_referenced_uuids({"payload": None}) == set()

tests/wp_api/test_modules_endpoints.py

@@ -0,0 +1,160 @@
+"""Endpoint tests for embed-bundle + hashes + ModuleRow shape extension.
+Spec §4.2."""
+from __future__ import annotations
+
+import sqlite3
+
+import pytest
+from aiohttp import web
+
+from engine.db.migrations import migrate
+from engine.db.repositories import ModuleRepository
+from wp_api import modules as modules_api
+
+
+@pytest.fixture
+def app_with_db(tmp_path):
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    migrate(conn)
+    app = web.Application()
+    app["wp_db"] = conn
+    modules_api.register(app.router)
+    return app, conn
+
+
+async def test_embed_bundle_returns_picks_and_transitive_wildcards(
+    aiohttp_client, app_with_db,
+):
+    """Picking outfit (which @{}-refs color) returns:
+      - modules: [outfit_payload]                            (only the picks)
+      - snapshots: {outfit_uuid: ..., color_uuid: ...}       (picks + transitive wildcards)
+      - pickOrder: [outfit_uuid]
+      - walkOverflow: []"""
+    app, conn = app_with_db
+    repo = ModuleRepository(conn)
+    color = repo.create(
+        type="wildcard", name="color", description="",
+        category_id=None, tags=[], payload={
+            "options": [{"value": "red", "weight": 1}],
+        },
+    )
+    outfit_payload = {
+        "options": [{"value": f"@{{{color['uuid']}}} dress", "weight": 1}],
+    }
+    outfit = repo.create(
+        type="wildcard", name="outfit", description="",
+        category_id=None, tags=[], payload=outfit_payload,
+    )
+
+    client = await aiohttp_client(app)
+    resp = await client.post("/wp/api/modules/embed-bundle", json={
+        "uuids": [outfit["uuid"]],
+    })
+    assert resp.status == 200
+    body = await resp.json()
+    assert body["pickOrder"] == [outfit["uuid"]]
+    assert len(body["modules"]) == 1
+    assert body["modules"][0] == outfit_payload
+    assert set(body["snapshots"].keys()) == {outfit["uuid"], color["uuid"]}
+    assert body["walkOverflow"] == []
+
+
+async def test_embed_bundle_records_missing_target_in_walk_overflow(
+    aiohttp_client, app_with_db,
+):
+    app, conn = app_with_db
+    repo = ModuleRepository(conn)
+    a = repo.create(
+        type="wildcard", name="a", description="",
+        category_id=None, tags=[], payload={
+            "options": [{"value": "@{deadbeef}", "weight": 1}],
+        },
+    )
+    client = await aiohttp_client(app)
+    resp = await client.post("/wp/api/modules/embed-bundle", json={
+        "uuids": [a["uuid"]],
+    })
+    assert resp.status == 200
+    overflow = (await resp.json())["walkOverflow"]
+    assert any(o["uuid"] == "deadbeef" and o["reason"] == "missing_target"
+               for o in overflow)
+
+
+async def test_embed_bundle_rejects_non_list_uuids(aiohttp_client, app_with_db):
+    app, _ = app_with_db
+    client = await aiohttp_client(app)
+    resp = await client.post("/wp/api/modules/embed-bundle", json={
+        "uuids": "not-a-list",
+    })
+    assert resp.status == 400
+
+
+async def test_hashes_endpoint_returns_uuid_hash_map_for_wildcards_only(
+    aiohttp_client, app_with_db,
+):
+    """Drift-detection primitive. Returns {hashes: {uuid: hash}} keyed
+    by 8hex uuid. Wildcards-only because catalog only contains wildcards
+    (spec §2.7) — no point shipping hashes for kinds that never drift
+    embedded snapshots."""
+    app, conn = app_with_db
+    repo = ModuleRepository(conn)
+    wc = repo.create(
+        type="wildcard", name="x", description="",
+        category_id=None, tags=[], payload={"options": []},
+    )
+    cb = repo.create(
+        type="combine", name="y", description="",
+        category_id=None, tags=[], payload={"template": "$a"},
+    )
+    client = await aiohttp_client(app)
+    resp = await client.get("/wp/api/modules/hashes")
+    assert resp.status == 200
+    body = await resp.json()
+    assert wc["uuid"] in body["hashes"]
+    assert cb["uuid"] not in body["hashes"]  # combines excluded
+    assert body["hashes"][wc["uuid"]] == wc["payload_hash"]
+
+
+# ── ModuleRow shape extension (spec §4.2 — Task 11) ──────────────────────
+# Task 2 already wired `_row_to_module` to include `uuid` and `payload_hash`
+# on every returned row. These tests pin that contract at the API boundary
+# so a future refactor (e.g. swapping the row helper, adding a slim list
+# variant) can't accidentally drop either field. Pure regression coverage.
+
+
+async def test_list_modules_response_includes_uuid_and_payload_hash(
+    aiohttp_client, app_with_db,
+):
+    app, conn = app_with_db
+    repo = ModuleRepository(conn)
+    repo.create(
+        type="wildcard", name="x", description="",
+        category_id=None, tags=[], payload={"options": []},
+    )
+    client = await aiohttp_client(app)
+    resp = await client.get("/wp/api/modules")
+    assert resp.status == 200
+    items = (await resp.json())["items"]
+    assert len(items) == 1
+    assert "uuid" in items[0]
+    assert "payload_hash" in items[0]
+    assert len(items[0]["uuid"]) == 8
+    assert len(items[0]["payload_hash"]) == 64
+
+
+async def test_get_module_response_includes_uuid_and_payload_hash(
+    aiohttp_client, app_with_db,
+):
+    app, conn = app_with_db
+    repo = ModuleRepository(conn)
+    row = repo.create(
+        type="wildcard", name="x", description="",
+        category_id=None, tags=[], payload={"options": []},
+    )
+    client = await aiohttp_client(app)
+    resp = await client.get(f"/wp/api/modules/{row['id']}")
+    assert resp.status == 200
+    data = await resp.json()
+    assert data["uuid"] == row["uuid"]
+    assert data["payload_hash"] == row["payload_hash"]

tests/wp_api/test_test_runner.py

@@ -0,0 +1,92 @@
+"""End-to-end Test Runner tests. Spec §6 / 5.5.1.
+
+These run the real aiohttp handler against a real in-memory DB. After
+this task, posting a wildcard whose option references @{B_uuid} resolves
+B and produces the expanded value, instead of leaking raw @{B} text."""
+from __future__ import annotations
+
+import sqlite3
+
+import pytest
+from aiohttp import web
+
+from engine.db.migrations import migrate
+from engine.db.repositories import ModuleRepository
+from wp_api import test_runner as test_runner_mod
+
+
+@pytest.fixture
+def app_with_db(tmp_path):
+    """Build an aiohttp app with a freshly-migrated in-memory DB and the
+    test_runner endpoint wired up."""
+    conn = sqlite3.connect(":memory:")
+    conn.row_factory = sqlite3.Row
+    migrate(conn)
+    app = web.Application()
+    app["wp_db"] = conn
+    app.router.add_post("/wp/api/test", test_runner_mod.run_test)
+    return app, conn
+
+
+async def test_nested_wildcard_ref_resolves_via_catalog(aiohttp_client, app_with_db):
+    """Library has wildcards 'color' (UUID c0000001) and 'outfit' (UUID
+    o0000001) where outfit.options[0].value = "@{c0000001} dress". Test
+    Runner must load color into the catalog and resolve the ref so the
+    sample is "red dress" or "blue dress", never raw "@{c0000001} dress"."""
+    app, conn = app_with_db
+    repo = ModuleRepository(conn)
+    color = repo.create(
+        type="wildcard", name="color", description="",
+        category_id=None, tags=[],
+        payload={"options": [
+            {"value": "red", "weight": 1},
+            {"value": "blue", "weight": 1},
+        ]},
+    )
+    outfit_payload = {
+        "options": [
+            {"value": f"@{{{color['uuid']}}} dress", "weight": 1},
+        ],
+    }
+    repo.create(
+        type="wildcard", name="outfit", description="",
+        category_id=None, tags=[], payload=outfit_payload,
+    )
+
+    client = await aiohttp_client(app)
+    resp = await client.post("/wp/api/test", json={
+        "type": "wildcard",
+        "payload": outfit_payload,
+        "instance": {"variable_binding": "outfit"},
+        "samples": 5,
+    })
+    assert resp.status == 200
+    body = await resp.json()
+    samples = body["samples"]
+    assert len(samples) == 5
+    for s in samples:
+        assert s in ("red dress", "blue dress"), (
+            f"@{{{color['uuid']}}} did not resolve — got raw text {s!r}"
+        )
+
+
+async def test_unknown_ref_emits_warning_does_not_crash(aiohttp_client, app_with_db):
+    """A missing target produces a graceful warning + raw token in the
+    output (lenient mode) — Test Runner never strict-raises."""
+    app, conn = app_with_db
+    payload = {
+        "options": [{"value": "@{deadbeef} thing", "weight": 1}],
+    }
+    client = await aiohttp_client(app)
+    resp = await client.post("/wp/api/test", json={
+        "type": "wildcard",
+        "payload": payload,
+        "instance": {"variable_binding": "x"},
+        "samples": 1,
+    })
+    assert resp.status == 200
+    body = await resp.json()
+    warnings = body.get("warnings", [])
+    assert any("deadbeef" in str(w) for w in warnings), (
+        f"expected 'Unknown wildcard ref' warning for deadbeef, got: {warnings}"
+    )

tests/wp_nodes/test_context_node_catalog.py

@@ -0,0 +1,93 @@
+"""Tests for catalog injection at WP_Context node execution + the
+backwards-compat path for old workflow JSON. Spec §4.4 / §6 / 5.5.3."""
+from __future__ import annotations
+
+from wp_nodes.types import deserialize_node_input
+
+
+def test_deserialize_returns_modules_with_empty_snapshots_for_legacy_input():
+    """Old workflow JSON had only `modules` (a list). The deserializer
+    returns `(modules, snapshots, pickOrder)` where snapshots defaults to
+    empty dict and pickOrder defaults to empty list — backwards compat,
+    spec §4.5."""
+    legacy = {"modules": [
+        {"type": "fixed_values", "name": "x", "payload": {"values": []}},
+    ]}
+    modules, snapshots, pick_order = deserialize_node_input(legacy)
+    assert len(modules) == 1
+    assert snapshots == {}
+    assert pick_order == []
+
+
+def test_deserialize_extracts_snapshots_and_pickorder_from_new_input():
+    new_shape = {
+        "modules": [
+            {"type": "wildcard", "name": "outfit",
+             "payload": {"options": [{"value": "x"}]}},
+        ],
+        "snapshots": {
+            "ou111111": {
+                "snapshot_version": 1, "uuid": "ou111111", "type": "wildcard",
+                "name": "outfit",
+                "payload": {"options": [{"value": "x"}]},
+                "payload_hash": "h" * 64,
+                "source": {"kind": "user"},
+            },
+        },
+        "pickOrder": ["ou111111"],
+        "snapshotVersion": 1,
+    }
+    modules, snapshots, pick_order = deserialize_node_input(new_shape)
+    assert len(modules) == 1
+    assert "ou111111" in snapshots
+    assert pick_order == ["ou111111"]
+
+
+def _ctx_capture():
+    """Spy that captures the ctx dict passed to PipelineEngine.run()."""
+    captured: dict = {}
+
+    class _FakePipeline:
+        def run(self, modules, *, ctx, seed):
+            captured.clear()
+            captured.update(ctx)
+            return ctx
+    return _FakePipeline, captured
+
+
+def test_execute_injects_catalog_from_snapshots_field(monkeypatch):
+    """Snapshots from the input become ctx['__wp_catalog__'] exactly once
+    at the top of execute. Spec §2.6."""
+    from wp_nodes import context_node as cn
+
+    fake_cls, captured = _ctx_capture()
+    monkeypatch.setattr(cn, "PipelineEngine", fake_cls)
+
+    snapshots = {
+        "ou111111": {"snapshot_version": 1, "uuid": "ou111111",
+                     "type": "wildcard", "name": "outfit",
+                     "payload": {"options": []}, "payload_hash": "h" * 64,
+                     "source": {"kind": "user"}},
+    }
+    cn.WPContext.execute(
+        seed=42,
+        modules={"modules": [], "snapshots": snapshots, "pickOrder": ["ou111111"]},
+        upstream=None,
+    )
+    assert captured["__wp_catalog__"] == snapshots
+
+
+def test_execute_defaults_to_empty_catalog_for_old_workflow_json(monkeypatch):
+    """Old JSON without `snapshots` field — catalog stays {} so resolver
+    emits "Unknown ref" warning rather than crashing. Spec §4.5."""
+    from wp_nodes import context_node as cn
+
+    fake_cls, captured = _ctx_capture()
+    monkeypatch.setattr(cn, "PipelineEngine", fake_cls)
+
+    cn.WPContext.execute(
+        seed=0,
+        modules={"modules": []},  # no snapshots
+        upstream=None,
+    )
+    assert captured["__wp_catalog__"] == {}

wp_api/_helpers.py

@@ -7,14 +7,26 @@ from typing import Any
 from aiohttp import web
 
 from engine.db.connection import get_connection
+from engine.syntax.tokenize import tokenize_text
+from engine.syntax.types import TokenKind
 
 
 @contextmanager
-def db_session(_request: web.Request):
-    """Yield a fresh SQLite connection that is auto-closed on exit.
+def db_session(request: web.Request):
+    """Yield a SQLite connection that is auto-closed on exit.
+
+    If the aiohttp app has a ``wp_db`` key (injected by tests or the
+    application startup), that connection is yielded directly and is NOT
+    closed on exit (lifetime owned by the injector). Otherwise, a fresh
+    file-based connection is opened via ``get_connection()`` and closed
+    when the context manager exits.
 
     SQLite WAL mode + per-request connections avoid cross-task locking.
     """
+    injected = request.app.get("wp_db") if hasattr(request, "app") else None
+    if injected is not None:
+        yield injected
+        return
     conn = get_connection()
     try:
         yield conn
@@ -28,3 +40,31 @@ def json_ok(data: Any, *, status: int = 200) -> web.Response:
 
 def json_error(message: str, *, status: int) -> web.Response:
     return web.json_response({"error": message}, status=status)
+
+
+def extract_referenced_uuids(payload: Any) -> set[str]:
+    """Recursively scan strings inside `payload` for `@{8hex}` refs and
+    return the set of unique uuids referenced.
+
+    Used by the SPA Test Runner to determine which wildcards must be
+    loaded from the library before resolution can proceed (lazy catalog,
+    spec §2.10). Reuses the engine tokenizer so the regex shape stays
+    locked in one place."""
+    refs: set[str] = set()
+
+    def _scan(value: Any) -> None:
+        if isinstance(value, str):
+            for tok in tokenize_text(value):
+                if tok.kind == TokenKind.REF:
+                    uuid = (tok.meta or {}).get("uuid")
+                    if isinstance(uuid, str):
+                        refs.add(uuid)
+        elif isinstance(value, dict):
+            for v in value.values():
+                _scan(v)
+        elif isinstance(value, list):
+            for item in value:
+                _scan(item)
+
+    _scan(payload)
+    return refs

wp_api/modules.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 from aiohttp import web
 
 from engine.db.repositories import ModuleNotFound, ModuleRepository
-from engine.modules.snapshot import freeze_snapshot, payload_hash
+from engine.modules.snapshot import freeze_snapshot, payload_hash, walk_transitive_refs
 from wp_api._helpers import db_session, json_error, json_ok
 
 _UPDATABLE_FIELDS = (
@@ -162,10 +162,80 @@ async def toggle_favorite(request: web.Request) -> web.Response:
     return json_ok(updated)
 
 
+async def list_hashes(request: web.Request) -> web.Response:
+    """Lightweight bulk hash fetch for drift detection. Spec §4.2.
+
+    Returns wildcards-only because catalog never contains other kinds
+    (spec §2.7). Lightweight (no payload, no metadata) so the SPA can
+    poll on every workflow load without measurable cost."""
+    with db_session(request) as conn:
+        rows = ModuleRepository(conn).list(type="wildcard")
+    return json_ok({
+        "hashes": {row["uuid"]: row["payload_hash"] for row in rows},
+    })
+
+
+async def embed_bundle(request: web.Request) -> web.Response:
+    """Lazy walk for SPA library picker. Spec §4.2.
+
+    Body: {"uuids": [str, ...]}. Server fetches each picked uuid + walks
+    transitive @{} refs through wildcard payloads only. Returns split
+    response so the client doesn't have to filter:
+      - modules: list of picked-module payloads in input order (executor's view)
+      - snapshots: dict[uuid, SnapshotEntry] of wildcards reachable from picks
+      - pickOrder: list of uuid strings, original input order
+      - walkOverflow: [{uuid, reason}] for cycle/depth/missing
+    """
+    try:
+        body = await request.json()
+    except Exception:
+        return json_error("invalid JSON body", status=400)
+    if not isinstance(body, dict):
+        return json_error("body must be a JSON object", status=400)
+
+    uuids = body.get("uuids")
+    if not isinstance(uuids, list) or not all(isinstance(u, str) for u in uuids):
+        return json_error("uuids must be a list of strings", status=400)
+
+    with db_session(request) as conn:
+        repo = ModuleRepository(conn)
+
+        def _fetch(uuid: str) -> dict | None:
+            try:
+                return repo.get_by_uuid(uuid)
+            except Exception:
+                return None
+
+        walk = walk_transitive_refs(uuids, fetch_module=_fetch)
+        # Picks (executor's view): keep input order, drop any uuids that
+        # weren't found in DB.
+        picks: list[dict] = []
+        for uuid in uuids:
+            entry = walk.snapshots.get(uuid)
+            if entry is not None:
+                picks.append(entry["payload"])
+
+        # Catalog (resolver's view): only wildcards. Spec §2.7.
+        catalog = {
+            uuid: entry
+            for uuid, entry in walk.snapshots.items()
+            if entry["type"] == "wildcard"
+        }
+
+    return json_ok({
+        "modules": picks,
+        "snapshots": catalog,
+        "pickOrder": uuids,
+        "walkOverflow": walk.walk_overflow,
+    })
+
+
 def register(router) -> None:
     router.add_get("/wp/api/modules", list_modules)
     router.add_post("/wp/api/modules", create_module)
+    router.add_get("/wp/api/modules/hashes", list_hashes)
     router.add_post("/wp/api/modules/match", match_module)
+    router.add_post("/wp/api/modules/embed-bundle", embed_bundle)
     router.add_get("/wp/api/modules/{id}", get_module)
     router.add_put("/wp/api/modules/{id}", update_module)
     router.add_delete("/wp/api/modules/{id}", delete_module)

wp_api/test_runner.py

@@ -6,23 +6,46 @@ from collections import Counter
 
 from aiohttp import web
 
+from engine.db.repositories import ModuleRepository
 from engine.modules.dispatcher import UnknownModuleType, resolve_module
-from wp_api._helpers import json_error, json_ok
+from engine.modules.snapshot import walk_transitive_refs
+from wp_api._helpers import db_session, extract_referenced_uuids, json_error, json_ok
 
 _MAX_SAMPLES = 10000
 
 
-def _make_test_ctx() -> dict:
-    """Build a minimal pipeline ctx for the test-runner endpoint.
+def _build_request_catalog(request: web.Request, body: dict) -> dict[str, dict]:
+    """Lazy-walk the request body for `@{8hex}` refs and load just the
+    referenced wildcards (plus transitive deps) from the DB.
 
-    Uses a fresh random.Random (seeded from os entropy) so each /wp/api/test
-    request is independently sampled. The endpoint never needs a fixed seed —
-    it samples N times and returns the distribution.
-    """
+    Spec §2.10 — never the full library. Run ONCE per request; the catalog
+    is stable across all sample iterations within the same `run_test` call,
+    so callers should hoist this above their per-sample loop and reuse the
+    returned dict for every `__wp_catalog__` injection."""
+    roots = extract_referenced_uuids(body.get("payload"))
+    roots |= extract_referenced_uuids(body.get("instance"))
+    if not roots:
+        return {}
+    with db_session(request) as conn:
+        repo = ModuleRepository(conn)
+
+        def _fetch(uuid: str) -> dict | None:
+            try:
+                return repo.get_by_uuid(uuid)
+            except Exception:
+                return None
+
+        return walk_transitive_refs(roots, fetch_module=_fetch).snapshots
+
+
+def _make_sample_ctx(catalog: dict[str, dict]) -> dict:
+    """Per-sample ctx — fresh `__wp_rng__` and `__wp_warnings__`, but the
+    `__wp_catalog__` reference is shared across all samples (the resolver
+    treats it as read-only)."""
     return {
         "__wp_rng__": random.Random(),
         "__wp_warnings__": [],
-        "__wp_catalog__": {},
+        "__wp_catalog__": catalog,
     }
 
 
@@ -51,10 +74,17 @@ async def run_test(request: web.Request) -> web.Response:
         "payload": body["payload"],
         "instance": body["instance"],
     }
+    # Build the runtime catalog ONCE per request (§2.10 lazy walk). The
+    # walked dict is stable across all sample iterations — only the rng
+    # and warnings list need to be fresh per sample.
+    catalog = _build_request_catalog(request, body)
     results: list[dict[str, str]] = []
+    all_warnings: list[dict] = []
     try:
         for _ in range(samples):
-            results.append(resolve_module(snap, ctx=_make_test_ctx()))
+            ctx = _make_sample_ctx(catalog)
+            results.append(resolve_module(snap, ctx=ctx))
+            all_warnings.extend(ctx["__wp_warnings__"])
     except UnknownModuleType as e:
         return json_error(f"unknown module type: {e.args[0]!r}", status=400)
 
@@ -63,7 +93,30 @@ async def run_test(request: web.Request) -> web.Response:
         for v in r.values():
             histogram[v] += 1
 
-    return json_ok({"results": results, "histogram": dict(histogram)})
+    # Deduplicate warnings by (type, message) to avoid N*samples noise.
+    seen: set[tuple] = set()
+    deduped_warnings: list[dict] = []
+    for w in all_warnings:
+        key = (w.get("type"), w.get("message"))
+        if key not in seen:
+            seen.add(key)
+            deduped_warnings.append(w)
+
+    # Extract flat sample values for the SPA Test Runner display. Each
+    # resolve_module call returns {binding: value}; we surface the values.
+    sample_values: list[str] = []
+    for r in results:
+        for v in r.values():
+            sample_values.append(v)
+        if not r:
+            sample_values.append("")
+
+    return json_ok({
+        "results": results,
+        "histogram": dict(histogram),
+        "samples": sample_values,        # NEW: additive top-level
+        "warnings": deduped_warnings,    # NEW: additive top-level
+    })
 
 
 def register(router) -> None:

wp_nodes/context_node.py

@@ -7,7 +7,7 @@ from wp_nodes.types import (
     ContextModulesInput,
     PipelineContext,
     build_payload,
-    deserialize_modules,
+    deserialize_node_input,
 )
 
 
@@ -40,8 +40,17 @@ class WPContext(io.ComfyNode):
         upstream_ctx: dict = upstream.context if upstream is not None else {}
         upstream_debug: dict = upstream.debug if upstream is not None else {}
 
-        module_list = deserialize_modules(modules)
+        # Three orthogonal concerns from one widget value (spec §4.4):
+        #   - module_list:  what the pipeline executes (any kind, ordered)
+        #   - snapshots:    catalog for @{} ref resolution (wildcards only)
+        #   - pickOrder:    explicit user picks (UI-only, not used here)
+        module_list, snapshots, _pick_order = deserialize_node_input(modules)
+
         ctx: dict = dict(upstream_ctx)
+        # Inject catalog ONCE at the top of execute, before pipeline run.
+        # Spec §2.6 — never re-inject mid-run.
+        ctx["__wp_catalog__"] = snapshots
+
         ctx = PipelineEngine().run(module_list, ctx=ctx, seed=int(seed))
 
         payload = build_payload(ctx, upstream_debug=upstream_debug, seed=int(seed))

wp_nodes/types.py

@@ -124,6 +124,57 @@ def deserialize_modules(payload: str) -> list[Module]:
     return [module_from_dict(entry) for entry in raw_modules]
 
 
+def deserialize_node_input(
+    raw: "dict[str, Any] | str | None",
+) -> "tuple[list[dict[str, Any]], dict[str, Any], list[str]]":
+    """Parse a WP_Context node's ``modules`` widget value into the three
+    parallel concerns the engine needs: an executable module list, a
+    catalog snapshot dict, and the user's explicit pick order.
+
+    Spec §4.4. Backwards-compat: old workflows have no ``snapshots`` /
+    ``pickOrder`` fields — those default to empty (catalog stays {}, no
+    @{} resolution, raw token leaks in output, current behavior).
+
+    Returns the modules portion as a ``list[dict]`` rather than the typed
+    ``list[Module]`` that ``deserialize_modules`` produces. Two reasons:
+    the legacy converter ``module_from_dict`` requires an ``id`` field
+    and only knows how to type ``fixed_values`` (Module is currently a
+    TypeAlias for FixedValueModule), so SPA-shaped wildcard / combine /
+    derivation entries would all be rejected. ``PipelineEngine.run``
+    already accepts dicts directly (engine/pipeline.py:57) — coercing
+    them through ``coerce_legacy_module`` at execute time — so the raw-
+    dict path is the safe one for the new SPA snapshot shape.
+
+    Robust to malformed input: any failure path returns ``([], {}, [])``
+    so graph runs never crash on a broken widget value.
+    """
+    if raw is None:
+        return [], {}, []
+    if isinstance(raw, str):
+        try:
+            data = json.loads(raw)
+        except Exception:
+            return [], {}, []
+    else:
+        data = raw
+    if not isinstance(data, dict):
+        return [], {}, []
+
+    raw_modules = data.get("modules")
+    modules: list[dict[str, Any]] = (
+        list(raw_modules) if isinstance(raw_modules, list) else []
+    )
+    snapshots_raw = data.get("snapshots")
+    snapshots: dict[str, Any] = (
+        snapshots_raw if isinstance(snapshots_raw, dict) else {}
+    )
+    pick_order_raw = data.get("pickOrder")
+    pick_order: list[str] = (
+        pick_order_raw if isinstance(pick_order_raw, list) else []
+    )
+    return modules, snapshots, pick_order
+
+
 def build_payload(
     ctx: dict[str, Any],
     upstream_debug: dict[str, Any],