From 00e4386833747c6f424a081d4f7ba85d32773692 Mon Sep 17 00:00:00 2001
From: ysun67 <ysun67@syr.edu>
Date: Tue, 19 May 2026 12:23:57 -0400
Subject: [PATCH] A3-2: remove deprecated ExecutePipeline.work_stealing bool
 field

Per docs/phase_a3_remove_deprecated_bool_fields.md section 3.2,
retire the legacy `work_stealing: bool` shadow field across MIR,
HIR, DSL, and runner-side consumers. The typed `WorkStealing()`
pragma (with its `WSScope` wrap op materialized by
`materialize_work_stealing`) becomes the sole emission driver.

Surface changes:
- mir.ExecutePipeline: drop `work_stealing` field.
- HirRuleVariant: drop `work_stealing` field; hir/plan.py and
  hir/lower.py stop propagating it.
- dsl.PlanEntry: drop `work_stealing`; Rule.with_plan loses the
  kwarg; `_BUILTIN_BOOL_SHADOW_PRAGMAS` loses the WorkStealing
  entry. `Rule.with_pragma(WorkStealing())` is the sole DSL surface.
- materialize_work_stealing: drop the `if op.work_stealing: return`
  short-circuit; the wrap step is now unconditional.
- mir/passes.py: add public `ep_has_work_stealing(ep)` helper
  (typed-pragma + WSScope-wrap presence check) that orchestrator,
  complete_runner, batchfile, and the concurrent-write predicate
  consume in place of the deleted `ep.work_stealing` read.
- codegen/cuda/context.py: drop unused `is_work_stealing` field.
- sorted_array/lowerings: `lower_scan_pipeline` now unwraps trailing
  `WSScope(InsertInto)` while flipping `ctx.ws_enabled=True` for the
  call, so the legacy `_lower_insert_into` body keeps emitting the
  WS-count shape byte-equivalent to today. The C4 DEAD CODE NOTE
  block is retired (D19 ratchet cap drops 8 -> 7).
- mir/print.py: WSScope unwraps to its inner during s-expression
  printing so existing MIR goldens remain stable.

Test updates: the WS-tagged integration fixtures and the WS e2e
test migrate to `with_pragma(WorkStealing())`; the parallel
atomic_ws plugin byte-equivalence test picks the WS EP via
`ep_has_work_stealing` (wrap-op presence) instead of the deleted
bool. All 253 jit-batch + 276 runner byte-equivalence fixtures
remain green.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 examples/polonius_test.py                     |  6 +-
 src/srdatalog/dsl.py                          | 18 ++--
 src/srdatalog/ir/codegen/cuda/batchfile.py    |  5 +-
 .../ir/codegen/cuda/complete_runner.py        |  5 +-
 src/srdatalog/ir/codegen/cuda/context.py      |  1 -
 src/srdatalog/ir/codegen/cuda/orchestrator.py | 12 ++-
 .../dialects/parallel/atomic_ws/__init__.py   | 10 +-
 .../atomic_ws/pragmas/work_stealing.py        | 59 +++++-------
 .../sorted_array/lowerings/__init__.py        | 41 +++++---
 src/srdatalog/ir/hir/lower.py                 |  9 +-
 src/srdatalog/ir/hir/plan.py                  | 11 +--
 src/srdatalog/ir/hir/types.py                 |  1 -
 src/srdatalog/ir/mir/passes.py                | 33 ++++++-
 src/srdatalog/ir/mir/print.py                 | 10 ++
 src/srdatalog/ir/mir/types.py                 | 19 ++--
 tests/test_codegen_batchfile.py               | 10 +-
 tests/test_cuda_context.py                    |  4 +-
 .../test_discipline_obsolete_code_ratchet.py  | 10 +-
 tests/test_hir_user_plan.py                   |  4 +-
 tests/test_integration_lsqb_q6_nosj.py        |  7 +-
 tests/test_parallel_atomic_ws_as_plugin.py    | 44 +++++----
 tests/test_pragma_work_stealing_end_to_end.py | 95 ++++++-------------
 22 files changed, 226 insertions(+), 188 deletions(-)

diff --git a/examples/polonius_test.py b/examples/polonius_test.py
index 947e9f4..1f35c0d 100644
--- a/examples/polonius_test.py
+++ b/examples/polonius_test.py
@@ -7,6 +7,9 @@
 from __future__ import annotations
 
 from srdatalog.dsl import SPLIT, Filter, Program, Relation, Var
+from srdatalog.ir.dialects.parallel.atomic_ws.pragmas.work_stealing import (
+  WorkStealing,
+)
 
 # ----- Relations ----------------------------------------------
 
@@ -396,7 +399,8 @@ def build_poloniusdb_program() -> Program:
         )
       )
       .named('subset_trans')
-      .with_plan(delta=0, work_stealing=True),
+      .with_plan(delta=0)
+      .with_pragma(WorkStealing()),
       (
         subset(origin1, origin2, point2)
         <= subset(origin1, origin2, point1)
diff --git a/src/srdatalog/dsl.py b/src/srdatalog/dsl.py
index be20464..a6c78a1 100644
--- a/src/srdatalog/dsl.py
+++ b/src/srdatalog/dsl.py
@@ -340,9 +340,11 @@ class PlanEntry:
 
   The pragma flags flow through to HirRuleVariant so codegen sees them:
     - fanout          -> fan-out work-stealing for Cartesian products
-    - work_stealing   -> mid-level work-stealing (task queue + steal loop)
     - block_group     -> block-group work partitioning
     - dedup_hash      -> GPU hash table for in-kernel existential dedup
+  Mid-level work-stealing is no longer a `with_plan(...)` keyword:
+  it migrated to `Rule.with_pragma(WorkStealing())` in C4 and the
+  shadow bool field was retired in A3-2.
   `balanced_root` / `balanced_sources` drive balanced partitioning for
   skewed joins (not yet lowered in Python).
 
@@ -360,7 +362,6 @@ class PlanEntry:
   var_order: tuple[str, ...] = ()
   clause_order: tuple[int, ...] = ()
   fanout: bool = False
-  work_stealing: bool = False
   block_group: bool = False
   dedup_hash: bool = False
   balanced_root: tuple[str, ...] = ()
@@ -414,7 +415,6 @@ def with_plan(
     var_order: tuple[str, ...] | list[str] | None = None,
     clause_order: tuple[int, ...] | list[int] | None = None,
     fanout: bool = False,
-    work_stealing: bool = False,
     block_group: bool = False,
     dedup_hash: bool = False,
     balanced_root: tuple[str, ...] | list[str] | None = None,
@@ -422,13 +422,15 @@ def with_plan(
   ) -> Rule:
     '''Append a single PlanEntry. Can be called multiple times to add
     entries for different deltas (or use .with_plans(entries) to replace).
+
+    Note: the legacy `work_stealing=True` keyword was retired in
+    A3-2 — use `.with_pragma(WorkStealing())` instead.
     '''
     entry = PlanEntry(
       delta=delta,
       var_order=tuple(var_order) if var_order is not None else (),
       clause_order=tuple(clause_order) if clause_order is not None else (),
       fanout=fanout,
-      work_stealing=work_stealing,
       block_group=block_group,
       dedup_hash=dedup_hash,
       balanced_root=tuple(balanced_root) if balanced_root is not None else (),
@@ -750,10 +752,10 @@ def _validate_pragma_registered(pragma: Pragma) -> None:
     'srdatalog.ir.dialects.relation.sorted_array.pragmas.dedup_hash.DedupHash',
     'dedup_hash',
   ),
-  (
-    'srdatalog.ir.dialects.parallel.atomic_ws.pragmas.work_stealing.WorkStealing',
-    'work_stealing',
-  ),
+  # A3-2: `WorkStealing` shadow entry removed alongside the
+  # `PlanEntry.work_stealing` / `mir.ExecutePipeline.work_stealing`
+  # field deletion. The typed pragma is now the sole driver; see
+  # `docs/phase_a3_remove_deprecated_bool_fields.md` §3.2.
   # C6: `FanOut` shadows `PlanEntry.fanout` -> `variant.fanout` ->
   # `ep.use_fan_out`. Same dual-write contract as `DedupHash`:
   # `with_pragma(FanOut())` sets both `fanout=True` on each plan
diff --git a/src/srdatalog/ir/codegen/cuda/batchfile.py b/src/srdatalog/ir/codegen/cuda/batchfile.py
index 99ab8f1..65eea30 100644
--- a/src/srdatalog/ir/codegen/cuda/batchfile.py
+++ b/src/srdatalog/ir/codegen/cuda/batchfile.py
@@ -30,6 +30,7 @@
 )
 from srdatalog.ir.codegen.cuda.schema import SchemaDefinition
 from srdatalog.ir.hir.types import Version
+from srdatalog.ir.mir.passes import ep_has_work_stealing
 
 # -----------------------------------------------------------------------------
 # Prelude
@@ -360,7 +361,7 @@ def generate_runner(
       ctx.output_name = output_var
   full += generate_pipeline(mutable_pipeline, ctx)
 
-  if pipeline.work_stealing:
+  if ep_has_work_stealing(pipeline):
     full += "    // TODO: Implement work-stealing logic (jit_kernel.nim 1084-1589)\n"
 
   # -- LaunchParams struct --
@@ -386,7 +387,7 @@ def generate_runner(
       uint64_t work_per_warp = 0;'''
   for i, _ in enumerate(pipeline.dest_specs):
     full += f"\n    uint32_t old_size_{i} = 0;"
-  if pipeline.work_stealing:
+  if ep_has_work_stealing(pipeline):
     full += ""  # jitWSLaunchParamsFields() TODO
   full += "\n  };\n\n"
 
diff --git a/src/srdatalog/ir/codegen/cuda/complete_runner.py b/src/srdatalog/ir/codegen/cuda/complete_runner.py
index d039d73..039ebf2 100644
--- a/src/srdatalog/ir/codegen/cuda/complete_runner.py
+++ b/src/srdatalog/ir/codegen/cuda/complete_runner.py
@@ -43,6 +43,7 @@
   has_tiled_cartesian_eligible,
 )
 from srdatalog.ir.codegen.cuda.plugin import plugin_gen_host_view_setup, plugin_view_count
+from srdatalog.ir.mir.passes import ep_has_work_stealing
 
 # Pure-template phase emitters now live in the dialect's runner module.
 # Local aliases preserve the legacy call sites until the rest of this
@@ -922,7 +923,7 @@ def gen_complete_runner(
     )
 
   # Feature-flag guards: not covered in Phase 2 baseline port.
-  if node.work_stealing:
+  if ep_has_work_stealing(node):
     raise NotImplementedError("gen_complete_runner: work_stealing not yet ported")
   if has_balanced_scan(node.pipeline):
     raise NotImplementedError("gen_complete_runner: balanced-scan runner not yet ported")
@@ -958,7 +959,7 @@ def gen_complete_runner(
   # can't interleave safely with concurrent kernels into the same region.
   tiled_cartesian_eligible = (
     has_tiled_cartesian_eligible(mutable_pipe)
-    and not node.work_stealing
+    and not ep_has_work_stealing(node)
     and not node.block_group
     and not node.dedup_hash
     and not is_count
diff --git a/src/srdatalog/ir/codegen/cuda/context.py b/src/srdatalog/ir/codegen/cuda/context.py
index 9e54b27..307c5a8 100644
--- a/src/srdatalog/ir/codegen/cuda/context.py
+++ b/src/srdatalog/ir/codegen/cuda/context.py
@@ -189,7 +189,6 @@ class RunnerGenState:
   dest_arities: list[int] = field(default_factory=list)
   total_view_count: int = 0
   is_balanced: bool = False
-  is_work_stealing: bool = False
   is_block_group: bool = False
   is_dedup_hash: bool = False
   is_count: bool = False
diff --git a/src/srdatalog/ir/codegen/cuda/orchestrator.py b/src/srdatalog/ir/codegen/cuda/orchestrator.py
index 2e1e8a3..2996841 100644
--- a/src/srdatalog/ir/codegen/cuda/orchestrator.py
+++ b/src/srdatalog/ir/codegen/cuda/orchestrator.py
@@ -36,6 +36,7 @@
 
 import srdatalog.ir.mir.types as m
 from srdatalog.ir.hir.types import Version
+from srdatalog.ir.mir.passes import ep_has_work_stealing
 
 # -----------------------------------------------------------------------------
 # C++ type expression generators
@@ -173,7 +174,7 @@ def _gen_execute_pipeline(
 
   has_fused = (
     not instr.dedup_hash
-    and not instr.work_stealing
+    and not ep_has_work_stealing(instr)
     and not instr.block_group
     and len(instr.dest_specs) == 1
   )
@@ -243,7 +244,10 @@ def _gen_parallel_group(
   i2 = indent + "  "
 
   all_fused_eligible = all(
-    not op.dedup_hash and not op.work_stealing and not op.block_group and len(op.dest_specs) == 1
+    not op.dedup_hash
+    and not ep_has_work_stealing(op)
+    and not op.block_group
+    and len(op.dest_specs) == 1
     for op in exec_ops
   )
 
@@ -277,7 +281,7 @@ def _gen_parallel_group(
   single_dest_indices: list[int] = []
   multi_head_indices: list[int] = []
   for idx, op in enumerate(exec_ops):
-    if len(op.dest_specs) == 1 and not op.work_stealing and not op.dedup_hash:
+    if len(op.dest_specs) == 1 and not ep_has_work_stealing(op) and not op.dedup_hash:
       single_dest_indices.append(idx)
     else:
       multi_head_indices.append(idx)
@@ -406,7 +410,7 @@ def _gen_parallel_group(
     multi_in_dest: list[int] = []
     for rule_idx in rule_indices:
       op = exec_ops[rule_idx]
-      if len(op.dest_specs) == 1 and not op.work_stealing and not op.dedup_hash:
+      if len(op.dest_specs) == 1 and not ep_has_work_stealing(op) and not op.dedup_hash:
         single_in_dest.append(rule_idx)
       else:
         multi_in_dest.append(rule_idx)
diff --git a/src/srdatalog/ir/dialects/parallel/atomic_ws/__init__.py b/src/srdatalog/ir/dialects/parallel/atomic_ws/__init__.py
index 567373e..198acf4 100644
--- a/src/srdatalog/ir/dialects/parallel/atomic_ws/__init__.py
+++ b/src/srdatalog/ir/dialects/parallel/atomic_ws/__init__.py
@@ -17,9 +17,13 @@
 
 C4 scope (per spec §5, PR row C4): only the kernel-functor-level
 WS emit variants migrate to typed pragma + wrap op. The runner-side
-scaffolding (WCOJTask queue construction, steal-loop emission) stays
-on the legacy `ExecutePipeline.work_stealing: bool` path; A3 will
-drop the bool field and the C4 lowering will become the sole driver.
+scaffolding (WCOJTask queue construction, steal-loop emission) was
+historically gated by the deprecated `ExecutePipeline.work_stealing:
+bool` field; A3-2 retired that field, leaving the C4 lowering as
+the sole driver for the kernel-functor-level emit. Runner-side
+WS scaffolding integration is a separate follow-up that consumes
+the typed `WorkStealing` pragma (or its post-pass `WSScope` wrap
+op) directly.
 
 Ops currently registered: none — the only WS-specific op today is
 the MIR wrap op `mir.WSScope`, which lives next to `DedupGate` in
diff --git a/src/srdatalog/ir/dialects/parallel/atomic_ws/pragmas/work_stealing.py b/src/srdatalog/ir/dialects/parallel/atomic_ws/pragmas/work_stealing.py
index 12b57c0..266defa 100644
--- a/src/srdatalog/ir/dialects/parallel/atomic_ws/pragmas/work_stealing.py
+++ b/src/srdatalog/ir/dialects/parallel/atomic_ws/pragmas/work_stealing.py
@@ -1,13 +1,15 @@
 '''Pragma: `work_stealing` — atomic work-stealing kernel-functor emit.
 
-Trigger: `Rule(...).with_pragma(WorkStealing())` (preferred typed
-   form) or the legacy `with_plan(..., work_stealing=True)` keyword
-   (back-compat, slated for removal in A3).
+Trigger: `Rule(...).with_pragma(WorkStealing())` (the sole typed
+   form post-A3-2; the legacy `with_plan(..., work_stealing=True)`
+   keyword and the `mir.ExecutePipeline.work_stealing: bool` field
+   have been deleted).
 
 Materialization (this module): wrap each `mir.InsertInto` at the
    tail of an `ExecutePipeline.pipeline` with `mir.WSScope(inner=
    ...)` during `MirPragmaPass`, then strip the `WorkStealing`
-   instance from the EP's `pragmas` tuple.
+   instance from the EP's `pragmas` tuple. Unconditional after A3-2
+   — there is no dual-write bool-shadow to short-circuit on.
 
 Lowering (this module): a `@lowering(target=iir.cf, source=mir.
    WSScope)` rule (registered via the parent dialect's
@@ -23,18 +25,20 @@
    The lowering body delegates to the legacy `_lower_insert_into`
    with `ctx.ws_enabled` flipped True for the duration of the call,
    so the emission is literally the same code path the legacy
-   bool-field flow produces (byte-equivalence by construction).
+   bool-field flow produced (byte-equivalence by construction).
 
 C4 scope (per `docs/phase_c_pragma_materialization.md` §5, PR row
 C4 + §4.2 sub-dialect description): only the kernel-functor-level
 WS emit variants migrate. The runner-level WS scaffolding (WCOJTask
-queue, steal-loop emission, etc.) stays on the legacy
-`ExecutePipeline.work_stealing: bool` field path; A3 drops the bool
-and the wrap op becomes the sole driver.
+queue, steal-loop emission, etc.) was historically gated by the
+deleted `ExecutePipeline.work_stealing: bool` field — A3-2 retires
+that field; future runner-side scheduling integration is a
+separate work item that consumes the typed pragma directly.
 
 Spec: `docs/phase_c_pragma_materialization.md` §4.2, §5 (C4 PR
 row), §5.1 (per-PR acceptance gate); `docs/pragma_as_typed_object.
-md` §2 (Pragma base class), §3 (`@pragma_handler` decorator).
+md` §2 (Pragma base class), §3 (`@pragma_handler` decorator);
+`docs/phase_a3_remove_deprecated_bool_fields.md` §3.2 (A3-2 cut).
 '''
 
 from __future__ import annotations
@@ -99,7 +103,7 @@ def materialize_work_stealing(
   op: Any,  # ExecutePipeline at runtime — typed as Any to match the registry's
   # Callable[[Any, Pragma, PragmaCtx], Any] signature (see
   # core/pragma.py:PragmaRegistration). Body re-narrows via the typed
-  # `op.pragmas` / `op.work_stealing` accesses below.
+  # `op.pragmas` access below.
   pragma: Any,  # WorkStealing at runtime; same Any reason.
   ctx: Any,  # PragmaCtx, unused for this no-config pragma.
 ) -> Any:
@@ -108,32 +112,19 @@ def materialize_work_stealing(
 
   Returns a new `ExecutePipeline` with:
     - `pipeline` rewritten so every `InsertInto` is replaced by
-      `WSScope(inner=that_insert)` — **except** in the C4
-      dual-write transition state (see below),
+      `WSScope(inner=that_insert)`,
     - `pragmas` filtered to drop the consumed `WorkStealing`
       instance (per the `MirPragmaPass` post-flight invariant; see
       `docs/pragma_as_typed_object.md` §3).
 
-  C4 dual-write transition: the DSL `Rule.with_pragma(WorkStealing
-  ())` sets BOTH the typed pragma AND the legacy `work_stealing:
-  bool` field on the resulting `ExecutePipeline`, so the legacy
-  emitter (`compile_kernel_body` -> `_lower_insert_into`'s
-  `if ctx.ws_enabled:` branch, plus the runner-side WS scaffolding
-  in `codegen/cuda/batchfile.py` / `orchestrator.py`) keeps
-  producing byte-equivalent output without seeing the new `WSScope`
-  wrap op (which the monolith doesn't know how to traverse). When
-  `ep.work_stealing` is True we therefore SKIP the wrap step:
-  stripping the pragma is enough to satisfy `MirPragmaPass`'s
-  post-flight invariant, and the bool field continues to drive
-  emission via the legacy path.
-
-  When `ep.work_stealing` is False (the post-A3 pure-typed state,
-  or test fixtures that exercise only the typed surface), the
-  handler produces the wrap op and the registered
-  `@lowering(target=iir.cf, source=WSScope)` rule lowers it. The
-  resulting IIR is byte-equivalent to the WS branch of
-  `_lower_insert_into` by construction (the lowering delegates to
-  that helper with `ctx.ws_enabled` flipped True for the call).
+  Unconditional after A3-2: the legacy `ExecutePipeline.
+  work_stealing: bool` field has been retired (per
+  `docs/phase_a3_remove_deprecated_bool_fields.md` §3.2), so the
+  pragma is the sole emission driver. The registered `@lowering(
+  target=iir.cf, source=WSScope)` rule delegates back to
+  `_lower_insert_into` with `ctx.ws_enabled` flipped True for the
+  call, so the resulting IIR is byte-equivalent to the WS branch
+  of the legacy lowering by construction.
 
   Idempotency: the EP carries at most one `WorkStealing` (the DSL
   constructs at most one per `with_pragma(WorkStealing())` call;
@@ -147,10 +138,6 @@ def materialize_work_stealing(
   trailing `InsertInto` run is exactly the leaf position.
   '''
   new_pragmas = tuple(p for p in op.pragmas if not isinstance(p, WorkStealing))
-  # Dual-write transition: skip the wrap if the legacy bool is set
-  # (the legacy lowering will handle emission). See docstring above.
-  if op.work_stealing:
-    return dataclasses.replace(op, pragmas=new_pragmas)
   new_pipeline = _wrap_inserts_in_ws_scope(list(op.pipeline))
   return dataclasses.replace(op, pipeline=new_pipeline, pragmas=new_pragmas)
 
diff --git a/src/srdatalog/ir/dialects/relation/sorted_array/lowerings/__init__.py b/src/srdatalog/ir/dialects/relation/sorted_array/lowerings/__init__.py
index adaf1c6..a289d09 100644
--- a/src/srdatalog/ir/dialects/relation/sorted_array/lowerings/__init__.py
+++ b/src/srdatalog/ir/dialects/relation/sorted_array/lowerings/__init__.py
@@ -299,7 +299,24 @@ def lower_scan_pipeline(
   registered `@lowering`) because the wrap op only carries the
   root; the trailing `rest = ops[1:]` lives on the enclosing
   pipeline and is naturally accessible at this dispatch site.
+
+  A3-2: trailing `WSScope(InsertInto)` wrap ops (inserted by
+  `materialize_work_stealing`) are unwrapped in place + the
+  enclosing call is wrapped in a `ctx.ws_enabled=True` save/restore.
+  Mirrors the legacy bool-driven path: the same `_lower_insert_into`
+  body fires with `ctx.ws_enabled` set, byte-equivalent by
+  construction.
   '''
+  has_ws_scope = any(isinstance(o, mir.WSScope) for o in ops)
+  if has_ws_scope:
+    unwrapped: list[mir.MirNode] = [o.inner if isinstance(o, mir.WSScope) else o for o in ops]
+    prev_ws = ctx.ws_enabled
+    try:
+      ctx.ws_enabled = True
+      return lower_scan_pipeline(unwrapped, ctx)
+    finally:
+      ctx.ws_enabled = prev_ws
+
   head = ops[0] if ops else None
   if isinstance(head, mir.BlockGroupRoot):
     inner = head.inner
@@ -2620,21 +2637,17 @@ def _lower_insert_into(node: mir.InsertInto, ctx: LoweringCtx) -> list[Op]:
   lowering out of this monolith — until then they are the
   byte-equivalence anchor for the dual-write transition.
 
-  DEAD CODE NOTE (C4): the `if ctx.ws_enabled:` branch below
-  (count-phase WS, search for `ws_enabled`) is superseded by the
+  Post-A3-2: the `if ctx.ws_enabled:` branch below (count-phase WS,
+  search for `ws_enabled`) is reached only via the
   `@lowering(target=iir.cf, source=mir.WSScope)` rule registered by
-  `srdatalog.ir.dialects.parallel.atomic_ws.pragmas.work_stealing`.
-  Same dual-write contract as C2: the new lowering delegates back
-  into this helper with `ctx.ws_enabled` forced True, so the branch
-  remains functionally LIVE during the C4 transition (the
-  `with_pragma(WorkStealing())` DSL dual-writes both the bool field
-  AND the typed pragma; see `pragmas/work_stealing.py:
-  materialize_work_stealing`). The branch will be removable once
-  Phase A3 drops the `work_stealing: bool` field on
-  `mir.ExecutePipeline`. The `ws_cartesian_valid_var` branch
-  further below is unrelated to the C4 migration — it's a separate
-  legacy plumbing path (batched-Cartesian WS materialize) not
-  driven by `ctx.ws_enabled` and not in C4 scope.
+  `srdatalog.ir.dialects.parallel.atomic_ws.pragmas.work_stealing`,
+  which delegates back into this helper with `ctx.ws_enabled`
+  forced True. The deprecated `ExecutePipeline.work_stealing: bool`
+  shadow field that previously dual-drove this branch has been
+  retired (per `docs/phase_a3_remove_deprecated_bool_fields.md`
+  §3.2). The `ws_cartesian_valid_var` branch further below is
+  unrelated — it's a separate legacy plumbing path (batched-
+  Cartesian WS materialize) not driven by `ctx.ws_enabled`.
 
   DEAD CODE NOTE (C6 - count): the `if ctx.is_counting:` branches
   in this function (and elsewhere in this monolith) are superseded
diff --git a/src/srdatalog/ir/hir/lower.py b/src/srdatalog/ir/hir/lower.py
index 4058a54..cb4e495 100644
--- a/src/srdatalog/ir/hir/lower.py
+++ b/src/srdatalog/ir/hir/lower.py
@@ -599,7 +599,6 @@ def wrap_in_execute_pipeline(
   clause_order: list[int],
   rule_name: str,
   use_fan_out: bool = False,
-  work_stealing: bool = False,
   block_group: bool = False,
   count: bool = False,
   dedup_hash: bool = False,
@@ -613,6 +612,9 @@ def wrap_in_execute_pipeline(
   through into `mir.ExecutePipeline.pragmas` for `MirPragmaPass` to
   consume. Default `()` preserves the legacy callers that only
   supply the named bool fields.
+
+  A3-2: the `work_stealing` kwarg / field was removed; the typed
+  `WorkStealing` pragma in `pragmas` is now the sole driver.
   '''
   sources: list[mir.ColumnSource | mir.Scan | mir.Negation | mir.Aggregate] = []
   dests: list[mir.InsertInto] = []
@@ -627,7 +629,6 @@ def wrap_in_execute_pipeline(
     rule_name=rule_name,
     clause_order=list(clause_order),
     use_fan_out=use_fan_out,
-    work_stealing=work_stealing,
     block_group=block_group,
     dedup_hash=dedup_hash,
     count=count,
@@ -734,7 +735,6 @@ def lower_hir_to_mir_steps(hir: HirProgram) -> list[tuple[mir.MirNode, bool]]:
                 variant.clause_order,
                 nvtx,
                 use_fan_out=variant.fanout,
-                work_stealing=variant.work_stealing,
                 block_group=variant.block_group,
                 count=variant.count,
                 dedup_hash=variant.dedup_hash,
@@ -749,7 +749,6 @@ def lower_hir_to_mir_steps(hir: HirProgram) -> list[tuple[mir.MirNode, bool]]:
               variant.clause_order,
               nvtx,
               use_fan_out=variant.fanout,
-              work_stealing=variant.work_stealing,
               block_group=variant.block_group,
               count=variant.count,
               dedup_hash=variant.dedup_hash,
@@ -855,7 +854,6 @@ def lower_hir_to_mir_steps(hir: HirProgram) -> list[tuple[mir.MirNode, bool]]:
                 variant.clause_order,
                 nvtx,
                 use_fan_out=variant.fanout,
-                work_stealing=variant.work_stealing,
                 block_group=variant.block_group,
                 count=variant.count,
                 dedup_hash=variant.dedup_hash,
@@ -870,7 +868,6 @@ def lower_hir_to_mir_steps(hir: HirProgram) -> list[tuple[mir.MirNode, bool]]:
               variant.clause_order,
               nvtx,
               use_fan_out=variant.fanout,
-              work_stealing=variant.work_stealing,
               block_group=variant.block_group,
               count=variant.count,
               dedup_hash=variant.dedup_hash,
diff --git a/src/srdatalog/ir/hir/plan.py b/src/srdatalog/ir/hir/plan.py
index c2df69d..af29380 100644
--- a/src/srdatalog/ir/hir/plan.py
+++ b/src/srdatalog/ir/hir/plan.py
@@ -466,14 +466,13 @@ def _plan_variant(v: HirRuleVariant) -> None:
     var_order = compute_var_order_from_clauses(rule, clause_order, analysis.join_vars, delta_idx=d)
 
   # Propagate pragma flags whenever a plan is attached — the pragma
-  # branch above is gated on `plan.var_order`, but pragmas like
-  # `work_stealing: true` frequently appear on plan entries that have
-  # no custom var_order (e.g. Polonius subset_trans). Using the flags
-  # regardless of var_order matches Nim's planJoins which copies the
-  # pragma set unconditionally.
+  # branch above is gated on `plan.var_order`, but pragmas (e.g. the
+  # typed `WorkStealing` instance) frequently appear on plan entries
+  # that have no custom var_order (e.g. Polonius subset_trans). Using
+  # the flags regardless of var_order matches Nim's planJoins which
+  # copies the pragma set unconditionally.
   if plan is not None:
     v.fanout = plan.fanout
-    v.work_stealing = plan.work_stealing
     v.block_group = plan.block_group
     v.dedup_hash = plan.dedup_hash
     v.balanced_root = list(plan.balanced_root)
diff --git a/src/srdatalog/ir/hir/types.py b/src/srdatalog/ir/hir/types.py
index bffbb19..fc41bdd 100644
--- a/src/srdatalog/ir/hir/types.py
+++ b/src/srdatalog/ir/hir/types.py
@@ -98,7 +98,6 @@ class HirRuleVariant:
   balanced_sources: list[str] = field(default_factory=list)
   # Codegen hints (pragma-driven)
   fanout: bool = False
-  work_stealing: bool = False
   block_group: bool = False
   dedup_hash: bool = False
   count: bool = False
diff --git a/src/srdatalog/ir/mir/passes.py b/src/srdatalog/ir/mir/passes.py
index d38f6ed..6c9fb12 100644
--- a/src/srdatalog/ir/mir/passes.py
+++ b/src/srdatalog/ir/mir/passes.py
@@ -385,8 +385,39 @@ def _ep_is_single_dest_concat_eligible(ep: mir.ExecutePipeline) -> bool:
   '''Same predicate `_gen_parallel_group` uses to filter `single_in_dest`:
   exactly one dest, no work-stealing, no dedup-hash. Mirrors the
   upstream Nim orchestrator decision so byte-equivalence holds.
+
+  A3-2: the `ep.work_stealing` bool is gone; work-stealing presence
+  is now signalled by the typed `WorkStealing` pragma. Detect by
+  checking for the `WSScope` wrap op in the EP's pipeline (the
+  post-MirPragmaPass form) or for an unconsumed `WorkStealing`
+  instance in `ep.pragmas` (the pre-pass form, when this predicate
+  fires from `apply_concurrent_write_marking` running before the
+  pragma pass).
+  '''
+  return len(ep.dest_specs) == 1 and not ep_has_work_stealing(ep) and not ep.dedup_hash
+
+
+def ep_has_work_stealing(ep: mir.ExecutePipeline) -> bool:
+  '''True iff `ep` carries `WorkStealing` either as a still-unconsumed
+  typed pragma or as a `WSScope` wrap already materialized into
+  `ep.pipeline`. Added by A3-2 to replace the deprecated
+  `ep.work_stealing: bool` read across orchestrator / complete_runner
+  / mir-pass call sites.
   '''
-  return len(ep.dest_specs) == 1 and not ep.work_stealing and not ep.dedup_hash
+  # Late import to avoid pulling the parallel sub-dialect during
+  # base MIR-pass import (`mir/passes.py` is imported early in the
+  # default pipeline; `parallel.atomic_ws` is a plugin).
+  from srdatalog.ir.dialects.parallel.atomic_ws.pragmas.work_stealing import (
+    WorkStealing,
+  )
+
+  for p in ep.pragmas:
+    if isinstance(p, WorkStealing):
+      return True
+  for child in ep.pipeline:
+    if isinstance(child, mir.WSScope):
+      return True
+  return False
 
 
 def _mark_concurrent_writes_in_parallel_group(
diff --git a/src/srdatalog/ir/mir/print.py b/src/srdatalog/ir/mir/print.py
index 4aa181d..ebab3a8 100644
--- a/src/srdatalog/ir/mir/print.py
+++ b/src/srdatalog/ir/mir/print.py
@@ -276,6 +276,16 @@ def print_mir_sexpr(node: m.MirNode, indent: int = 0) -> str:
       + ")"
     )
 
+  # A3-2: WSScope wraps an InsertInto post-MirPragmaPass. The
+  # legacy bool-driven path printed the bare InsertInto; preserve
+  # that golden surface by unwrapping the gate when printing (the
+  # WS-vs-non-WS distinction lives in the pragma topology, not in
+  # the printed s-expression). Symmetric with the dedup_hash /
+  # block_group / fanout wrap ops that A3-1 / A3-3 / etc. will
+  # need to handle the same way.
+  if isinstance(node, m.WSScope):
+    return print_mir_sexpr(node.inner, indent)
+
   # --- Fixpoint maintenance ---
 
   if isinstance(node, m.RebuildIndex):
diff --git a/src/srdatalog/ir/mir/types.py b/src/srdatalog/ir/mir/types.py
index 3981dac..b80ed16 100644
--- a/src/srdatalog/ir/mir/types.py
+++ b/src/srdatalog/ir/mir/types.py
@@ -13,10 +13,11 @@
 
 ExecutePipeline gains a `pragmas: tuple[tuple[str, Any], ...]` field
 (open key/value list — typed as tuple from the start since it's a
-new field). The existing named bool fields (`dedup_hash`,
-`work_stealing`, `block_group`, `use_fan_out`, `count`,
-`concurrent_write`) stay for back-compat through Phase A; Phase A3
-removes them and the Phase C `MirPragmaPass` consumes only `pragmas`.
+new field). The remaining named bool fields (`dedup_hash`,
+`block_group`, `use_fan_out`, `count`, `concurrent_write`) stay for
+back-compat through Phase A; Phase A3 removes them and the Phase C
+`MirPragmaPass` consumes only `pragmas`. (A3-2 retired
+`work_stealing` — `WorkStealing()` is the sole driver now.)
 
 Mapping to Nim (unchanged):
   moColumnSource         -> ColumnSource
@@ -274,10 +275,11 @@ class WSScope(Op):
   The wrap op carries the inner `InsertInto` so the lowering has
   everything it needs to emit the WS-count increment / pass-through
   the materialize-phase write. The runner-level WS scaffolding
-  (WCOJTask queue + steal loop) is out of C4 scope per
-  `docs/phase_c_pragma_materialization.md` §5 (PR row C4); this op
-  covers only the kernel-functor-level WS emit variant that the
-  legacy `ws_enabled` flag drives in `_lower_insert_into`.
+  (WCOJTask queue + steal loop) was historically gated by the
+  deprecated `ExecutePipeline.work_stealing: bool` field that A3-2
+  retired (per `docs/phase_a3_remove_deprecated_bool_fields.md`
+  §3.2); this op covers only the kernel-functor-level WS emit
+  variant.
 
   Per spec §4.2, this op lives in the new `parallel.atomic_ws`
   sub-dialect's lowering surface; the type itself lives here in
@@ -508,7 +510,6 @@ class ExecutePipeline(Op):
   clause_order: list[int] = field(default_factory=list)
   pragmas: tuple[tuple[str, Any], ...] = ()  # Phase A: open key/value list
   use_fan_out: bool = False  # back-compat: removed in A3
-  work_stealing: bool = False
   block_group: bool = False
   dedup_hash: bool = False
   tiled_cartesian: bool = False  # C5 dual-write field; removed in A3
diff --git a/tests/test_codegen_batchfile.py b/tests/test_codegen_batchfile.py
index e476d3b..45e5c56 100644
--- a/tests/test_codegen_batchfile.py
+++ b/tests/test_codegen_batchfile.py
@@ -232,9 +232,15 @@ def test_generate_runner_balanced_scan_emits_histogram_kernel():
 def test_generate_runner_work_stealing_marker():
   import dataclasses
 
+  from srdatalog.ir.dialects.parallel.atomic_ws.pragmas.work_stealing import (
+    WorkStealing,
+  )
+
   ep = _andersen_base_pipeline()
-  # MIR is frozen post-Phase-A; replace rather than mutate.
-  ep = dataclasses.replace(ep, work_stealing=True)
+  # A3-2: WS presence is detected by `ep_has_work_stealing`, which
+  # sees either the typed `WorkStealing` pragma (pre-pass) or a
+  # `WSScope` wrap (post-pass). Construct the pre-pass form here.
+  ep = dataclasses.replace(ep, pragmas=(WorkStealing(),))
   full, _ = generate_runner(ep, "Andersen")
   assert "TODO: Implement work-stealing logic" in full
 
diff --git a/tests/test_cuda_context.py b/tests/test_cuda_context.py
index 5e0b056..8be01c7 100644
--- a/tests/test_cuda_context.py
+++ b/tests/test_cuda_context.py
@@ -296,7 +296,9 @@ def test_runner_gen_state_defaults():
   assert r.db_type_name == ""
   assert r.dest_arities == []
   assert r.is_balanced is False
-  assert r.is_work_stealing is False
+  # `r.is_work_stealing` removed by A3-2 alongside
+  # `ExecutePipeline.work_stealing`.
+  assert r.is_block_group is False
 
 
 if __name__ == "__main__":
diff --git a/tests/test_discipline_obsolete_code_ratchet.py b/tests/test_discipline_obsolete_code_ratchet.py
index bd26fef..f556a0c 100644
--- a/tests/test_discipline_obsolete_code_ratchet.py
+++ b/tests/test_discipline_obsolete_code_ratchet.py
@@ -55,13 +55,15 @@
 # new marker; CI requires you to bump the cap explicitly + mention it
 # in the PR description.
 #
-# Snapshot taken post-C6 + B-Filter/ConstantBind/Scan/InsertInto merged:
-#   sorted_array/lowerings/__init__.py    8
+# Snapshot taken post-C6 + B-Filter/ConstantBind/Scan/InsertInto merged,
+# decremented by A3-2 (one C4 marker retired alongside the
+# `ExecutePipeline.work_stealing` field deletion):
+#   sorted_array/lowerings/__init__.py    7
 #   core/passes.py (one informational use) 0  (excluded — framework infra)
 #                                        ----
-#                                  total  8
+#                                  total  7
 
-_MAX_DEAD_CODE_MARKERS = 8
+_MAX_DEAD_CODE_MARKERS = 7
 
 
 def _count_dead_code_markers() -> int:
diff --git a/tests/test_hir_user_plan.py b/tests/test_hir_user_plan.py
index 2ec212b..b15932c 100644
--- a/tests/test_hir_user_plan.py
+++ b/tests/test_hir_user_plan.py
@@ -106,7 +106,9 @@ def test_plan_pragma_flags_propagate_to_variant():
   v = hir.strata[0].base_variants[0]
   assert v.block_group is True
   assert v.fanout is True
-  assert v.work_stealing is False  # default
+  # `v.work_stealing` removed by A3-2 — `WorkStealing()` pragma is
+  # the sole driver now; this variant has no WS pragma attached.
+  assert not any(type(p).__name__ == 'WorkStealing' for p in v.pragmas)
   assert v.dedup_hash is False  # default
 
 
diff --git a/tests/test_integration_lsqb_q6_nosj.py b/tests/test_integration_lsqb_q6_nosj.py
index 03e2b4a..a247a7d 100644
--- a/tests/test_integration_lsqb_q6_nosj.py
+++ b/tests/test_integration_lsqb_q6_nosj.py
@@ -3,6 +3,9 @@
 from integration_helpers import diff_hir, diff_mir
 
 from srdatalog.dsl import Filter, Program, Relation, Var
+from srdatalog.ir.dialects.parallel.atomic_ws.pragmas.work_stealing import (
+  WorkStealing,
+)
 
 
 def build_lsqb_q6_nosj() -> Program:
@@ -31,8 +34,8 @@ def build_lsqb_q6_nosj() -> Program:
       .named("TwoHopPath")
       .with_plan(
         var_order=["p1", "p2", "p3", "t"],
-        work_stealing=True,
-      ),
+      )
+      .with_pragma(WorkStealing()),
     ],
   )
 
diff --git a/tests/test_parallel_atomic_ws_as_plugin.py b/tests/test_parallel_atomic_ws_as_plugin.py
index 130f381..daae26e 100644
--- a/tests/test_parallel_atomic_ws_as_plugin.py
+++ b/tests/test_parallel_atomic_ws_as_plugin.py
@@ -70,10 +70,9 @@ def _tc_program_with_ws() -> Program:
   '''Tiny TC-style program — same shape as
   `tests/test_parallel_data_as_plugin.py::_tc_program`, but with the
   recursive `TCRec` rule carrying `with_pragma(WorkStealing())` so
-  the resulting `ExecutePipeline` carries the WS dual-write signal
-  (typed `WorkStealing` instance on each plan's `pragmas` tuple AND
-  the legacy `PlanEntry.work_stealing=True` shadow field; see
-  `dsl._BUILTIN_BOOL_SHADOW_PRAGMAS`).
+  the resulting `ExecutePipeline` carries the WS typed pragma.
+  Post-A3-2 the legacy `PlanEntry.work_stealing` shadow bool is
+  gone — the typed pragma is the sole signal.
 
   Why the explicit `.with_plan(delta=0)`: the planner's variant-
   level propagation (`_plan_variant` in `ir/hir/plan.py`) keys on
@@ -128,16 +127,17 @@ def _find_execute_pipelines(node: object) -> list[mir.ExecutePipeline]:
 
 
 def _ws_ep(mir_prog: mir.Program) -> mir.ExecutePipeline:
-  '''Return the first `ExecutePipeline` whose `work_stealing` bool is
-  True (i.e. the WS-tagged rule). The DSL's `.with_pragma(
-  WorkStealing())` dual-writes this bool, so picking by it is robust
-  to MIR pass reorderings (e.g. if `MirPragmaPass` consumed the
-  typed pragma already, the bool is still the unambiguous marker
-  during the dual-write window).
+  '''Return the first `ExecutePipeline` carrying the WS signal — i.e.
+  either a still-unconsumed `WorkStealing` typed pragma in
+  `ep.pragmas`, or (post-MirPragmaPass) a `WSScope` wrap op in
+  `ep.pipeline`. Robust to MIR pass reorderings around the pragma
+  pass. Post-A3-2 the legacy `ep.work_stealing` bool is gone.
   '''
+  from srdatalog.ir.mir.passes import ep_has_work_stealing
+
   for step, _is_rec in mir_prog.steps:
     for ep in _find_execute_pipelines(step):
-      if ep.work_stealing:
+      if ep_has_work_stealing(ep):
         return ep
   raise AssertionError('no work-stealing ExecutePipeline in mir_prog')
 
@@ -254,14 +254,14 @@ def test_compile_kernel_body_byte_equivalent_across_plugin_paths_with_ws() -> No
   output.
 
   The fixture program `_tc_program_with_ws` tags `TCRec` with
-  `Rule.with_pragma(WorkStealing())`, so the dual-write contract
+  `Rule.with_pragma(WorkStealing())`, so the typed-pragma path
   (`docs/phase_c_pragma_materialization.md` §4.2) fires on each path:
 
-    - `MirPragmaPass` strips the typed pragma instance off the EP
-      (per `materialize_work_stealing`'s bool-shadow short-circuit)
-      but leaves the `pipeline` shape untouched; the legacy
-      `work_stealing=True` PlanEntry bool drives the WS-count emit
-      inside `_lower_insert_into`.
+    - `MirPragmaPass` consumes the typed pragma and replaces each
+      trailing `InsertInto` with a `WSScope` wrap; the registered
+      `@lowering(target=iir.cf, source=WSScope)` rule emits the
+      WS-count IIR shape (delegating back to `_lower_insert_into`
+      with `ctx.ws_enabled=True`).
     - The plugin-discovered Compiler and the explicit-register
       Compiler both see the same MIR + run the same kernel-pipeline,
       so the rendered CUDA must match byte-for-byte.
@@ -281,9 +281,13 @@ def test_compile_kernel_body_byte_equivalent_across_plugin_paths_with_ws() -> No
   compiler_explicit.register_plugin(register_atomic_ws)
   ep_explicit = _build_ws_ep_via(compiler_explicit)
 
-  # Sanity: both EPs are the WS-tagged one (the fixture's TCRec).
-  assert ep_via_plugin.work_stealing is True
-  assert ep_explicit.work_stealing is True
+  # Sanity: both EPs are the WS-tagged one (the fixture's TCRec) —
+  # post-A3-2 they carry a `WSScope` wrap op (the typed pragma was
+  # already consumed by `MirPragmaPass` during program-level run).
+  from srdatalog.ir.mir.passes import ep_has_work_stealing
+
+  assert ep_has_work_stealing(ep_via_plugin)
+  assert ep_has_work_stealing(ep_explicit)
 
   # Materialize phase.
   text_a = compile_kernel_body(ep_via_plugin, is_counting=False)
diff --git a/tests/test_pragma_work_stealing_end_to_end.py b/tests/test_pragma_work_stealing_end_to_end.py
index 8c7d1fe..225ae66 100644
--- a/tests/test_pragma_work_stealing_end_to_end.py
+++ b/tests/test_pragma_work_stealing_end_to_end.py
@@ -90,14 +90,14 @@ def _scan_insert_ep(
   *,
   arity: int = 2,
   rule_name: str = 'WSy',
-  work_stealing: bool = False,
   pragmas: tuple[Pragma, ...] = (),
 ) -> m.ExecutePipeline:
-  '''Build a minimal `Scan -> InsertInto` EP shape, optionally
-  carrying a typed `WorkStealing` pragma instead of (or in addition
-  to) the legacy bool field. Mirrors the C2 `_scan_insert_ep`
-  fixture in `tests/test_pragma_dedup_hash_end_to_end.py` so the two
-  per-pragma tests share their structural assumptions.
+  '''Build a minimal `Scan -> InsertInto` EP shape carrying a typed
+  `WorkStealing` pragma (post-A3-2: the legacy bool shadow field
+  was retired; this fixture exercises only the typed surface).
+  Mirrors the C2 `_scan_insert_ep` fixture in
+  `tests/test_pragma_dedup_hash_end_to_end.py` so the two per-pragma
+  tests share their structural assumptions.
   '''
   vars_ = [f'v{i}' for i in range(arity)]
   cols = list(range(arity))
@@ -119,7 +119,6 @@ def _scan_insert_ep(
     source_specs=[scan],
     dest_specs=[insert],
     rule_name=rule_name,
-    work_stealing=work_stealing,
     pragmas=pragmas,  # type: ignore[arg-type]
   )
 
@@ -141,21 +140,20 @@ def _empty_rule() -> Rule:
 
 def test_with_pragma_attaches_typed_pragma_to_default_plan():
   '''Calling `.with_pragma(WorkStealing())` on a rule with no plans
-  appends a default `PlanEntry(delta=-1)` carrying the pragma AND
-  the matching legacy bool field (the C4 dual-write contract).'''
+  appends a default `PlanEntry(delta=-1)` carrying the typed pragma.
+  Post-A3-2: the legacy `PlanEntry.work_stealing` shadow bool has
+  been retired, so the pragma is the sole signal.'''
   rule = _empty_rule().with_pragma(WorkStealing())
   assert len(rule.plans) == 1
   plan = rule.plans[0]
   assert plan.delta == -1
-  assert plan.work_stealing is True
   assert len(plan.pragmas) == 1
   assert isinstance(plan.pragmas[0], WorkStealing)
 
 
 def test_with_pragma_appends_to_existing_plans():
   '''When the rule already has plans, `.with_pragma(WorkStealing())`
-  appends the pragma to EVERY plan's `pragmas` tuple AND sets each
-  plan's `work_stealing=True` (per-variant dual-write). Matches the
+  appends the pragma to EVERY plan's `pragmas` tuple. Matches the
   per-rule semantic intent of `with_pragma`.'''
   rule = (
     _empty_rule()
@@ -165,14 +163,13 @@ def test_with_pragma_appends_to_existing_plans():
   )
   assert len(rule.plans) == 2
   for plan in rule.plans:
-    assert plan.work_stealing is True
     assert any(isinstance(p, WorkStealing) for p in plan.pragmas)
 
 
 def test_with_pragma_does_not_set_unrelated_bool_fields():
-  '''The C4 dual-write must touch ONLY the `work_stealing` bool, not
-  `dedup_hash` / `block_group` / `fanout`. Guards against typos in
-  `_BUILTIN_BOOL_SHADOW_PRAGMAS`.'''
+  '''Guards against typos in `_BUILTIN_BOOL_SHADOW_PRAGMAS`: setting
+  `WorkStealing` must not toggle the dual-write bools of OTHER
+  pragmas that still have a legacy shadow field.'''
   rule = _empty_rule().with_pragma(WorkStealing())
   plan = rule.plans[0]
   assert plan.dedup_hash is False
@@ -213,15 +210,13 @@ class _GhostWSPragma(Pragma):
 # -----------------------------------------------------------------------------
 
 
-def test_pragma_pass_inserts_ws_scope_when_bool_is_false(mir_compiler):
-  '''Pure typed path (bool=False, only pragma set): MirPragmaPass
-  wraps every `InsertInto` in `pipeline` with `WSScope`.
-
-  This is the post-A3 target state. Today only synthesized EPs
-  reach it — the DSL `.with_pragma(WorkStealing())` dual-writes the
-  bool field as well, hitting the dual-write skip branch below.
+def test_pragma_pass_inserts_ws_scope_when_pragma_set(mir_compiler):
+  '''Typed path: MirPragmaPass wraps every `InsertInto` in
+  `pipeline` with `WSScope` whenever the EP carries a `WorkStealing`
+  pragma. Post-A3-2 this is the ONLY path — the legacy bool shadow
+  field is gone.
   '''
-  ep = _scan_insert_ep(work_stealing=False, pragmas=(WorkStealing(),))
+  ep = _scan_insert_ep(pragmas=(WorkStealing(),))
   out = MirPragmaPass().apply(ep, mir_compiler)
   # Pragma instance consumed.
   assert out.pragmas == ()
@@ -230,31 +225,6 @@ def test_pragma_pass_inserts_ws_scope_when_bool_is_false(mir_compiler):
   assert isinstance(out.pipeline[0], m.Scan)
   assert isinstance(out.pipeline[1], m.WSScope)
   assert isinstance(out.pipeline[1].inner, m.InsertInto)
-  # Legacy bool was NOT toggled; the wrap op is the sole signal.
-  assert out.work_stealing is False
-
-
-def test_pragma_pass_skips_wrap_in_dual_write_mode(mir_compiler):
-  '''Dual-write transition (bool=True AND pragma set, the C4 DSL
-  shape): MirPragmaPass strips the pragma but leaves `pipeline`
-  untouched. The legacy `compile_kernel_body` -> `_lower_insert_into`
-  bool-driven path produces the WS-count emit when ctx.ws_enabled
-  is set; the wrap op never appears so the monolith doesn't need to
-  know about WSScope.
-
-  This is the load-bearing test for the dual-write contract: the
-  C4 PR must not break the byte-equivalence harness, and that
-  requires the bool-field path to remain the sole emission driver
-  until A3 drops the bool.
-  '''
-  ep = _scan_insert_ep(work_stealing=True, pragmas=(WorkStealing(),))
-  out = MirPragmaPass().apply(ep, mir_compiler)
-  # Pragma still consumed.
-  assert out.pragmas == ()
-  # No WSScope inserted.
-  assert all(not isinstance(child, m.WSScope) for child in out.pipeline)
-  # Bool preserved for the legacy emitter.
-  assert out.work_stealing is True
 
 
 def test_pragma_pass_via_apply_all_mir_passes_is_noop_for_empty_pragmas(
@@ -263,7 +233,7 @@ def test_pragma_pass_via_apply_all_mir_passes_is_noop_for_empty_pragmas(
   '''Sanity: the integration of `MirPragmaPass` into
   `apply_all_mir_passes` does not change MIR shape for EPs that
   carry no typed pragmas — the dominant case today.'''
-  ep = _scan_insert_ep(work_stealing=False, pragmas=())
+  ep = _scan_insert_ep(pragmas=())
   steps_in = [(ep, False)]
   steps_out = apply_all_mir_passes(steps_in)
   # Same number of steps, same EP identity-or-equal at the top.
@@ -300,7 +270,6 @@ def test_pragma_pass_only_wraps_inserts_not_other_ops(mir_compiler):
     source_specs=[scan],
     dest_specs=[insert],
     rule_name='WSWithFilter',
-    work_stealing=False,
     pragmas=(WorkStealing(),),  # type: ignore[arg-type]
   )
   out = MirPragmaPass().apply(ep, mir_compiler)
@@ -331,7 +300,7 @@ def test_ws_scope_lowering_emits_count_increment(mir_compiler):
   from srdatalog.ir.codegen.cuda.emit import EmitCtx, emit
   from srdatalog.ir.dialects.relation.sorted_array.lowerings import LoweringCtx
 
-  typed_ep = _scan_insert_ep(arity=2, work_stealing=False, pragmas=(WorkStealing(),))
+  typed_ep = _scan_insert_ep(arity=2, pragmas=(WorkStealing(),))
   materialized = MirPragmaPass().apply(typed_ep, mir_compiler)
   scope = materialized.pipeline[1]
   assert isinstance(scope, m.WSScope)
@@ -458,12 +427,11 @@ def test_atomic_ws_dialect_name_pinned():
 # -----------------------------------------------------------------------------
 
 
-def test_plan_entry_work_stealing_default_false():
-  '''Pin the legacy `PlanEntry.work_stealing` default. Tests that
-  depend on the dual-write contract assume this default flips to
-  True only when `.with_pragma(WorkStealing())` is called.'''
+def test_plan_entry_pragmas_default_empty():
+  '''Pin the `PlanEntry.pragmas` default. Post-A3-2 the legacy
+  `PlanEntry.work_stealing` shadow bool is gone — the typed
+  `WorkStealing()` pragma in `pragmas` is the sole signal.'''
   pe = PlanEntry()
-  assert pe.work_stealing is False
   assert pe.pragmas == ()
 
 
@@ -472,10 +440,9 @@ def test_plan_entry_replace_with_work_stealing_pragma_round_trip():
   preserves the rest of the entry. Symmetric with how
   `Rule.with_pragma` builds new plans.'''
   pe = PlanEntry(delta=0, var_order=('a', 'b'))
-  pe2 = dataclasses.replace(pe, pragmas=(WorkStealing(),), work_stealing=True)
+  pe2 = dataclasses.replace(pe, pragmas=(WorkStealing(),))
   assert pe2.delta == 0
   assert pe2.var_order == ('a', 'b')
-  assert pe2.work_stealing is True
   assert isinstance(pe2.pragmas[0], WorkStealing)
 
 
@@ -492,7 +459,7 @@ def test_apply_mir_pragma_pass_handles_work_stealing(mir_compiler):
   path's bool-skip behavior is covered by
   `test_pragma_pass_skips_wrap_in_dual_write_mode`.
   '''
-  ep = _scan_insert_ep(work_stealing=False, pragmas=(WorkStealing(),))
+  ep = _scan_insert_ep(pragmas=(WorkStealing(),))
   steps = [(ep, False)]
   out_steps = apply_mir_pragma_pass(steps)
   ep_out = out_steps[0][0]
@@ -508,7 +475,7 @@ def test_apply_mir_pragma_pass_is_idempotent_for_empty_pragmas():
   (`test_pragmas_empty_after_materialization`) applied through the
   chain entry, pinned for the C4 pragma class.
   '''
-  ep = _scan_insert_ep(work_stealing=False, pragmas=())
+  ep = _scan_insert_ep(pragmas=())
   steps = [(ep, False)]
   once = apply_mir_pragma_pass(steps)
   twice = apply_mir_pragma_pass(once)
@@ -559,11 +526,11 @@ def test_work_stealing_coexists_with_dedup_hash_on_same_ep(mir_compiler):
   c.register_dialect(WS_DIALECT)
 
   ep = _scan_insert_ep(
-    work_stealing=False,
     pragmas=(WorkStealing(), DedupHash()),
   )
-  # Force the dedup_hash bool off too so both pragmas hit their
-  # wrap-emitting (non-dual-write) branch.
+  # Force the dedup_hash bool off so the DedupHash handler hits its
+  # wrap-emitting (non-dual-write) branch. (WorkStealing's
+  # corresponding bool shadow was removed in A3-2.)
   ep = dataclasses.replace(ep, dedup_hash=False)
   out = MirPragmaPass().apply(ep, c)
   # Both pragmas consumed.