diff --git a/AGENT_HANDOFF.md b/AGENT_HANDOFF.md
index bd8dd2992..c89696363 100644
--- a/AGENT_HANDOFF.md
+++ b/AGENT_HANDOFF.md
@@ -3,6 +3,101 @@
 > **Update this file with every PR and every push to main.**
 > Last updated: 2026-08-12
 
+## 2026-08-12 Localized torque static-gate closure (#4142)
+
+The final cumulative changed-source MyPy blockers are closed without runtime
+changes: `dataset_io.read_csv` explicitly types its NumPy input/success arrays,
+and the Rate pipeline removes a redundant `SwingSource` cast around the already
+typed source factory return. The source factory also relies on the validated
+`DoublePendulumRunConfig | None` narrowing instead of recasting its non-`None`
+branch. SPEC change-log rows 1.16.55 through 1.16.58 restore the monotonic audit
+trail without replacing any mission text. The exact pinned Python 3.12 / MyPy
+1.13 16-file delta command, 147 focused localized tests, Ruff, formatting, and
+diff checks pass. UI, Rust, protected-publication, and #4142 completion gates
+remain open.
+
+## 2026-08-12 Source execution/dataset discriminator hardening (#4142)
+
+The source factory no longer uses `run_config or default`. It validates the
+raw value as `None` or `DoublePendulumRunConfig`, then defaults only the exact
+`None` case. Manual and triple-pendulum sources reject prescribed mode/profile,
+joint locks, and localized offsets while preserving explicit or implicit
+default passive empty execution.
+
+The outer variation-dataset JSON schema version now requires a genuine
+non-Boolean integer before normalization. Boolean, float, and string lookalikes
+fail closed, consistent with the strict nested plan and sibling Morris reader
+contracts. Evidence is 34/34 focused and 1,483/1,483 broader shared-swing,
+variation, and Rate tests, with one expected missing-Rust-wheel skip. UI, Rust,
+protected-publication, and #4142 completion gates remain open.
+
+## 2026-08-12 Localized torque source/wire hardening (#4142)
+
+The Rate source factory now enforces the same double-pendulum-only capability
+already declared by `SimulationConfig`: manual and triple-pendulum source
+discriminators reject non-empty localized torque commands instead of silently
+discarding them. `DoublePendulumRunConfig` validates the raw command collection
+before tuple normalization, guaranteeing typed contract failures for `None`
+and other malformed collection domains.
+
+`VariationPlan.from_json_dict` no longer coerces its schema discriminator with
+`int(...)`; only a genuine non-Boolean integer may select supported v1/v2
+behavior. Regression evidence is 102/102 focused and 1,464/1,464 broader
+shared-swing, variation, and Rate tests, with one expected missing-Rust-wheel
+skip. The broader UI, Rust, protected-publication, and #4142 completion gates
+remain open.
+
+## 2026-08-12 Localized torque adversarial corrections (#4142)
+
+The affected localized variation and helper seams now fail closed on their raw
+numeric domains. `NoiseSpec`/`VariationPlan` fields reject Boolean, string, and
+nonfinite values without coercion while preserving ordinary JSON integer/float
+plans and v1 migration. Localized helper functions validate command
+collections, base torque pairs, sample times, and durations with typed contract
+errors.
+
+One canonical fixed-step helper computes the effective RK4 duration used by
+Rate request validation, `SimulationConfig`, source construction, and fallback
+trace-grid construction. A locus inside the requested duration but outside the
+rounded integration grid is therefore rejected during request construction,
+not during a trial. The current PyQt variable picker hides
+`localized_torque_only` entries until a locus editor exists; loading such a
+plan remains fail-closed and atomic with an explicit locus-editor message.
+
+Local evidence is 118/118 correction-focused tests and 1,455/1,455 broader
+shared-swing/variation and Rate tests, with one expected missing-Rust-wheel
+skip. PyQt/React locus authoring, Rust parity, complete raw persistence,
+protected publication, and #4142 completion remain open.
+
+## 2026-08-12 Localized double-pendulum torque execution (#4142)
+
+Local child `codex/4142-localized-double-torque-core` starts from exact commit
+`11a699155588d3d948990c5f08b72c5cc8d2c746`. It implements the first bounded
+localized-perturbation execution path without widening the UI surface.
+
+- Immutable `LocalizedTorqueOffset` commands use only the topological IDs
+  `joint.shoulder` and `joint.wrist`, a required finite half-open
+  `time_window_s = [start, end)`, and a finite additive torque in N.m. These IDs
+  remain deliberately distinct from spatial trace points such as
+  `swing.wrist`.
+- Passive and prescribed double-pendulum runs add every active command at each
+  Python RK4 stage. Recorded joint-torque samples use the same half-open rule;
+  exact shared boundaries cannot double-apply.
+- Rate variation requests map the two registered commanded-torque variables to
+  exact one-point loci and deterministic pre-sampled values. Missing, multiple,
+  mismatched, out-of-duration, base-only, wrong-source, and unsupported
+  localized specifications fail before trial execution.
+- Explicit Rust execution fails closed. `auto` selects the Python forced path
+  when localized commands are present. Valid misses remain typed
+  `evaluated_no_impact` results with closest-approach evidence.
+
+Exact local evidence is 99/99 focused tests and 1,413/1,413 broader shared-
+swing/variation and Rate tests (one expected Rust-wheel skip), plus Ruff, Ruff
+format, and changed-source MyPy. This is a narrow core seam: PyQt/React locus
+authoring and presentation, other source/locus kinds, Rust parity, complete
+state/event/torque persistence, protected CI/publication, and epic completion
+remain open.
+
 ## 2026-08-12 Bounded ensemble chunk lifecycle foundation (#4142 R11.5)
 
 Local child `codex/4142-ensemble-chunks` starts from published #4405 head
diff --git a/SPEC.md b/SPEC.md
index b3c318331..be72c8d98 100644
--- a/SPEC.md
+++ b/SPEC.md
@@ -26,12 +26,108 @@
 | **Owner**               | D-sorganization                            |
 | **Primary Language(s)** | Python 3.11+, Rust, JavaScript, TypeScript |
 | **License**             | MIT                                        |
-| **Current Version**     | 1.16.53                                    |
-| **Spec Version**        | 1.16.53                                    |
+| **Current Version**     | 1.16.58                                    |
+| **Spec Version**        | 1.16.58                                    |
 | **Last Spec Update**    | 2026-08-12                                 |
 
 ## 2. Purpose & Mission
 
+### 2026-08-12 Localized torque static-gate closure (#4142)
+
+Version 1.16.58 closes the final cumulative changed-source static-analysis
+findings without changing runtime behavior. The variation CSV reader explicitly
+types its input and success arrays as NumPy arrays, satisfying the repository's
+`follow-imports=skip` delta MyPy gate. The Rate simulation pipeline returns the
+already typed `SwingSource` from `make_source` directly instead of wrapping it
+in a redundant cast. The source factory also uses the type narrowing guaranteed
+by its run-config contract instead of recasting the validated non-`None` branch.
+Behavioral contracts and serialized data are unchanged.
+
+Evidence is the exact cumulative 16-file changed-source MyPy command, 147/147
+localized contract tests, and scoped Ruff/format/diff gates. UI locus authoring,
+Rust parity, protected publication, and epic completion remain open.
+
+### 2026-08-12 Source execution and dataset discriminator hardening (#4142)
+
+Version 1.16.57 removes a truthiness-based source-configuration fallback.
+`make_source` now requires `run_config` to be `None` or an actual
+`DoublePendulumRunConfig` before constructing a default, so falsey and truthy
+wrong-type objects cannot silently select passive execution or reach incidental
+attribute errors. Manual and triple-pendulum sources accept only the default
+passive, profile-free, lock-free, localized-offset-free execution declaration;
+all non-default double-pendulum semantics fail before source construction.
+
+The outer variation dataset JSON reader now applies the same genuine
+non-Boolean integer schema discriminator used by `VariationPlan`. `True`,
+`1.5`, and `"1"` cannot select dataset schema v1 through coercion. The sibling
+Morris observation reader already performs an exact integer type check. Local
+evidence is 34/34 focused and 1,483/1,483 broader shared-swing, variation, and
+Rate tests, with one expected missing-Rust-wheel skip. UI locus authoring, Rust
+parity, protected publication, and epic completion remain open.
+
+### 2026-08-12 Localized torque source and wire hardening (#4142)
+
+Version 1.16.56 makes the double-pendulum-only localized-torque capability
+fail closed at every nearby public boundary. The source factory rejects a
+non-empty `commanded_torque_offsets` collection for both manual and triple-
+pendulum discriminators instead of silently dropping the command. The run
+configuration validates the raw collection as a tuple or list before
+canonical tuple conversion, so `None` and other malformed domains raise
+`ContractViolationError` rather than incidental `TypeError`.
+
+`VariationPlan.from_json_dict` now requires `schema_version` to be a genuine
+non-Boolean integer before normalization. Boolean, float, and string lookalikes
+cannot select a wire schema through coercion; emitted v2 and supported integer
+v1 migration documents retain their existing behavior. Evidence is 102/102
+focused tests and 1,464/1,464 broader shared-swing, variation, and Rate tests,
+with one expected missing-Rust-wheel skip. UI locus authoring, Rust parity,
+protected publication, and epic completion remain open.
+
+### 2026-08-12 Localized torque adversarial corrections (#4142)
+
+Version 1.16.55 closes three fail-closed gaps in the initial localized torque
+core. `NoiseSpec` numeric scale, bounds, and time loci and `VariationPlan` base
+values reject Boolean, string, and nonfinite raw values; run count and seed
+require genuine non-Boolean integers. Normal JSON integer/float documents and
+v1 migration remain supported. Public localized helpers likewise validate base
+torques, command collections, sample times, and duration before use and report
+contract violations rather than coercion or incidental Python exceptions.
+
+A shared fixed-step grid function is now authoritative for effective RK4
+duration in request, configuration, source, and fallback trace-grid paths.
+Localized windows must fit that effective duration before sampling or trial
+execution. The existing PyQt variation picker excludes contextual localized
+torque entries because it has no locus editor; an imported localized plan fails
+atomically with an explicit unrepresentable/locus-editor explanation. This does
+not claim PyQt or React locus authoring. Evidence is 118 correction-focused and
+1,455 broader passing tests, with one expected missing-Rust-wheel skip.
+
+### 2026-08-12 Localized double-pendulum torque execution (#4142)
+
+Version 1.16.54 introduces the first dynamics-backed localized perturbation
+contract. A `LocalizedTorqueOffset` targets exactly one topological
+double-pendulum joint, `joint.shoulder` or `joint.wrist`, over a required finite
+half-open time window `[start_s, end_s)` wholly inside the run. The finite N.m
+value adds to passive or prescribed commanded torque at every Python RK4 stage.
+Topological torque IDs remain intentionally distinct from spatial output point
+IDs, including `swing.wrist`.
+
+Two registry variables map deterministic variation samples to those exact
+joint loci. Validation rejects unsupported variables, non-torque localized
+sources, absent/multiple/mismatched point IDs or windows, out-of-duration
+windows, base-only localized use, incompatible swing sources, and explicit Rust
+before simulation. Automatic backend selection uses Python whenever localized
+commands are present. Recorded torque history obeys the same half-open rule;
+chunk-size changes do not alter deterministic outcomes; and a physically valid
+miss remains typed no-impact data with closest-approach evidence.
+
+This version is a narrow core execution seam. PyQt and React authoring and
+presentation, other localized variables/source kinds, Rust parity, complete
+raw state/event/torque persistence, protected publication, and #4142 completion
+remain open. Evidence is 99 focused tests and 1,413 broader shared-swing,
+variation, and Rate tests with one expected missing-Rust-wheel skip, plus Ruff,
+format, and changed-source MyPy.
+
 ### 2026-08-12 Bounded ensemble chunk lifecycle foundation (#4142 R11.5)
 
 Version 1.16.53 adds a bounded in-process execution lifecycle for complete Rate
@@ -3309,6 +3405,11 @@ Active development with stable core, continuous tool expansion, and web API in p
 
 | Date | Version | Changes |
 | ---- | ------- | ------- |
+| 2026-08-12 | 1.16.58 | fix(rate-of-closure, #4142): close the cumulative 16-source static gate with explicit NumPy CSV array annotations and removal of redundant pipeline/source-config casts; restore the missing 1.16.55-1.16.58 append-only history while preserving runtime and wire behavior. |
+| 2026-08-12 | 1.16.57 | fix(rate-of-closure, #4142): validate source run configurations before exact-`None` fallback; reject prescribed mode/profile, locks, and localized offsets on manual/triple sources; and require a genuine non-Boolean integer outer variation-dataset schema discriminator. |
+| 2026-08-12 | 1.16.56 | fix(rate-of-closure, #4142): reject localized torque offsets on unsupported manual/triple sources; validate run-config offset collections before tuple conversion; and require genuine non-Boolean integer variation-plan schema versions. |
+| 2026-08-12 | 1.16.55 | fix(rate-of-closure, #4142): fail closed on malformed localized numeric/collection domains; make the fixed-step effective RK4 duration authoritative for windows; and hide or atomically reject localized PyQt factors until locus authoring exists. |
+| 2026-08-12 | 1.16.54 | feat(rate-of-closure, #4142): execute additive shoulder/wrist commanded-torque offsets over strict half-open one-point loci at every Python RK4 stage; bind deterministic variation samples to exact topological joint IDs; fail closed on unsupported source, locus, duration, and Rust contracts; preserve typed no-impact and distinct spatial provenance while keeping UI, persistence, protected release, and epic completion open. |
 | 2026-08-12 | 1.16.53 | feat(rate-of-closure, #4142 R11.5): add immutable resource-bounded ensemble stream headers/result chunks and an injected commit/abort sink lifecycle; project and release one chunk of complete runs at a time; retain the existing materialized API through a compatibility collector; and keep durable streaming/archive/memory claims explicitly open. |
 | 2026-08-12 | 1.16.52 | fix(rate-of-closure, #4142): satisfy the exact protected Python 3.12 / NumPy 2.3.5 / Mypy 1.13 typing boundary with explicit array annotations/casts and built-in-float `finfo` normalization; retain unchanged numerical and wire behavior. |
 | 2026-08-12 | 1.16.51 | fix(rate-of-closure, #4142 R11.4): require complete trial output scalars to be finite real non-booleans; normalize accepted NumPy real scalars to built-in floats; and prove typed-object writer/reader domain closure with five TDD cases and 39 focused persistence tests. |
diff --git a/docs/development/RATE_OF_CLOSURE_CAMPAIGN_HANDOFF.md b/docs/development/RATE_OF_CLOSURE_CAMPAIGN_HANDOFF.md
index f51a2cc22..775c8d3bf 100644
--- a/docs/development/RATE_OF_CLOSURE_CAMPAIGN_HANDOFF.md
+++ b/docs/development/RATE_OF_CLOSURE_CAMPAIGN_HANDOFF.md
@@ -1,5 +1,95 @@
 # Rate of Closure Campaign Handoff
 
+## 2026-08-12 #4142 localized torque static-gate closure
+
+- Explicit NumPy annotations on variation CSV input/success arrays close the
+  cumulative changed-source MyPy findings.
+- The Rate pipeline removes the redundant cast around `make_source`, whose
+  public return contract is already `SwingSource`; runtime behavior is unchanged.
+- `make_source` uses its validated `DoublePendulumRunConfig | None` narrowing
+  directly instead of recasting the non-`None` branch.
+- SPEC change-log rows 1.16.55 through 1.16.58 restore the complete monotonic
+  audit trail without superseding their existing mission sections.
+- Evidence: pinned Python 3.12 / MyPy 1.13 exact 16-file delta MyPy, 147/147
+  focused localized tests, and scoped Ruff/format/diff gates pass.
+- UI locus authoring, Rust parity, protected publication, and epic #4142 remain
+  open.
+
+## 2026-08-12 #4142 source execution/dataset discriminator hardening
+
+- `make_source` validates the raw run config before defaulting; only exact
+  `None` selects a new default passive declaration, while falsey and truthy
+  wrong-type values fail with `ContractViolationError`.
+- Manual and triple-pendulum sources reject prescribed mode/profile, joint
+  locks, and localized torque offsets. Explicit default passive empty configs
+  remain supported.
+- The outer variation-dataset JSON schema version requires a genuine
+  non-Boolean integer. `True`, `1.5`, and `"1"` regressions fail without
+  coercion; the sibling Morris observation reader was confirmed already strict.
+- Evidence: 34/34 focused and 1,483/1,483 broader shared-swing, variation, and
+  Rate tests; one missing-Rust-wheel skip remains expected.
+- UI locus authoring, Rust parity, protected publication, and epic #4142 remain
+  open.
+
+## 2026-08-12 #4142 localized torque source/wire hardening
+
+- Manual and triple-pendulum source discriminators now reject non-empty
+  localized torque commands, preserving the double-pendulum-only execution
+  capability instead of silently discarding unsupported input.
+- `DoublePendulumRunConfig` validates the raw offset collection before tuple
+  conversion, so malformed `None`/scalar/mapping/string domains fail with a
+  typed contract violation.
+- `VariationPlan` wire-schema selection requires a genuine non-Boolean integer;
+  `True`, `2.5`, and `"2"` regressions are rejected without coercion while
+  integer v1 migration and emitted v2 documents remain supported.
+- Evidence: 102/102 focused and 1,464/1,464 broader shared-swing, variation,
+  and Rate tests; one missing-Rust-wheel skip remains expected.
+- UI locus authoring, Rust parity, protected publication, and epic #4142 remain
+  open.
+
+## 2026-08-12 #4142 localized torque adversarial corrections
+
+- Strict shared numeric validators reject raw Boolean/string/nonfinite
+  variation values and non-integer run counts/seeds while retaining valid JSON
+  integer/float and v1 migration documents.
+- Public localized helpers validate base pairs, typed command collections,
+  sample times, and positive duration with contract errors.
+- One canonical effective-RK4-duration calculation is used by request,
+  `SimulationConfig`, source, and fallback trace-grid preflight. A command that
+  extends beyond the rounded grid now fails before sampling/trial execution.
+- The existing PyQt picker hides `localized_torque_only` factors until locus
+  controls exist. Imported contextual plans fail atomically with an explicit
+  locus-editor/unrepresentable message; no UI authoring completion is claimed.
+- Evidence: 118/118 focused tests and 1,455/1,455 broader shared-swing,
+  variation, and Rate tests; one missing-Rust-wheel skip remains expected.
+- React/PyQt locus authoring, Rust parity, complete raw persistence, protected
+  publication, and epic #4142 remain open.
+
+## 2026-08-12 #4142 localized double-pendulum torque core
+
+- Local child `codex/4142-localized-double-torque-core` starts from exact
+  commit `11a699155588d3d948990c5f08b72c5cc8d2c746` and remains local/unpublished.
+- The first executable localized factors are additive shoulder and wrist
+  commanded torques. Their exact topological point IDs are `joint.shoulder`
+  and `joint.wrist`; spatial output IDs remain unchanged and cannot substitute
+  for a torque target.
+- Every command requires a finite half-open `[start, end)` window within the
+  run. The Python forced integrator evaluates offsets at each RK4 stage and
+  adds them to either passive zero commands or prescribed profile commands.
+- Request construction binds deterministic samples to one exact target locus.
+  Unsupported variables, missing/multiple/mismatched points or windows,
+  base-only localized use, wrong swing sources, and explicit Rust fail before
+  execution. `auto` selects Python when offsets exist.
+- Deterministic replay is invariant to ensemble chunk size; torque histories
+  pin the half-open boundary; valid misses remain typed no-impact data; and
+  spatial point provenance remains separate.
+- Evidence: 99/99 focused tests; 1,413/1,413 broader shared-swing/variation and
+  Rate tests with one expected Rust-wheel skip; Ruff, format, and changed-source
+  MyPy.
+- This closes only the narrow core execution seam. PyQt/React locus controls,
+  additional perturbation types and sources, Rust parity, complete raw state/
+  event/torque persistence, protected CI/publication, and #4142 remain open.
+
 ## 2026-08-12 #4142 R11.5 bounded ensemble chunk lifecycle foundation
 
 - Local child `codex/4142-ensemble-chunks` starts from exact published #4405
diff --git a/src/rate_of_closure/AGENT_HANDOFF.md b/src/rate_of_closure/AGENT_HANDOFF.md
index a3ff414ef..ceb3b12e7 100644
--- a/src/rate_of_closure/AGENT_HANDOFF.md
+++ b/src/rate_of_closure/AGENT_HANDOFF.md
@@ -3,6 +3,86 @@
 > **Update this file with every PR and every push to main.**
 > Last updated: 2026-08-12
 
+## 2026-08-12 Localized torque static-gate closure (#4142)
+
+Static cleanup now allows the full cumulative localized slice to pass the
+repository's changed-source MyPy gate: shared dataset CSV arrays have explicit
+NumPy types and `_make_source` returns the source factory's declared
+`SwingSource` directly. `make_source` likewise uses the type narrowing already
+established by its run-config contract rather than a redundant config cast.
+SPEC rows 1.16.55 through 1.16.58 now preserve the complete append-only history.
+Runtime/source/wire behavior is unchanged. Evidence is the pinned Python 3.12 /
+MyPy 1.13 16-file delta command, 147 focused tests, and clean Ruff, formatting,
+and diff gates. Remaining UI, Rust, publication, and epic work stays open.
+
+## 2026-08-12 Source execution/dataset discriminator hardening (#4142)
+
+`make_source` validates `run_config` before fallback, accepting only `None` or
+the exact shared run-config type. Unsupported manual/triple source kinds reject
+every non-default double-pendulum execution semantic: prescribed mode/profile,
+locks, and localized torque offsets. Explicit default passive configuration and
+implicit `None` retain their existing behavior.
+
+The shared variation-dataset reader now rejects coercive outer schema versions
+just like the nested plan reader. Evidence is 34 focused and 1,483 broader
+passing shared-swing, variation, and Rate tests, with one expected Rust-wheel
+skip. UI locus authoring, Rust parity, protected publication, and the remainder
+of #4142 remain open.
+
+## 2026-08-12 Localized torque source/wire hardening (#4142)
+
+`make_source` now rejects localized commanded-torque offsets for manual and
+triple-pendulum source kinds, matching the existing `SimulationConfig` and
+capability-registry contract that only the double pendulum can execute them.
+The shared run config validates malformed raw offset collections before
+canonicalization, and the shared `VariationPlan` JSON reader accepts only a
+genuine non-Boolean integer schema discriminator rather than coercing Boolean,
+float, or string lookalikes.
+
+Local evidence is 102 focused and 1,464 broader passing shared-swing,
+variation, and Rate tests, with one expected Rust-wheel skip. PyQt/React locus
+authoring, Rust parity, protected publication, and the remainder of #4142 stay
+open.
+
+## 2026-08-12 Localized torque fail-closed correction (#4142)
+
+Rate request construction now validates localized windows against the same
+rounded effective RK4 duration used by `SimulationConfig`, the source, and the
+fallback trace grid. A window that fits only the user-requested duration can no
+longer survive request construction and fail inside a trial. Shared variation
+numeric inputs and public localized helpers reject coercive Boolean/string and
+nonfinite domains with contract errors.
+
+The present PyQt variation row intentionally filters contextual localized
+torque registry entries because it has no locus editor. Loading an existing
+localized plan is atomic and fails with an explicit locus-editor/unrepresentable
+message. No PyQt/React authoring completion is claimed. Local evidence is 118
+focused and 1,455 broader passing tests with one expected Rust-wheel skip;
+protected release and the broader #4142 work remain open.
+
+## 2026-08-12 Localized joint-torque execution seam (#4142)
+
+Local child `codex/4142-localized-double-torque-core`, based on exact commit
+`11a699155588d3d948990c5f08b72c5cc8d2c746`, connects version-2 variation loci
+to actual double-pendulum dynamics for two commanded-torque variables.
+
+The only accepted targets are topological `joint.shoulder` and `joint.wrist`.
+Each requires one matching point ID and a finite half-open time window wholly
+inside `swing_duration_s`; values are additive N.m offsets evaluated at every
+Python RK4 stage over passive or prescribed commands. Constructor, request,
+and source validation reject missing/mismatched loci, invalid windows,
+unsupported variables/source kinds, base-only use, and explicit Rust before
+execution. Deterministic replay is chunk-size independent, torque-history IDs
+remain distinct from spatial trace IDs, and physically valid no-impact trials
+retain typed closest-approach evidence.
+
+Local evidence is 99 focused passes and 1,413 broader shared-swing/variation
+plus Rate passes with one expected missing-Rust-wheel skip. Ruff, format, and
+changed-source MyPy are green. UI authoring/presentation, additional localized
+variables and source types, Rust support, complete state/event/torque archive
+authority, protected release, and #4142 completion remain explicit follow-up
+work.
+
 ## 2026-08-12 Bounded complete-ensemble chunk execution seam (#4142 R11.5)
 
 Local child `codex/4142-ensemble-chunks` is based on exact #4405 head
diff --git a/src/rate_of_closure/simulation/pipeline.py b/src/rate_of_closure/simulation/pipeline.py
index b128205bb..7428f88a3 100644
--- a/src/rate_of_closure/simulation/pipeline.py
+++ b/src/rate_of_closure/simulation/pipeline.py
@@ -33,9 +33,13 @@
     ImpactSolverAPI,
     PostImpactState,
 )
+from shared.python.swing_sim.integration_grid import (
+    DEFAULT_SWING_RK4_DT_S,
+    effective_rk4_duration,
+)
 from shared.python.swing_sim.swing_source import SwingSource
 
-SWING_SAMPLE_DT_S = 1e-3
+SWING_SAMPLE_DT_S = DEFAULT_SWING_RK4_DT_S
 
 
 def swing_sample_times(duration_s: float) -> np.ndarray:
@@ -54,9 +58,7 @@ def configured_swing_sample_times(config: SimulationConfig) -> np.ndarray:
     """Return the exact grid produced by a validated simulation config."""
     if config.source_kind == "manual":
         return swing_sample_times(MANUAL_SWING_DURATION_S)
-    source_duration_s = (
-        round(config.swing_duration_s / SWING_SAMPLE_DT_S) * SWING_SAMPLE_DT_S
-    )
+    source_duration_s = effective_rk4_duration(config.swing_duration_s)
     return swing_sample_times(source_duration_s)
 
 
diff --git a/src/rate_of_closure/simulation/records.py b/src/rate_of_closure/simulation/records.py
index 7ecef5981..be39d93f3 100644
--- a/src/rate_of_closure/simulation/records.py
+++ b/src/rate_of_closure/simulation/records.py
@@ -26,6 +26,8 @@
     DeliveryDerived,
     PostImpactState,
 )
+from shared.python.swing_sim.integration_grid import effective_rk4_duration
+from shared.python.swing_sim.localized_torque import require_offsets_within_duration
 from shared.python.swing_sim.run_config import (
     DOUBLE_PENDULUM_JOINT_IDS,
     DoublePendulumRunConfig,
@@ -137,6 +139,12 @@ def __post_init__(self) -> None:
                 "joint locks currently require the double-pendulum source",
                 self.source_kind,
             )
+        if self.swing_run_config.commanded_torque_offsets:
+            require(
+                self.source_kind == "double_pendulum",
+                "localized torque offsets require the double-pendulum source",
+                self.source_kind,
+            )
         FlightModelType(self.flight_model)
         _validate_optional_impact_time(self.impact_time_s)
         require(
@@ -149,6 +157,14 @@ def __post_init__(self) -> None:
             "swing_duration_s must be finite and > 0",
             self.swing_duration_s,
         )
+        offset_duration_s = (
+            effective_rk4_duration(self.swing_duration_s)
+            if self.swing_run_config.commanded_torque_offsets
+            else self.swing_duration_s
+        )
+        require_offsets_within_duration(
+            self.swing_run_config.commanded_torque_offsets, offset_duration_s
+        )
 
     @property
     def ball_position_m(self) -> np.ndarray:
diff --git a/src/rate_of_closure/simulation/sources.py b/src/rate_of_closure/simulation/sources.py
index 6400a8d2c..e406ab1c9 100644
--- a/src/rate_of_closure/simulation/sources.py
+++ b/src/rate_of_closure/simulation/sources.py
@@ -45,7 +45,7 @@
 from rate_of_closure.simulation.triple_pendulum import (
     triple_total_energy as triple_total_energy,
 )
-from shared.python.swing_sim.run_config import DoublePendulumRunConfig
+from shared.python.swing_sim.run_config import DoublePendulumRunConfig, SwingRunMode
 from shared.python.swing_sim.swing_source import DoublePendulumSwing, SwingSource
 from shared.python.swing_sim.torque_library import TorqueProfileLibrary
 from shared.python.swing_sim.types import (
@@ -254,13 +254,33 @@ def make_source(
     Returns:
         A source whose samples are in the app frame.
     """
+    require(
+        run_config is None or isinstance(run_config, DoublePendulumRunConfig),
+        "run_config must be a DoublePendulumRunConfig or None",
+        run_config,
+    )
     require(kind in SOURCE_KINDS, f"unknown swing source kind {kind!r}", kind)
-    execution = run_config or DoublePendulumRunConfig()
+    execution = DoublePendulumRunConfig() if run_config is None else run_config
+    uses_default_mode = (
+        execution.mode is SwingRunMode.PASSIVE
+        and execution.prescribed_profile_id is None
+    )
+    require(
+        kind == "double_pendulum" or uses_default_mode,
+        "non-default execution policy is unsupported outside the "
+        "double-pendulum source",
+        kind,
+    )
     require(
         kind == "double_pendulum" or not execution.joint_locks.has_locks,
         "joint locks are unsupported outside the double-pendulum source",
         kind,
     )
+    require(
+        kind == "double_pendulum" or not execution.commanded_torque_offsets,
+        "localized torque offsets are unsupported outside the double-pendulum source",
+        kind,
+    )
     if kind == "manual":
         return ManualSwingSource(scenario, delivery=manual_delivery)
     if kind == "double_pendulum":
diff --git a/src/rate_of_closure/ui/pyqt6/variation_rows.py b/src/rate_of_closure/ui/pyqt6/variation_rows.py
index c49e23784..b0141bb63 100644
--- a/src/rate_of_closure/ui/pyqt6/variation_rows.py
+++ b/src/rate_of_closure/ui/pyqt6/variation_rows.py
@@ -98,6 +98,8 @@ def set_mode(self, mode: str) -> None:
         self.variable.blockSignals(True)
         self.variable.clear()
         for key in keys_for_mode(mode):
+            if variable_registry()[key].applicability == "localized_torque_only":
+                continue
             self.variable.addItem(short_label(key), key)
         self.variable.blockSignals(False)
         index = self.variable.findData(current)
diff --git a/src/rate_of_closure/ui/pyqt6/variation_tab_io.py b/src/rate_of_closure/ui/pyqt6/variation_tab_io.py
index 4b51a9163..fbfa4526e 100644
--- a/src/rate_of_closure/ui/pyqt6/variation_tab_io.py
+++ b/src/rate_of_closure/ui/pyqt6/variation_tab_io.py
@@ -30,6 +30,7 @@
     VariationDataset,
     VariationPlan,
     keys_for_mode,
+    variable_registry,
 )
 from shared.python.swing_sim.variation.dataset_io import write_csv, write_json
 
@@ -229,6 +230,15 @@ def _validate_plan_for_editors(self, plan: VariationPlan) -> None:
         legal = set(keys_for_mode(plan.mode))
         numeric_editor = self._rows[0]
         for spec in plan.noise:
+            definition = variable_registry().get(spec.variable_key)
+            if (
+                definition is not None
+                and definition.applicability == "localized_torque_only"
+            ):
+                raise ValueError(
+                    "localized torque requires a locus editor and is not "
+                    f"representable by this UI: {spec.variable_key}"
+                )
             if spec.variable_key not in legal:
                 raise ValueError(
                     "noise variable is not representable in this mode: "
diff --git a/src/rate_of_closure/variation/request_builder.py b/src/rate_of_closure/variation/request_builder.py
index a1c7dce59..2af9e9f0a 100644
--- a/src/rate_of_closure/variation/request_builder.py
+++ b/src/rate_of_closure/variation/request_builder.py
@@ -13,6 +13,12 @@
 from rate_of_closure.simulation import BallSetup, BallSupportMode, SimulationConfig
 from rate_of_closure.variation.simulation_types import SimulationEnsembleRequest
 from shared.python.contracts import require
+from shared.python.swing_sim.integration_grid import effective_rk4_duration
+from shared.python.swing_sim.run_config import (
+    SHOULDER_JOINT_ID,
+    WRIST_JOINT_ID,
+    LocalizedTorqueOffset,
+)
 from shared.python.swing_sim.types import PlaneOrientation
 from shared.python.swing_sim.variation import (
     CATEGORY_BALL_SETUP,
@@ -34,13 +40,15 @@ def _key(category: str, name: str) -> str:
 _IMPACT_TIME_OFFSET = _key(CATEGORY_SWING, "impact_time_offset_s")
 _DAMPING_SHOULDER = _key(CATEGORY_SWING, "damping_shoulder")
 _DAMPING_WRIST = _key(CATEGORY_SWING, "damping_wrist")
+_SHOULDER_TORQUE_OFFSET = _key(CATEGORY_SWING, "shoulder_commanded_torque_offset_nm")
+_WRIST_TORQUE_OFFSET = _key(CATEGORY_SWING, "wrist_commanded_torque_offset_nm")
 _TOE_OFFSET = _key(CATEGORY_DELIVERY, "impact_offset_toe_mm")
 _HIGH_OFFSET = _key(CATEGORY_DELIVERY, "impact_offset_high_mm")
 _HEAD_MASS = _key(CATEGORY_CLUB, "head_mass_kg")
 _HEAD_MOI = _key(CATEGORY_CLUB, "head_moi_kg_m2")
 _TEE_HEIGHT = _key(CATEGORY_BALL_SETUP, "tee_height_m")
 
-TRACE_CAPABLE_VARIABLE_KEYS = frozenset(
+GLOBAL_TRACE_VARIABLE_KEYS = frozenset(
     {
         _YAW,
         _SIDE_TILT,
@@ -56,6 +64,17 @@ def _key(category: str, name: str) -> str:
     }
 )
 
+LOCALIZED_TORQUE_VARIABLE_JOINTS = MappingProxyType(
+    {
+        _SHOULDER_TORQUE_OFFSET: SHOULDER_JOINT_ID,
+        _WRIST_TORQUE_OFFSET: WRIST_JOINT_ID,
+    }
+)
+
+TRACE_CAPABLE_VARIABLE_KEYS = frozenset(
+    GLOBAL_TRACE_VARIABLE_KEYS | set(LOCALIZED_TORQUE_VARIABLE_JOINTS)
+)
+
 
 def _is_real_scalar(value: object) -> bool:
     return isinstance(value, Real) and not isinstance(value, (bool, np.bool_))
@@ -82,13 +101,10 @@ def build_simulation_ensemble_request(
         "trace ensembles currently require the double_pendulum source",
         base_config.source_kind,
     )
-    require(
-        all(spec.is_global for spec in plan.noise),
-        "trace ensembles currently support only global perturbations",
-    )
     requested = {spec.variable_key for spec in plan.noise} | set(plan.base_variables)
     unsupported = sorted(requested - TRACE_CAPABLE_VARIABLE_KEYS)
     require(not unsupported, "variables are not trace-capable", unsupported)
+    _validate_noise_loci(plan, base_config)
     samples = sample_inputs(plan)
     configs = tuple(
         _apply_row(base_config, plan, row) for row in np.asarray(samples, dtype=float)
@@ -102,9 +118,83 @@ def _apply_row(
     row: np.ndarray,
 ) -> SimulationConfig:
     """Apply one sampled row plus explicit plan bases to ``base``."""
-    values = dict(plan.base_variables)
-    values.update(zip((spec.variable_key for spec in plan.noise), row, strict=True))
-    return apply_global_simulation_values(base, values)
+    localized_keys = set(LOCALIZED_TORQUE_VARIABLE_JOINTS)
+    values = {
+        key: value
+        for key, value in plan.base_variables.items()
+        if key not in localized_keys
+    }
+    offsets: list[LocalizedTorqueOffset] = []
+    for spec, sampled_value in zip(plan.noise, row, strict=True):
+        if spec.variable_key not in localized_keys:
+            values[spec.variable_key] = float(sampled_value)
+            continue
+        offsets.append(
+            LocalizedTorqueOffset(
+                joint_id=LOCALIZED_TORQUE_VARIABLE_JOINTS[spec.variable_key],
+                time_window_s=spec.time_window_s or (0.0, 0.0),
+                torque_nm=float(sampled_value),
+            )
+        )
+    updated = apply_global_simulation_values(base, values)
+    run_config = updated.swing_run_config
+    return replace(
+        updated,
+        swing_run_config=replace(
+            run_config,
+            commanded_torque_offsets=run_config.commanded_torque_offsets
+            + tuple(offsets),
+        ),
+    )
+
+
+def _validate_noise_loci(plan: VariationPlan, base_config: SimulationConfig) -> None:
+    """Validate global variables and exact localized torque loci."""
+    localized_specs = {
+        spec.variable_key: spec
+        for spec in plan.noise
+        if spec.variable_key in LOCALIZED_TORQUE_VARIABLE_JOINTS
+    }
+    effective_duration_s = (
+        effective_rk4_duration(base_config.swing_duration_s)
+        if localized_specs
+        else base_config.swing_duration_s
+    )
+    base_only = (
+        set(plan.base_variables) & set(LOCALIZED_TORQUE_VARIABLE_JOINTS)
+    ) - set(localized_specs)
+    require(
+        not base_only,
+        "localized torque base variables require a matching noise specification",
+        sorted(base_only),
+    )
+    for spec in plan.noise:
+        expected_joint = LOCALIZED_TORQUE_VARIABLE_JOINTS.get(spec.variable_key)
+        if expected_joint is None:
+            require(
+                spec.is_global,
+                "localized perturbation is unsupported for this variable",
+                spec.spec_id,
+            )
+            continue
+        window = spec.time_window_s
+        require(
+            window is not None,
+            "localized torque perturbation requires time_window_s",
+            spec.spec_id,
+        )
+        require(
+            spec.point_ids == (expected_joint,),
+            "localized torque perturbation requires its exact topological joint point",
+            (spec.point_ids, expected_joint),
+        )
+        assert window is not None
+        start_s, end_s = window
+        require(
+            0.0 <= start_s < end_s <= effective_duration_s,
+            "localized torque time window must lie within the effective RK4 duration",
+            (window, effective_duration_s),
+        )
 
 
 def apply_global_simulation_values(
@@ -121,7 +211,7 @@ def apply_global_simulation_values(
         all(isinstance(key, str) for key in values),
         "global simulation value keys must be strings",
     )
-    unsupported = sorted(set(values) - TRACE_CAPABLE_VARIABLE_KEYS)
+    unsupported = sorted(set(values) - GLOBAL_TRACE_VARIABLE_KEYS)
     require(not unsupported, "variables are not trace-capable", unsupported)
     require(
         all(_is_real_scalar(value) for value in values.values()),
@@ -212,12 +302,17 @@ def _apply_tee(
 
 
 TRACE_CAPABILITIES = MappingProxyType(
-    {"mode": "swing", "source_kind": "double_pendulum"}
+    {
+        "mode": "swing",
+        "source_kind": "double_pendulum",
+        "localized_torque_offsets": tuple(LOCALIZED_TORQUE_VARIABLE_JOINTS.items()),
+    }
 )
 
 __all__ = [
     "TRACE_CAPABILITIES",
     "TRACE_CAPABLE_VARIABLE_KEYS",
+    "LOCALIZED_TORQUE_VARIABLE_JOINTS",
     "apply_global_simulation_values",
     "build_simulation_ensemble_request",
 ]
diff --git a/src/shared/python/swing_sim/__init__.py b/src/shared/python/swing_sim/__init__.py
index 73a6cea49..0b057f678 100644
--- a/src/shared/python/swing_sim/__init__.py
+++ b/src/shared/python/swing_sim/__init__.py
@@ -18,6 +18,7 @@
     WRIST_JOINT_ID,
     DoublePendulumRunConfig,
     JointLockConfig,
+    LocalizedTorqueOffset,
     SwingRunMode,
 )
 from .swing_source import DoublePendulumSwing, SwingSource
@@ -56,6 +57,7 @@
     "FitMetadata",
     "JointTorqueAssignment",
     "JointLockConfig",
+    "LocalizedTorqueOffset",
     "PendulumParameters",
     "PendulumState",
     "PlaneOrientation",
diff --git a/src/shared/python/swing_sim/_numeric_contracts.py b/src/shared/python/swing_sim/_numeric_contracts.py
new file mode 100644
index 000000000..3f76b9e45
--- /dev/null
+++ b/src/shared/python/swing_sim/_numeric_contracts.py
@@ -0,0 +1,36 @@
+"""Strict scalar-domain validators shared by swing simulation contracts."""
+
+from __future__ import annotations
+
+import math
+from numbers import Integral, Real
+from typing import cast
+
+from shared.python.contracts import require
+
+
+def finite_real(value: object, name: str) -> float:
+    """Return one finite real scalar without Boolean or string coercion."""
+    require(
+        isinstance(value, Real) and not isinstance(value, bool),
+        f"{name} must be a real non-boolean scalar",
+        value,
+    )
+    normalized = float(cast(Real, value))
+    require(math.isfinite(normalized), f"{name} must be finite", value)
+    return normalized
+
+
+def integer(value: object, name: str, *, minimum: int = 0) -> int:
+    """Return one integer scalar without Boolean, float, or string coercion."""
+    require(
+        isinstance(value, Integral) and not isinstance(value, bool),
+        f"{name} must be an integer",
+        value,
+    )
+    normalized = int(cast(Integral, value))
+    require(normalized >= minimum, f"{name} must be >= {minimum}", value)
+    return normalized
+
+
+__all__ = ["finite_real", "integer"]
diff --git a/src/shared/python/swing_sim/integration_grid.py b/src/shared/python/swing_sim/integration_grid.py
new file mode 100644
index 000000000..01a3e6763
--- /dev/null
+++ b/src/shared/python/swing_sim/integration_grid.py
@@ -0,0 +1,24 @@
+"""Canonical fixed-step integration-grid calculations."""
+
+from __future__ import annotations
+
+from shared.python.contracts import require
+
+from ._numeric_contracts import finite_real
+
+DEFAULT_SWING_RK4_DT_S = 1e-3
+
+
+def effective_rk4_duration(
+    duration_s: object, dt_s: object = DEFAULT_SWING_RK4_DT_S
+) -> float:
+    """Return the duration represented by the nearest fixed-step RK4 grid."""
+    duration = finite_real(duration_s, "duration_s")
+    dt = finite_real(dt_s, "dt_s")
+    require(duration > 0.0, "duration_s must be > 0", duration_s)
+    require(dt > 0.0, "dt_s must be > 0", dt_s)
+    require(dt <= duration, "dt_s must not exceed duration_s", (dt_s, duration_s))
+    return float(int(round(duration / dt)) * dt)
+
+
+__all__ = ["DEFAULT_SWING_RK4_DT_S", "effective_rk4_duration"]
diff --git a/src/shared/python/swing_sim/localized_torque.py b/src/shared/python/swing_sim/localized_torque.py
new file mode 100644
index 000000000..92333d03a
--- /dev/null
+++ b/src/shared/python/swing_sim/localized_torque.py
@@ -0,0 +1,133 @@
+"""Topological joint/time contracts for additive torque commands."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import cast
+
+from shared.python.contracts import require
+
+from ._numeric_contracts import finite_real
+
+SHOULDER_JOINT_ID = "joint.shoulder"
+WRIST_JOINT_ID = "joint.wrist"
+DOUBLE_PENDULUM_JOINT_IDS = (SHOULDER_JOINT_ID, WRIST_JOINT_ID)
+
+
+@dataclass(frozen=True)
+class LocalizedTorqueOffset:
+    """One additive commanded torque over a half-open joint/time locus.
+
+    Topological joint IDs are deliberately distinct from spatial trace IDs.
+    The command is active on ``[start_s, end_s)`` so adjacent windows cannot
+    double-apply at their shared boundary.
+    """
+
+    joint_id: str
+    time_window_s: tuple[float, float]
+    torque_nm: float
+
+    def __post_init__(self) -> None:
+        require(
+            self.joint_id in DOUBLE_PENDULUM_JOINT_IDS,
+            "localized torque joint_id must belong to the double-pendulum model",
+            self.joint_id,
+        )
+        raw_window = cast(object, self.time_window_s)
+        require(
+            isinstance(raw_window, (tuple, list)) and len(raw_window) == 2,
+            "localized torque time_window_s must contain two real values",
+            raw_window,
+        )
+        window = cast(tuple[object, object] | list[object], raw_window)
+        start_s = finite_real(window[0], "localized torque time_window_s start")
+        end_s = finite_real(window[1], "localized torque time_window_s end")
+        require(
+            0.0 <= start_s < end_s,
+            "localized torque time_window_s must satisfy 0 <= start < end",
+            raw_window,
+        )
+        torque_nm = finite_real(
+            cast(object, self.torque_nm),
+            "localized torque offset",
+        )
+        object.__setattr__(self, "time_window_s", (start_s, end_s))
+        object.__setattr__(self, "torque_nm", torque_nm)
+
+    def is_active(self, time_s: float) -> bool:
+        """Return whether ``time_s`` lies in the declared half-open window."""
+        sample_time_s = finite_real(
+            cast(object, time_s),
+            "localized torque sample time",
+        )
+        start_s, end_s = self.time_window_s
+        return bool(start_s <= sample_time_s < end_s)
+
+
+def add_localized_offsets(
+    base_torques_nm: tuple[float, float],
+    offsets: tuple[LocalizedTorqueOffset, ...],
+    time_s: float,
+) -> tuple[float, float]:
+    """Add all active offsets to one shoulder/wrist command pair."""
+    raw_base = cast(object, base_torques_nm)
+    require(
+        isinstance(raw_base, (tuple, list)) and len(raw_base) == 2,
+        "base_torques_nm must contain two finite real values",
+        raw_base,
+    )
+    base = cast(tuple[object, object] | list[object], raw_base)
+    torques = [
+        finite_real(base[0], "base_torques_nm shoulder value"),
+        finite_real(base[1], "base_torques_nm wrist value"),
+    ]
+    commands = _validated_offsets(offsets)
+    sample_time_s = finite_real(cast(object, time_s), "localized torque sample time")
+    for offset in commands:
+        if offset.is_active(sample_time_s):
+            torques[DOUBLE_PENDULUM_JOINT_IDS.index(offset.joint_id)] += (
+                offset.torque_nm
+            )
+    return torques[0], torques[1]
+
+
+def require_offsets_within_duration(
+    offsets: tuple[LocalizedTorqueOffset, ...], duration_s: float
+) -> None:
+    """Require every half-open command window to lie inside one run."""
+    commands = _validated_offsets(offsets)
+    duration = finite_real(cast(object, duration_s), "run duration")
+    require(duration > 0.0, "run duration must be > 0", duration_s)
+    for offset in commands:
+        require(
+            offset.time_window_s[1] <= duration,
+            "localized torque time window must lie within the run duration",
+            (offset.time_window_s, duration),
+        )
+
+
+def _validated_offsets(offsets: object) -> tuple[LocalizedTorqueOffset, ...]:
+    """Return one strictly typed command collection."""
+    require(
+        isinstance(offsets, (tuple, list)),
+        "offsets must be a tuple or list of LocalizedTorqueOffset values",
+        offsets,
+    )
+    raw_commands = cast(tuple[object, ...] | list[object], offsets)
+    commands = tuple(raw_commands)
+    require(
+        all(isinstance(offset, LocalizedTorqueOffset) for offset in commands),
+        "offsets must contain only LocalizedTorqueOffset values",
+        commands,
+    )
+    return cast(tuple[LocalizedTorqueOffset, ...], commands)
+
+
+__all__ = [
+    "DOUBLE_PENDULUM_JOINT_IDS",
+    "SHOULDER_JOINT_ID",
+    "WRIST_JOINT_ID",
+    "LocalizedTorqueOffset",
+    "add_localized_offsets",
+    "require_offsets_within_duration",
+]
diff --git a/src/shared/python/swing_sim/run_config.py b/src/shared/python/swing_sim/run_config.py
index c38a3c43d..dee086e29 100644
--- a/src/shared/python/swing_sim/run_config.py
+++ b/src/shared/python/swing_sim/run_config.py
@@ -4,15 +4,19 @@
 
 from dataclasses import dataclass, field
 from enum import Enum
+from typing import cast
 
 from shared.python.contracts import require
 
 from ._torque_profile_validation import stable_id
+from .localized_torque import (
+    DOUBLE_PENDULUM_JOINT_IDS,
+    SHOULDER_JOINT_ID,
+    WRIST_JOINT_ID,
+    LocalizedTorqueOffset,
+)
 
 DOUBLE_PENDULUM_MODEL_ID = "model.double_pendulum.v1"
-SHOULDER_JOINT_ID = "joint.shoulder"
-WRIST_JOINT_ID = "joint.wrist"
-DOUBLE_PENDULUM_JOINT_IDS = (SHOULDER_JOINT_ID, WRIST_JOINT_ID)
 
 
 class SwingRunMode(str, Enum):  # noqa: UP042 - Python 3.10 compatibility
@@ -76,6 +80,7 @@ class DoublePendulumRunConfig:
     mode: SwingRunMode = SwingRunMode.PASSIVE
     prescribed_profile_id: str | None = None
     joint_locks: JointLockConfig = field(default_factory=JointLockConfig)
+    commanded_torque_offsets: tuple[LocalizedTorqueOffset, ...] = ()
 
     def __post_init__(self) -> None:
         require(
@@ -86,6 +91,32 @@ def __post_init__(self) -> None:
             "joint_locks must be a JointLockConfig",
             self.joint_locks,
         )
+        raw_offsets = cast(object, self.commanded_torque_offsets)
+        require(
+            isinstance(raw_offsets, (tuple, list)),
+            "commanded_torque_offsets must be a tuple or list",
+            raw_offsets,
+        )
+        offsets = tuple(cast(tuple[object, ...] | list[object], raw_offsets))
+        require(
+            all(isinstance(offset, LocalizedTorqueOffset) for offset in offsets),
+            "commanded_torque_offsets must contain LocalizedTorqueOffset values",
+            offsets,
+        )
+        validated_offsets = cast(tuple[LocalizedTorqueOffset, ...], offsets)
+        joint_order = {
+            joint_id: index for index, joint_id in enumerate(DOUBLE_PENDULUM_JOINT_IDS)
+        }
+        canonical = tuple(
+            sorted(
+                validated_offsets,
+                key=lambda offset: (
+                    joint_order[offset.joint_id],
+                    offset.time_window_s,
+                ),
+            )
+        )
+        object.__setattr__(self, "commanded_torque_offsets", canonical)
         if self.mode is SwingRunMode.PASSIVE:
             require(
                 self.prescribed_profile_id is None,
@@ -104,12 +135,14 @@ def prescribed(
         profile_id: str,
         *,
         joint_locks: JointLockConfig | None = None,
+        commanded_torque_offsets: tuple[LocalizedTorqueOffset, ...] = (),
     ) -> DoublePendulumRunConfig:
         """Build a prescribed-mode configuration for a stable profile ID."""
         return cls(
             mode=SwingRunMode.PRESCRIBED,
             prescribed_profile_id=profile_id,
             joint_locks=joint_locks or JointLockConfig(),
+            commanded_torque_offsets=commanded_torque_offsets,
         )
 
 
@@ -120,5 +153,6 @@ def prescribed(
     "WRIST_JOINT_ID",
     "DoublePendulumRunConfig",
     "JointLockConfig",
+    "LocalizedTorqueOffset",
     "SwingRunMode",
 ]
diff --git a/src/shared/python/swing_sim/swing_source.py b/src/shared/python/swing_sim/swing_source.py
index faf5a54da..cd47a06cc 100644
--- a/src/shared/python/swing_sim/swing_source.py
+++ b/src/shared/python/swing_sim/swing_source.py
@@ -11,13 +11,15 @@
 from __future__ import annotations
 
 import math
-from typing import Literal, Protocol, runtime_checkable
+from typing import Literal, Protocol, cast, runtime_checkable
 
 import numpy as np
 
 from shared.python.contracts import require
 
 from . import _rust_facade, reference
+from .integration_grid import DEFAULT_SWING_RK4_DT_S, effective_rk4_duration
+from .localized_torque import add_localized_offsets, require_offsets_within_duration
 from .run_config import (
     DOUBLE_PENDULUM_JOINT_IDS,
     DOUBLE_PENDULUM_MODEL_ID,
@@ -99,7 +101,7 @@ def sample(self, t: float) -> SwingSample:
 def _local_axis_rotation(angle: float) -> np.ndarray:
     """Rotation about the plane-local normal axis (local y) by ``angle``."""
     c, s = math.cos(angle), math.sin(angle)
-    return np.array([[c, 0.0, s], [0.0, 1.0, 0.0], [-s, 0.0, c]])
+    return cast(np.ndarray, np.array([[c, 0.0, s], [0.0, 1.0, 0.0], [-s, 0.0, c]]))
 
 
 class DoublePendulumSwing:
@@ -108,12 +110,8 @@ class DoublePendulumSwing:
     The trajectory is integrated once at construction on a uniform grid of
     step ``dt`` and sampled by linear interpolation of the joint state.
 
-    Backend policy (strict-Rust posture for hot loops):
-    - ``"rust"``: require the ``swing_core`` wheel; raise ``ImportError``
-      when absent.
-    - ``"python"``: always use the pure-Python reference (parity oracle).
-    - ``"auto"`` (default): use Rust when available, else fall back — this
-      constructor is a one-shot call, so the explicit fallback is allowed.
+    ``rust`` is strict, ``python`` is the parity oracle, and ``auto`` uses the
+    wheel when available. Python is mandatory for localized torque commands.
     """
 
     def __init__(
@@ -122,19 +120,13 @@ def __init__(
         plane: PlaneOrientation | None = None,
         initial_state: PendulumState | None = None,
         duration: float = 1.5,
-        dt: float = 1e-3,
+        dt: float = DEFAULT_SWING_RK4_DT_S,
         gravity_m_s2: float = DEFAULT_GRAVITY_M_S2,
         backend: Backend = "auto",
         run_config: DoublePendulumRunConfig | None = None,
         torque_library: TorqueProfileLibrary | None = None,
     ) -> None:
-        require(
-            math.isfinite(duration) and duration > 0.0,
-            "duration must be finite and > 0",
-            duration,
-        )
-        require(math.isfinite(dt) and dt > 0.0, "dt must be finite and > 0", dt)
-        require(dt <= duration, "dt must not exceed duration", dt)
+        duration_on_grid = effective_rk4_duration(duration, dt)
         require(
             math.isfinite(gravity_m_s2) and gravity_m_s2 >= 0.0,
             "gravity_m_s2 must be finite and >= 0",
@@ -159,10 +151,12 @@ def __init__(
             theta1=math.pi / 2.0, theta2=0.0, omega1=0.0, omega2=0.0
         )
         self._dt = float(dt)
-        self._n_steps = int(round(duration / dt))
-        self._duration = self._n_steps * self._dt
+        self._n_steps = int(round(duration_on_grid / self._dt))
+        self._duration = duration_on_grid
         self._run_config = config
         self._validate_locked_initial_velocity(config.joint_locks)
+        self._torque_offsets = config.commanded_torque_offsets
+        require_offsets_within_duration(self._torque_offsets, self._duration)
         self._torque_profile = _resolve_prescribed_profile(
             config, torque_library, self._duration
         )
@@ -173,14 +167,21 @@ def __init__(
         self._g_inplane = reference.in_plane_gravity(self._plane_r, gravity_m_s2)
         self._backend: Backend
 
+        if self._torque_offsets:
+            require(
+                backend != "rust",
+                "localized commanded torque offsets are not supported by the "
+                "Rust backend",
+            )
+
         if config.joint_locks.has_locks:
             require(
                 backend != "rust",
                 "locked joint integration is not supported by the Rust backend",
             )
             torque_at = (
-                self._prescribed_torque_tuple
-                if self._torque_profile is not None
+                self._commanded_torque_tuple
+                if self._torque_profile is not None or self._torque_offsets
                 else _zero_joint_torques
             )
             self._states = reference.simulate_locked(
@@ -193,7 +194,7 @@ def __init__(
                 config.joint_locks.mask,
             )
             self._backend = "python"
-        elif self._torque_profile is not None:
+        elif self._torque_profile is not None or self._torque_offsets:
             require(
                 backend != "rust",
                 "prescribed torque integration is not supported by the Rust backend",
@@ -204,7 +205,7 @@ def __init__(
                 self._g_inplane,
                 self._dt,
                 self._n_steps,
-                self._prescribed_torque_tuple,
+                self._commanded_torque_tuple,
             )
             self._backend = "python"
         elif backend == "rust" or (backend == "auto" and _rust_facade.rust_available()):
@@ -276,6 +277,15 @@ def _prescribed_torque_tuple(self, time_s: float) -> tuple[float, float]:
         values = profile.evaluate(sample_time)
         return values[SHOULDER_JOINT_ID], values[WRIST_JOINT_ID]
 
+    def _commanded_torque_tuple(self, time_s: float) -> tuple[float, float]:
+        """Return profile commands plus every active localized offset."""
+        base_torques = (
+            self._prescribed_torque_tuple(time_s)
+            if self._torque_profile is not None
+            else (0.0, 0.0)
+        )
+        return add_localized_offsets(base_torques, self._torque_offsets, time_s)
+
     def _validate_locked_initial_velocity(self, locks: JointLockConfig) -> None:
         """Reject lock activation that would require an unmodelled impulse."""
         state = self._initial_state
@@ -299,9 +309,8 @@ def joint_torques_at(self, t: float) -> dict[str, float]:
             t,
         )
         sample_time = min(max(t, 0.0), self._duration)
-        if self._torque_profile is None:
-            return {SHOULDER_JOINT_ID: 0.0, WRIST_JOINT_ID: 0.0}
-        return self._torque_profile.evaluate(sample_time)
+        shoulder_nm, wrist_nm = self._commanded_torque_tuple(sample_time)
+        return {SHOULDER_JOINT_ID: shoulder_nm, WRIST_JOINT_ID: wrist_nm}
 
     def _state_at(self, t: float) -> tuple[float, float, float, float]:
         """Linearly interpolate the joint state at time ``t``."""
@@ -342,7 +351,7 @@ def joint_positions(self, t: float) -> np.ndarray:
             math.sin(state.theta1) * x_axis - math.cos(state.theta1) * up_axis
         )
         tip = wrist + p.l2 * (math.sin(t12) * x_axis - math.cos(t12) * up_axis)
-        return np.vstack([np.zeros(3), wrist, tip])
+        return cast(np.ndarray, np.vstack([np.zeros(3), wrist, tip]))
 
     def sample(self, t: float) -> SwingSample:
         """Return the clubhead :class:`SwingSample` at time ``t``.
diff --git a/src/shared/python/swing_sim/tests/test_contract_api.py b/src/shared/python/swing_sim/tests/test_contract_api.py
index 53aa188d0..4044746a0 100644
--- a/src/shared/python/swing_sim/tests/test_contract_api.py
+++ b/src/shared/python/swing_sim/tests/test_contract_api.py
@@ -21,6 +21,7 @@
     "FitMetadata",
     "JointTorqueAssignment",
     "JointLockConfig",
+    "LocalizedTorqueOffset",
     "PendulumParameters",
     "PendulumState",
     "PlaneOrientation",
diff --git a/src/shared/python/swing_sim/tests/test_localized_torque.py b/src/shared/python/swing_sim/tests/test_localized_torque.py
new file mode 100644
index 000000000..0c352307c
--- /dev/null
+++ b/src/shared/python/swing_sim/tests/test_localized_torque.py
@@ -0,0 +1,179 @@
+"""Localized additive commanded-torque contracts and integration tests."""
+
+from __future__ import annotations
+
+import pytest
+
+from shared.python.contracts import ContractViolationError
+from shared.python.swing_sim.localized_torque import (
+    add_localized_offsets,
+    require_offsets_within_duration,
+)
+from shared.python.swing_sim.run_config import (
+    SHOULDER_JOINT_ID,
+    WRIST_JOINT_ID,
+    DoublePendulumRunConfig,
+    LocalizedTorqueOffset,
+)
+from shared.python.swing_sim.swing_source import DoublePendulumSwing
+from shared.python.swing_sim.types import PendulumState
+
+from .test_prescribed_run import _prescribed_swing, _profile
+
+pytestmark = pytest.mark.unit
+
+
+def _localized_swing(
+    offsets: tuple[LocalizedTorqueOffset, ...], **kwargs: object
+) -> DoublePendulumSwing:
+    defaults: dict[str, object] = {
+        "duration": 0.1,
+        "dt": 0.001,
+        "backend": "python",
+        "gravity_m_s2": 0.0,
+        "initial_state": PendulumState(0.0, 0.0, 0.0, 0.0),
+        "run_config": DoublePendulumRunConfig(commanded_torque_offsets=offsets),
+    }
+    defaults.update(kwargs)
+    return DoublePendulumSwing(**defaults)  # type: ignore[arg-type]
+
+
+def test_locus_is_exactly_half_open_and_rejects_invalid_sample_times() -> None:
+    offset = LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.02, 0.04), 3.5)
+
+    assert offset.is_active(0.02)
+    assert offset.is_active(0.039999999)
+    assert not offset.is_active(0.04)
+    assert not offset.is_active(0.01)
+    for invalid in (True, "0.03", float("nan")):
+        with pytest.raises(ContractViolationError, match="sample time"):
+            offset.is_active(invalid)  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize(
+    "kwargs",
+    [
+        {"joint_id": "swing.wrist"},
+        {"joint_id": "joint.elbow"},
+        {"time_window_s": (0.04, 0.02)},
+        {"time_window_s": None},
+        {"time_window_s": (0.02,)},
+        {"time_window_s": (-0.01, 0.02)},
+        {"time_window_s": (0.02, float("inf"))},
+        {"torque_nm": float("nan")},
+        {"torque_nm": True},
+    ],
+)
+def test_locus_rejects_invalid_contract(kwargs: dict[str, object]) -> None:
+    values: dict[str, object] = {
+        "joint_id": SHOULDER_JOINT_ID,
+        "time_window_s": (0.02, 0.04),
+        "torque_nm": 3.5,
+    }
+    values.update(kwargs)
+    with pytest.raises(ContractViolationError):
+        LocalizedTorqueOffset(**values)  # type: ignore[arg-type]
+
+
+def test_offsets_add_to_passive_and_prescribed_commands() -> None:
+    offsets = (
+        LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.02, 0.04), 3.0),
+        LocalizedTorqueOffset(WRIST_JOINT_ID, (0.02, 0.04), -2.0),
+    )
+    passive = _localized_swing(offsets)
+    prescribed = _prescribed_swing(
+        run_config=DoublePendulumRunConfig.prescribed(
+            _profile().profile_id,
+            commanded_torque_offsets=offsets,
+        )
+    )
+
+    assert passive.joint_torques_at(0.019) == pytest.approx(
+        {SHOULDER_JOINT_ID: 0.0, WRIST_JOINT_ID: 0.0}
+    )
+    assert passive.joint_torques_at(0.02) == pytest.approx(
+        {SHOULDER_JOINT_ID: 3.0, WRIST_JOINT_ID: -2.0}
+    )
+    assert passive.joint_torques_at(0.04) == pytest.approx(
+        {SHOULDER_JOINT_ID: 0.0, WRIST_JOINT_ID: 0.0}
+    )
+    assert prescribed.joint_torques_at(0.03) == pytest.approx(
+        {SHOULDER_JOINT_ID: 23.0, WRIST_JOINT_ID: -7.0}
+    )
+
+
+@pytest.mark.parametrize(
+    "base",
+    [(True, 0.0), ("2.0", 0.0), (float("nan"), 0.0), (0.0,)],
+)
+def test_add_helper_rejects_coercive_nonfinite_or_malformed_base(
+    base: object,
+) -> None:
+    offset = LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.02, 0.04), 1.0)
+    with pytest.raises(ContractViolationError, match="base_torques_nm"):
+        add_localized_offsets(base, (offset,), 0.03)  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize("offsets", [["bad"], "bad", (None,)])
+def test_public_helpers_reject_malformed_offset_collections(offsets: object) -> None:
+    with pytest.raises(ContractViolationError, match="offsets"):
+        add_localized_offsets((0.0, 0.0), offsets, 0.03)  # type: ignore[arg-type]
+    with pytest.raises(ContractViolationError, match="offsets"):
+        require_offsets_within_duration(offsets, 0.1)  # type: ignore[arg-type]
+
+
+@pytest.mark.parametrize("offsets", [None, 7, "bad", {"offset": "bad"}])
+def test_run_config_rejects_malformed_offset_collections_with_contract_error(
+    offsets: object,
+) -> None:
+    with pytest.raises(ContractViolationError, match="commanded_torque_offsets"):
+        DoublePendulumRunConfig(  # type: ignore[arg-type]
+            commanded_torque_offsets=offsets
+        )
+
+
+@pytest.mark.parametrize("duration", [True, "0.1", float("nan"), 0.0])
+def test_duration_helper_rejects_invalid_raw_domain(duration: object) -> None:
+    offset = LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.02, 0.04), 1.0)
+    with pytest.raises(ContractViolationError, match="duration"):
+        require_offsets_within_duration((offset,), duration)  # type: ignore[arg-type]
+
+
+def test_command_is_evaluated_at_every_rk4_substep(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    observed: list[float] = []
+    original_is_active = LocalizedTorqueOffset.is_active
+
+    def capture_active(offset: LocalizedTorqueOffset, time_s: float) -> bool:
+        observed.append(time_s)
+        return original_is_active(offset, time_s)
+
+    monkeypatch.setattr(LocalizedTorqueOffset, "is_active", capture_active)
+    _localized_swing(
+        (LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.0, 0.001), 2.0),),
+        duration=0.001,
+        dt=0.001,
+    )
+
+    assert observed == [0.0, 0.0005, 0.0005, 0.001]
+
+
+def test_auto_uses_python_and_explicit_rust_fails(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from shared.python.swing_sim import _rust_facade
+
+    monkeypatch.setattr(_rust_facade, "rust_available", lambda: True)
+    monkeypatch.setattr(
+        _rust_facade,
+        "simulate_rust",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(
+            AssertionError("localized commands reached Rust")
+        ),
+    )
+    offsets = (LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.0, 0.01), 2.0),)
+
+    assert _localized_swing(offsets, backend="auto").backend == "python"
+    with pytest.raises(ContractViolationError, match="localized.*Rust"):
+        _localized_swing(offsets, backend="rust")
diff --git a/src/shared/python/swing_sim/variation/contextual_registry.py b/src/shared/python/swing_sim/variation/contextual_registry.py
new file mode 100644
index 000000000..b9ee6649d
--- /dev/null
+++ b/src/shared/python/swing_sim/variation/contextual_registry.py
@@ -0,0 +1,44 @@
+"""Built-in variables that require an execution-context adapter."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class ContextualVariable:
+    """Registry source data for a context-gated variable."""
+
+    name: str
+    label: str
+    unit: str
+    default: float
+    typical_scale: float
+    guidance: str
+    applicability: str
+
+
+LOCALIZED_TORQUE_VARIABLES = (
+    ContextualVariable(
+        "shoulder_commanded_torque_offset_nm",
+        "Shoulder Commanded Torque Offset",
+        "N·m",
+        0.0,
+        2.0,
+        "Additive double-pendulum command over a required half-open time "
+        "window at joint.shoulder.",
+        "localized_torque_only",
+    ),
+    ContextualVariable(
+        "wrist_commanded_torque_offset_nm",
+        "Wrist Commanded Torque Offset",
+        "N·m",
+        0.0,
+        1.0,
+        "Additive double-pendulum command over a required half-open time "
+        "window at joint.wrist.",
+        "localized_torque_only",
+    ),
+)
+
+__all__ = ["LOCALIZED_TORQUE_VARIABLES", "ContextualVariable"]
diff --git a/src/shared/python/swing_sim/variation/dataset_io.py b/src/shared/python/swing_sim/variation/dataset_io.py
index ab841ba3c..7a602812e 100644
--- a/src/shared/python/swing_sim/variation/dataset_io.py
+++ b/src/shared/python/swing_sim/variation/dataset_io.py
@@ -36,6 +36,7 @@
 import numpy as np
 
 from shared.python.contracts import require
+from shared.python.swing_sim._numeric_contracts import integer
 
 from .engine import VariationDataset
 from .spec import VariationPlan
@@ -63,7 +64,11 @@ def to_json_dict(dataset: VariationDataset) -> dict[str, Any]:
 
 def from_json_dict(data: dict[str, Any]) -> VariationDataset:
     """Inverse of :func:`to_json_dict` (DbC-validated on construction)."""
-    version = int(data.get("schema_version", _SCHEMA_VERSION))
+    version = integer(
+        data.get("schema_version", _SCHEMA_VERSION),
+        "schema_version",
+        minimum=1,
+    )
     require(version == _SCHEMA_VERSION, "unsupported schema_version", version)
     outputs = np.array(
         [[math.nan if v is None else float(v) for v in row] for row in data["outputs"]],
@@ -132,9 +137,9 @@ def read_csv(path: str | Path, plan: VariationPlan) -> VariationDataset:
     require(input_names == expected, "CSV input columns must match plan", input_names)
     body = rows[1:]
     require(len(body) == plan.n_runs, "CSV row count must match plan", len(body))
-    inputs = np.empty((plan.n_runs, n_inputs), dtype=float)
+    inputs: np.ndarray = np.empty((plan.n_runs, n_inputs), dtype=float)
     outputs = np.full((plan.n_runs, len(output_names)), np.nan)
-    success = np.zeros(plan.n_runs, dtype=bool)
+    success: np.ndarray = np.zeros(plan.n_runs, dtype=bool)
     for row in body:
         i = int(row[0])
         require(0 <= i < plan.n_runs, "run index out of range", i)
diff --git a/src/shared/python/swing_sim/variation/registry.py b/src/shared/python/swing_sim/variation/registry.py
index bfbd4fdec..83231b632 100644
--- a/src/shared/python/swing_sim/variation/registry.py
+++ b/src/shared/python/swing_sim/variation/registry.py
@@ -16,6 +16,7 @@
 from shared.python.contracts import require
 
 from ..ball_setup import DEFAULT_DRIVER_TEE_HEIGHT_M
+from .contextual_registry import LOCALIZED_TORQUE_VARIABLES
 
 MODES: tuple[str, ...] = ("delivery", "swing", "launch")
 """Pipeline slices a plan can exercise (see :class:`VariationPlan`)."""
@@ -26,23 +27,19 @@
 CATEGORY_LAUNCH = "swing_sim.flight.launch"
 CATEGORY_BALL_SETUP = "swing_sim.ball_setup"
 
-APPLICABILITIES: tuple[str, ...] = ("always", "tee_only")
+APPLICABILITIES: tuple[str, ...] = (
+    "always",
+    "tee_only",
+    "localized_torque_only",
+)
 
 
 @dataclass(frozen=True)
 class VariableDef:
     """One registry entry: a variable other packages can perturb.
 
-    Attributes:
-        key: Namespaced registry key (``<category>.<name>``).
-        label: Title Case display label for UIs.
-        unit: Display unit suffix (e.g. ``"deg"``, ``"mph"``).
-        default: Base value used when a plan does not override it.
-        typical_scale: A sensible noise scale in the variable's unit
-            (seed for UI defaults and tooltips).
-        guidance: Sourced hover guidance in the FIELD_GUIDANCE style.
-        applicability: Context gate for variables that only have physical
-            meaning under a particular model configuration.
+    The fields bind a stable key to presentation metadata, a default and
+    typical scale, guidance, and any execution-context applicability gate.
     """
 
     key: str
@@ -319,6 +316,18 @@ def _register_builtins() -> None:
                 guidance=guidance,
             )
         )
+    for definition in LOCALIZED_TORQUE_VARIABLES:
+        register_variable(
+            VariableDef(
+                key=f"{CATEGORY_SWING}.{definition.name}",
+                label=definition.label,
+                unit=definition.unit,
+                default=definition.default,
+                typical_scale=definition.typical_scale,
+                guidance=definition.guidance,
+                applicability=definition.applicability,
+            )
+        )
     register_variable(
         VariableDef(
             key=f"{CATEGORY_BALL_SETUP}.tee_height_m",
diff --git a/src/shared/python/swing_sim/variation/spec.py b/src/shared/python/swing_sim/variation/spec.py
index a2117b8f9..9ba37de1a 100644
--- a/src/shared/python/swing_sim/variation/spec.py
+++ b/src/shared/python/swing_sim/variation/spec.py
@@ -23,6 +23,7 @@
 import numpy as np
 
 from shared.python.contracts import require
+from shared.python.swing_sim._numeric_contracts import finite_real, integer
 
 from .group_spec import PerturbationGroup
 from .registry import (
@@ -53,16 +54,24 @@ def _normalize_locus(
     point_ids: tuple[str, ...],
 ) -> tuple[tuple[float, float] | None, tuple[str, ...]]:
     """Validate and normalize optional temporal/spatial locus metadata."""
-    window = None if time_window_s is None else tuple(time_window_s)
-    if window is not None:
+    raw_window = cast(object, time_window_s)
+    require(
+        raw_window is None
+        or (isinstance(raw_window, (tuple, list)) and len(raw_window) == 2),
+        "time_window_s must contain finite start < end",
+        raw_window,
+    )
+    normalized_window: tuple[float, float] | None = None
+    if raw_window is not None:
+        window = cast(tuple[object, object] | list[object], raw_window)
+        start = finite_real(window[0], "time_window_s start")
+        end = finite_real(window[1], "time_window_s end")
         require(
-            len(window) == 2
-            and all(math.isfinite(float(value)) for value in window)
-            and float(window[0]) < float(window[1]),
+            start < end,
             "time_window_s must contain finite start < end",
             window,
         )
-        window = (float(window[0]), float(window[1]))
+        normalized_window = (start, end)
     points = tuple(point_ids)
     valid = all(
         isinstance(point, str) and bool(point) and point == point.strip()
@@ -73,7 +82,7 @@ def _normalize_locus(
         "point_ids must be unique, non-empty stable IDs",
         points,
     )
-    return window, points
+    return normalized_window, points
 
 
 @dataclass(frozen=True)
@@ -113,23 +122,20 @@ def __post_init__(self) -> None:
             f"distribution must be one of {DISTRIBUTIONS}",
             self.distribution,
         )
-        require(
-            math.isfinite(self.scale) and self.scale > 0.0,
-            "scale must be finite and > 0",
-            self.scale,
-        )
+        scale = finite_real(cast(object, self.scale), "scale")
+        require(scale > 0.0, "scale must be finite and > 0", self.scale)
+        bounds: dict[str, float | None] = {}
         for name in ("lower", "upper"):
             value = getattr(self, name)
-            require(
-                value is None or math.isfinite(float(value)),
-                f"{name} must be finite when given",
-                value,
+            bounds[name] = (
+                None if value is None else finite_real(cast(object, value), name)
             )
-        if self.lower is not None and self.upper is not None:
+        lower, upper = bounds["lower"], bounds["upper"]
+        if lower is not None and upper is not None:
             require(
-                self.lower < self.upper,
+                lower < upper,
                 "truncation bounds must satisfy lower < upper",
-                (self.lower, self.upper),
+                (lower, upper),
             )
         stable_id = self.variable_key if self.spec_id is None else self.spec_id
         require(
@@ -140,6 +146,9 @@ def __post_init__(self) -> None:
             stable_id,
         )
         window, points = _normalize_locus(self.time_window_s, self.point_ids)
+        object.__setattr__(self, "scale", scale)
+        object.__setattr__(self, "lower", lower)
+        object.__setattr__(self, "upper", upper)
         object.__setattr__(self, "spec_id", stable_id)
         object.__setattr__(self, "time_window_s", window)
         object.__setattr__(self, "point_ids", points)
@@ -170,17 +179,13 @@ def from_json_dict(cls, data: Mapping[str, Any]) -> NoiseSpec:
         return cls(
             variable_key=str(data["variable_key"]),
             distribution=str(data.get("distribution", "normal")),
-            scale=float(data.get("scale", 1.0)),
-            lower=None if data.get("lower") is None else float(data["lower"]),
-            upper=None if data.get("upper") is None else float(data["upper"]),
+            scale=cast(float, data.get("scale", 1.0)),
+            lower=cast(float | None, data.get("lower")),
+            upper=cast(float | None, data.get("upper")),
             spec_id=None if data.get("spec_id") is None else str(data["spec_id"]),
             time_window_s=cast(
                 tuple[float, float] | None,
-                (
-                    None
-                    if data.get("time_window_s") is None
-                    else tuple(float(value) for value in data["time_window_s"])
-                ),
+                data.get("time_window_s"),
             ),
             point_ids=tuple(str(value) for value in data.get("point_ids", [])),
         )
@@ -258,11 +263,14 @@ class VariationPlan:
 
     def __post_init__(self) -> None:
         require(self.mode in MODES, f"mode must be one of {MODES}", self.mode)
-        require(self.n_runs >= 1, "n_runs must be >= 1", self.n_runs)
-        require(self.seed >= 0, "seed must be >= 0", self.seed)
+        n_runs = integer(cast(object, self.n_runs), "n_runs", minimum=1)
+        seed = integer(cast(object, self.seed), "seed", minimum=0)
         require(len(self.noise) > 0, "plan must vary at least one variable", None)
         legal = set(keys_for_mode(self.mode))
-        base = {str(k): float(v) for k, v in self.base_variables.items()}
+        base = {
+            str(k): finite_real(cast(object, v), "base value")
+            for k, v in self.base_variables.items()
+        }
         for key, value in base.items():
             require(key in legal, f"base variable not legal in {self.mode} mode", key)
             require(math.isfinite(value), "base value must be finite", (key, value))
@@ -296,6 +304,8 @@ def __post_init__(self) -> None:
         object.__setattr__(self, "base_variables", MappingProxyType(base))
         object.__setattr__(self, "noise", specs)
         object.__setattr__(self, "groups", groups)
+        object.__setattr__(self, "n_runs", n_runs)
+        object.__setattr__(self, "seed", seed)
 
     def resolved_base(self) -> dict[str, float]:
         """Full base mapping: registry defaults overlaid with overrides."""
@@ -320,7 +330,11 @@ def to_json_dict(self) -> dict[str, Any]:
     @classmethod
     def from_json_dict(cls, data: Mapping[str, Any]) -> VariationPlan:
         """Inverse of :meth:`to_json_dict` (DbC-validated)."""
-        version = int(data.get("schema_version", 1))
+        version = integer(
+            cast(object, data.get("schema_version", 1)),
+            "schema_version",
+            minimum=1,
+        )
         require(
             version in _SUPPORTED_SCHEMA_VERSIONS,
             "unsupported schema_version",
@@ -328,15 +342,12 @@ def from_json_dict(cls, data: Mapping[str, Any]) -> VariationPlan:
         )
         return cls(
             mode=str(data["mode"]),
-            base_variables={
-                str(k): float(v)
-                for k, v in dict(data.get("base_variables", {})).items()
-            },
+            base_variables=dict(data.get("base_variables", {})),
             noise=tuple(
                 NoiseSpec.from_json_dict(entry) for entry in data.get("noise", [])
             ),
-            n_runs=int(data.get("n_runs", 200)),
-            seed=int(data.get("seed", 0)),
+            n_runs=cast(int, data.get("n_runs", 200)),
+            seed=cast(int, data.get("seed", 0)),
             flight_model=str(data.get("flight_model", "waterloo_penner")),
             groups=(
                 ()
diff --git a/src/shared/python/swing_sim/variation/tests/test_dataset_io.py b/src/shared/python/swing_sim/variation/tests/test_dataset_io.py
index b1ba0fd09..b16c75d88 100644
--- a/src/shared/python/swing_sim/variation/tests/test_dataset_io.py
+++ b/src/shared/python/swing_sim/variation/tests/test_dataset_io.py
@@ -7,6 +7,7 @@
 import numpy as np
 import pytest
 
+from shared.python.contracts import ContractViolationError
 from shared.python.swing_sim.variation import (
     CATEGORY_LAUNCH,
     NoiseSpec,
@@ -14,8 +15,10 @@
     run_variation,
 )
 from shared.python.swing_sim.variation.dataset_io import (
+    from_json_dict,
     read_csv,
     read_json,
+    to_json_dict,
     write_csv,
     write_json,
 )
@@ -53,6 +56,16 @@ def test_lossless_including_plan_and_failures(  # type: ignore[no-untyped-def]
         np.testing.assert_array_equal(loaded.success, dataset.success)
         assert loaded.elapsed_s == dataset.elapsed_s
 
+    @pytest.mark.parametrize("schema_version", [True, 1.5, "1"])
+    def test_rejects_coercive_outer_schema_version(
+        self, dataset, schema_version: object
+    ) -> None:  # type: ignore[no-untyped-def]
+        document = to_json_dict(dataset)
+        document["schema_version"] = schema_version
+
+        with pytest.raises(ContractViolationError, match="schema_version"):
+            from_json_dict(document)
+
     def test_grouped_v2_plan_is_retained_in_dataset_json(self, tmp_path: Path) -> None:
         speed = f"{CATEGORY_LAUNCH}.ball_speed_mph"
         spin = f"{CATEGORY_LAUNCH}.spin_rpm"
diff --git a/src/shared/python/swing_sim/variation/tests/test_spec.py b/src/shared/python/swing_sim/variation/tests/test_spec.py
index 5ac354dde..f8384568c 100644
--- a/src/shared/python/swing_sim/variation/tests/test_spec.py
+++ b/src/shared/python/swing_sim/variation/tests/test_spec.py
@@ -4,6 +4,7 @@
 
 import json
 
+import numpy as np
 import pytest
 
 from shared.python.contracts import ContractViolationError
@@ -54,6 +55,8 @@ def test_swing_category_pins_the_pendulum_variable_set(self) -> None:
             "impact_time_offset_s",
             "damping_shoulder",
             "damping_wrist",
+            "shoulder_commanded_torque_offset_nm",
+            "wrist_commanded_torque_offset_nm",
         )
 
     def test_club_and_launch_categories_pin(self) -> None:
@@ -129,6 +132,36 @@ def test_rejects_inverted_truncation(self) -> None:
                 upper=-2.0,
             )
 
+    @pytest.mark.parametrize("value", [True, np.bool_(False), "1.0", float("inf")])
+    @pytest.mark.parametrize("field", ["scale", "lower", "upper"])
+    def test_rejects_coercive_or_nonfinite_numeric_fields(
+        self, field: str, value: object
+    ) -> None:
+        kwargs = {field: value}
+        with pytest.raises(ContractViolationError, match=field):
+            NoiseSpec(
+                f"{CATEGORY_DELIVERY}.face_angle_deg",
+                **kwargs,
+            )
+
+    @pytest.mark.parametrize(
+        "window",
+        [
+            (False, True),
+            (np.bool_(False), 0.1),
+            ("0.0", "0.1"),
+            (0.0, float("inf")),
+        ],
+    )
+    def test_rejects_coercive_or_nonfinite_time_locus(
+        self, window: tuple[object, object]
+    ) -> None:
+        with pytest.raises(ContractViolationError, match="time_window_s"):
+            NoiseSpec(
+                f"{CATEGORY_SWING}.yaw_deg",
+                time_window_s=window,
+            )
+
     def test_json_round_trip(self) -> None:
         spec = NoiseSpec(
             variable_key=f"{CATEGORY_SWING}.yaw_deg",
@@ -300,6 +333,50 @@ def test_rejects_duplicate_noise_and_empty_noise(self) -> None:
         with pytest.raises(ContractViolationError):
             VariationPlan(mode="delivery", noise=())
 
+    @pytest.mark.parametrize("value", [True, np.bool_(False), "48.0", float("inf")])
+    def test_rejects_coercive_or_nonfinite_base_values(self, value: object) -> None:
+        with pytest.raises(ContractViolationError, match="base value"):
+            VariationPlan(
+                mode="delivery",
+                base_variables={f"{CATEGORY_DELIVERY}.clubhead_speed_mps": value},
+                noise=(NoiseSpec(f"{CATEGORY_DELIVERY}.face_angle_deg"),),
+            )
+
+    @pytest.mark.parametrize("field", ["n_runs", "seed"])
+    @pytest.mark.parametrize("value", [True, np.bool_(False), "2", 2.0])
+    def test_rejects_non_integer_run_and_seed_domains(
+        self, field: str, value: object
+    ) -> None:
+        kwargs = {field: value}
+        with pytest.raises(ContractViolationError, match=field):
+            VariationPlan(
+                mode="delivery",
+                noise=(NoiseSpec(f"{CATEGORY_DELIVERY}.face_angle_deg"),),
+                **kwargs,
+            )
+
+    def test_json_integer_and_float_domains_remain_supported(self) -> None:
+        raw = self._plan().to_json_dict()
+        raw["base_variables"] = {f"{CATEGORY_DELIVERY}.clubhead_speed_mps": 48}
+        raw["noise"][0]["scale"] = 2
+
+        restored = VariationPlan.from_json_dict(raw)
+
+        assert restored.n_runs == 64
+        assert restored.seed == 7
+        assert (
+            restored.base_variables[f"{CATEGORY_DELIVERY}.clubhead_speed_mps"] == 48.0
+        )
+        assert restored.noise[0].scale == 2.0
+
+    @pytest.mark.parametrize("schema_version", [True, 2.5, "2"])
+    def test_json_rejects_coercive_schema_version(self, schema_version: object) -> None:
+        raw = self._plan().to_json_dict()
+        raw["schema_version"] = schema_version
+
+        with pytest.raises(ContractViolationError, match="schema_version"):
+            VariationPlan.from_json_dict(raw)
+
     def test_rejects_duplicate_spec_ids_and_duplicate_variable_keys(self) -> None:
         face = f"{CATEGORY_DELIVERY}.face_angle_deg"
         speed = f"{CATEGORY_DELIVERY}.clubhead_speed_mps"
diff --git a/tests/rate_of_closure/test_localized_torque_variation.py b/tests/rate_of_closure/test_localized_torque_variation.py
new file mode 100644
index 000000000..5d924bd03
--- /dev/null
+++ b/tests/rate_of_closure/test_localized_torque_variation.py
@@ -0,0 +1,244 @@
+"""Localized torque variation mapping and complete-ensemble behavior."""
+
+from __future__ import annotations
+
+import dataclasses
+
+import numpy as np
+import pytest
+
+from rate_of_closure.simulation import ContactMode, run_simulation
+from rate_of_closure.variation.ensemble_chunks import CollectingEnsembleSink
+from rate_of_closure.variation.request_builder import (
+    LOCALIZED_TORQUE_VARIABLE_JOINTS,
+    TRACE_CAPABILITIES,
+)
+from rate_of_closure.variation.simulation_adapter import (
+    build_simulation_ensemble_request,
+    run_simulation_ensemble,
+    run_simulation_ensemble_chunks,
+)
+from shared.python.contracts import ContractViolationError
+from shared.python.swing_sim.run_config import (
+    SHOULDER_JOINT_ID,
+    WRIST_JOINT_ID,
+    DoublePendulumRunConfig,
+    LocalizedTorqueOffset,
+)
+from shared.python.swing_sim.variation import (
+    NoiseSpec,
+    VariationPlan,
+    run_variation,
+    variable_registry,
+)
+
+from .test_variation_simulation_request import (
+    _SHOULDER_TORQUE_OFFSET,
+    _WRIST_TORQUE_OFFSET,
+    _base_config,
+    _localized_spec,
+    _spec,
+)
+
+pytestmark = [pytest.mark.unit, pytest.mark.headless_safe]
+
+
+def test_builder_maps_deterministic_localized_joint_torque_offsets() -> None:
+    plan = VariationPlan(
+        mode="swing",
+        base_variables={
+            _SHOULDER_TORQUE_OFFSET: 1.0,
+            _WRIST_TORQUE_OFFSET: -0.5,
+        },
+        noise=(
+            _localized_spec(_SHOULDER_TORQUE_OFFSET, SHOULDER_JOINT_ID),
+            _localized_spec(_WRIST_TORQUE_OFFSET, WRIST_JOINT_ID, scale=1.0),
+        ),
+        n_runs=3,
+        seed=73,
+    )
+
+    first = build_simulation_ensemble_request(plan, _base_config())
+    second = build_simulation_ensemble_request(plan, _base_config())
+
+    assert first.sampled_inputs == pytest.approx(second.sampled_inputs)
+    for row, config in zip(first.sampled_inputs, first.configs, strict=True):
+        assert config.swing_run_config.commanded_torque_offsets == (
+            LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.02, 0.04), row[0]),
+            LocalizedTorqueOffset(WRIST_JOINT_ID, (0.02, 0.04), row[1]),
+        )
+
+
+def test_locus_does_not_replace_spatial_trace_ids() -> None:
+    plan = VariationPlan(
+        mode="swing",
+        noise=(_localized_spec(_SHOULDER_TORQUE_OFFSET, SHOULDER_JOINT_ID),),
+        n_runs=2,
+        seed=11,
+    )
+
+    result = run_simulation_ensemble(
+        build_simulation_ensemble_request(plan, _base_config())
+    )
+
+    assert result.traces.point_ids == (
+        "swing.pivot",
+        "swing.wrist",
+        "swing.clubhead.reference",
+    )
+    assert result.traces.point_ids != (SHOULDER_JOINT_ID, WRIST_JOINT_ID)
+
+
+def test_torque_history_pins_half_open_boundary_and_stable_ids() -> None:
+    plan = VariationPlan(
+        mode="swing",
+        noise=(
+            _localized_spec(_SHOULDER_TORQUE_OFFSET, SHOULDER_JOINT_ID),
+            _localized_spec(_WRIST_TORQUE_OFFSET, WRIST_JOINT_ID),
+        ),
+        n_runs=2,
+        seed=31,
+    )
+    request = build_simulation_ensemble_request(plan, _base_config())
+    run = run_simulation(request.configs[0])
+    start_index = int(round(0.02 / 0.001))
+    end_index = int(round(0.04 / 0.001))
+
+    assert run.swing_joint_ids == (SHOULDER_JOINT_ID, WRIST_JOINT_ID)
+    np.testing.assert_allclose(run.swing_applied_torques_nm[:start_index], 0.0)
+    np.testing.assert_allclose(
+        run.swing_applied_torques_nm[start_index:end_index],
+        np.broadcast_to(request.sampled_inputs[0], (end_index - start_index, 2)),
+    )
+    np.testing.assert_allclose(run.swing_applied_torques_nm[end_index:], 0.0)
+
+
+def test_replay_is_chunk_size_independent_and_retains_typed_misses() -> None:
+    plan = VariationPlan(
+        mode="swing",
+        noise=(
+            _localized_spec(_SHOULDER_TORQUE_OFFSET, SHOULDER_JOINT_ID),
+            _localized_spec(_WRIST_TORQUE_OFFSET, WRIST_JOINT_ID),
+        ),
+        n_runs=3,
+        seed=41,
+    )
+    base = dataclasses.replace(
+        _base_config(), contact_mode=ContactMode.FIXED_BALL_CONTACT
+    )
+    request = build_simulation_ensemble_request(plan, base)
+    first = run_simulation_ensemble_chunks(
+        request, CollectingEnsembleSink(), chunk_size=1
+    )
+    second = run_simulation_ensemble_chunks(
+        request, CollectingEnsembleSink(), chunk_size=2
+    )
+
+    assert tuple(item.status for item in first.outcomes) == tuple(
+        item.status for item in second.outcomes
+    )
+    assert all(item.status.value == "evaluated_no_impact" for item in first.outcomes)
+    assert all(item.value("closest_approach_m") is not None for item in first.outcomes)
+    np.testing.assert_array_equal(first.variation.inputs, second.variation.inputs)
+    np.testing.assert_allclose(first.variation.outputs, second.variation.outputs)
+    np.testing.assert_array_equal(first.traces.sample_valid, second.traces.sample_valid)
+    np.testing.assert_allclose(first.traces.positions_m, second.traces.positions_m)
+
+
+def test_capability_registry_is_explicit_and_topological() -> None:
+    assert TRACE_CAPABILITIES["localized_torque_offsets"] == (
+        (_SHOULDER_TORQUE_OFFSET, SHOULDER_JOINT_ID),
+        (_WRIST_TORQUE_OFFSET, WRIST_JOINT_ID),
+    )
+    assert (
+        tuple(LOCALIZED_TORQUE_VARIABLE_JOINTS.items())
+        == TRACE_CAPABILITIES["localized_torque_offsets"]
+    )
+    for key in LOCALIZED_TORQUE_VARIABLE_JOINTS:
+        definition = variable_registry()[key]
+        assert definition.unit == "N·m"
+        assert definition.applicability == "localized_torque_only"
+
+
+@pytest.mark.parametrize(
+    "spec",
+    [
+        _spec(_SHOULDER_TORQUE_OFFSET, 1.0),
+        NoiseSpec(
+            _SHOULDER_TORQUE_OFFSET,
+            scale=1.0,
+            point_ids=(SHOULDER_JOINT_ID,),
+        ),
+        NoiseSpec(
+            _SHOULDER_TORQUE_OFFSET,
+            scale=1.0,
+            time_window_s=(0.02, 0.04),
+        ),
+        _localized_spec(_SHOULDER_TORQUE_OFFSET, WRIST_JOINT_ID),
+        NoiseSpec(
+            _SHOULDER_TORQUE_OFFSET,
+            scale=1.0,
+            time_window_s=(0.02, 0.04),
+            point_ids=(SHOULDER_JOINT_ID, WRIST_JOINT_ID),
+        ),
+        _localized_spec(
+            _SHOULDER_TORQUE_OFFSET,
+            SHOULDER_JOINT_ID,
+            window=(0.19, 0.21),
+        ),
+    ],
+)
+def test_builder_rejects_incomplete_or_incompatible_loci(spec: NoiseSpec) -> None:
+    plan = VariationPlan(mode="swing", noise=(spec,), n_runs=2)
+
+    with pytest.raises(ContractViolationError, match="torque|locus|window|point"):
+        build_simulation_ensemble_request(plan, _base_config())
+
+
+def test_scalar_executor_rejects_commanded_torque_instead_of_ignoring_it() -> None:
+    plan = VariationPlan(
+        mode="swing",
+        noise=(_spec(_SHOULDER_TORQUE_OFFSET, 1.0),),
+        n_runs=2,
+    )
+
+    with pytest.raises(ContractViolationError, match="context-specific"):
+        run_variation(plan)
+
+
+def test_simulation_config_rejects_other_source_kind_and_overlong_window() -> None:
+    valid = DoublePendulumRunConfig(
+        commanded_torque_offsets=(
+            LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.02, 0.04), 1.0),
+        )
+    )
+    overlong = DoublePendulumRunConfig(
+        commanded_torque_offsets=(
+            LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.19, 0.21), 1.0),
+        )
+    )
+
+    with pytest.raises(ContractViolationError, match="double-pendulum"):
+        dataclasses.replace(
+            _base_config(), source_kind="manual", swing_run_config=valid
+        )
+    with pytest.raises(ContractViolationError, match="run duration"):
+        dataclasses.replace(_base_config(), swing_run_config=overlong)
+
+
+def test_builder_rejects_locus_beyond_effective_rk4_grid_before_sampling() -> None:
+    plan = VariationPlan(
+        mode="swing",
+        noise=(
+            _localized_spec(
+                _SHOULDER_TORQUE_OFFSET,
+                SHOULDER_JOINT_ID,
+                window=(0.2, 0.2002),
+            ),
+        ),
+        n_runs=2,
+    )
+    base = dataclasses.replace(_base_config(), swing_duration_s=0.2004)
+
+    with pytest.raises(ContractViolationError, match="effective RK4 duration"):
+        build_simulation_ensemble_request(plan, base)
diff --git a/tests/rate_of_closure/test_torque_run_history.py b/tests/rate_of_closure/test_torque_run_history.py
index 67010758f..59c436281 100644
--- a/tests/rate_of_closure/test_torque_run_history.py
+++ b/tests/rate_of_closure/test_torque_run_history.py
@@ -26,6 +26,7 @@
     WRIST_JOINT_ID,
     DoublePendulumRunConfig,
     JointLockConfig,
+    LocalizedTorqueOffset,
 )
 from shared.python.swing_sim.torque_library import TorqueProfileLibrary
 from shared.python.swing_sim.torque_profiles import (
@@ -275,3 +276,92 @@ def test_source_factory_rejects_joint_locks_for_unsupported_triple_source() -> N
                 joint_locks=JointLockConfig((WRIST_JOINT_ID,))
             ),
         )
+
+
+@pytest.mark.parametrize("source_kind", ["manual", "triple_pendulum"])
+def test_source_factory_rejects_localized_torque_for_unsupported_source(
+    source_kind: str,
+) -> None:
+    from rate_of_closure.simulation.sources import make_source
+
+    run_config = DoublePendulumRunConfig(
+        commanded_torque_offsets=(
+            LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.01, 0.02), 1.0),
+        )
+    )
+
+    with pytest.raises(ContractViolationError, match="torque offsets.*unsupported"):
+        make_source(source_kind, _SCENARIO, run_config=run_config)
+
+
+@pytest.mark.parametrize(
+    "run_config",
+    [
+        {},
+        [],
+        0,
+        False,
+        "",
+        {"bad": 1},
+        [1],
+        pytest.param(object(), id="object"),
+    ],
+)
+def test_source_factory_rejects_wrong_and_falsey_non_config_objects(
+    run_config: object,
+) -> None:
+    from rate_of_closure.simulation.sources import make_source
+
+    with pytest.raises(ContractViolationError, match="run_config"):
+        make_source(  # type: ignore[arg-type]
+            "manual", _SCENARIO, run_config=run_config
+        )
+
+
+@pytest.mark.parametrize("source_kind", ["manual", "triple_pendulum"])
+@pytest.mark.parametrize(
+    ("run_config", "message"),
+    [
+        (
+            DoublePendulumRunConfig.prescribed("profile.test.v1"),
+            "execution.*unsupported",
+        ),
+        (
+            DoublePendulumRunConfig(joint_locks=JointLockConfig((WRIST_JOINT_ID,))),
+            "locks.*unsupported",
+        ),
+        (
+            DoublePendulumRunConfig(
+                commanded_torque_offsets=(
+                    LocalizedTorqueOffset(SHOULDER_JOINT_ID, (0.01, 0.02), 1.0),
+                )
+            ),
+            "torque offsets.*unsupported",
+        ),
+    ],
+)
+def test_source_factory_rejects_all_nondefault_execution_for_unsupported_sources(
+    source_kind: str,
+    run_config: DoublePendulumRunConfig,
+    message: str,
+) -> None:
+    from rate_of_closure.simulation.sources import make_source
+
+    with pytest.raises(ContractViolationError, match=message):
+        make_source(source_kind, _SCENARIO, run_config=run_config)
+
+
+@pytest.mark.parametrize("source_kind", ["manual", "triple_pendulum"])
+def test_source_factory_accepts_explicit_default_execution_for_every_source(
+    source_kind: str,
+) -> None:
+    from rate_of_closure.simulation.sources import make_source
+
+    source = make_source(
+        source_kind,
+        _SCENARIO,
+        duration=0.05,
+        run_config=DoublePendulumRunConfig(),
+    )
+
+    assert source.sample(0.0).pose.shape == (4, 4)
diff --git a/tests/rate_of_closure/test_variation_gui.py b/tests/rate_of_closure/test_variation_gui.py
index 06a81b3d7..4502b0d86 100644
--- a/tests/rate_of_closure/test_variation_gui.py
+++ b/tests/rate_of_closure/test_variation_gui.py
@@ -30,12 +30,14 @@
     VariationPlan,
     keys_for_mode,
     run_variation,
+    variable_registry,
 )
 from shared.python.swing_sim.variation.dataset_io import read_json  # noqa: E402
 
 pytestmark = [pytest.mark.unit, pytest.mark.headless_safe]
 
 _BALL = f"{CATEGORY_LAUNCH}.ball_speed_mph"
+_SHOULDER_TORQUE = f"{CATEGORY_SWING}.shoulder_commanded_torque_offset_nm"
 
 
 @pytest.fixture
@@ -85,6 +87,19 @@ def test_mode_switch_repopulates_rows(self, tab: VariationTab) -> None:
         keys = tuple(row.variable.itemData(i) for i in range(row.variable.count()))
         assert keys == keys_for_mode("launch")
 
+    def test_swing_picker_hides_contextual_variables_without_locus_editor(
+        self, tab: VariationTab
+    ) -> None:
+        tab._mode_combo.setCurrentIndex(1)  # swing
+        row = tab._rows[0]
+        keys = tuple(row.variable.itemData(i) for i in range(row.variable.count()))
+
+        assert _SHOULDER_TORQUE not in keys
+        assert all(
+            variable_registry()[key].applicability != "localized_torque_only"
+            for key in keys
+        )
+
     def test_add_and_remove_rows_keeps_at_least_one(self, tab: VariationTab) -> None:
         tab._add_row()
         assert len(tab._rows) == 2
@@ -182,6 +197,28 @@ def test_load_plan_round_trips_including_base_and_truncation(
         tab.load_plan(plan)
         assert tab.build_plan() == plan
 
+    def test_load_contextual_plan_fails_atomically_with_locus_editor_message(
+        self, tab: VariationTab
+    ) -> None:
+        original = tab.build_plan()
+        plan = VariationPlan(
+            mode="swing",
+            noise=(
+                NoiseSpec(
+                    _SHOULDER_TORQUE,
+                    scale=1.0,
+                    time_window_s=(0.02, 0.04),
+                    point_ids=("joint.shoulder",),
+                ),
+            ),
+            n_runs=4,
+        )
+
+        with pytest.raises(ValueError, match="locus editor.*not representable"):
+            tab.load_plan(plan)
+
+        assert tab.build_plan() == original
+
     def test_v2_plan_round_trips_custom_ids_loci_groups_and_save(
         self, tab: VariationTab, tmp_path: Path, monkeypatch
     ) -> None:  # type: ignore[no-untyped-def]
diff --git a/tests/rate_of_closure/test_variation_simulation_request.py b/tests/rate_of_closure/test_variation_simulation_request.py
index 0c1c7a677..b2ce9f322 100644
--- a/tests/rate_of_closure/test_variation_simulation_request.py
+++ b/tests/rate_of_closure/test_variation_simulation_request.py
@@ -8,7 +8,11 @@
 
 from rate_of_closure.club import get_club
 from rate_of_closure.model import ImpactScenario
-from rate_of_closure.simulation import BallSetup, BallSupportMode, SimulationConfig
+from rate_of_closure.simulation import (
+    BallSetup,
+    BallSupportMode,
+    SimulationConfig,
+)
 from rate_of_closure.variation.request_builder import (
     apply_global_simulation_values,
 )
@@ -38,6 +42,8 @@ def _key(category: str, name: str) -> str:
 _FORWARD_TILT = _key(CATEGORY_SWING, "forward_tilt_deg")
 _DAMPING_SHOULDER = _key(CATEGORY_SWING, "damping_shoulder")
 _DAMPING_WRIST = _key(CATEGORY_SWING, "damping_wrist")
+_SHOULDER_TORQUE_OFFSET = _key(CATEGORY_SWING, "shoulder_commanded_torque_offset_nm")
+_WRIST_TORQUE_OFFSET = _key(CATEGORY_SWING, "wrist_commanded_torque_offset_nm")
 _IMPACT_OFFSET = _key(CATEGORY_SWING, "impact_time_offset_s")
 _TOE = _key(CATEGORY_DELIVERY, "impact_offset_toe_mm")
 _HIGH = _key(CATEGORY_DELIVERY, "impact_offset_high_mm")
@@ -61,6 +67,22 @@ def _spec(variable_key: str, scale: float) -> NoiseSpec:
     return NoiseSpec(variable_key, distribution="uniform", scale=scale)
 
 
+def _localized_spec(
+    variable_key: str,
+    point_id: str,
+    *,
+    window: tuple[float, float] = (0.02, 0.04),
+    scale: float = 2.0,
+) -> NoiseSpec:
+    return NoiseSpec(
+        variable_key,
+        distribution="uniform",
+        scale=scale,
+        time_window_s=window,
+        point_ids=(point_id,),
+    )
+
+
 def test_builder_maps_supported_samples_into_complete_configs() -> None:
     plan = VariationPlan(
         mode="swing",
@@ -157,7 +179,7 @@ def test_global_value_seam_rejects_coercive_or_nonfinite_values(value: object) -
         apply_global_simulation_values(_base_config(), {_YAW: value})  # type: ignore[dict-item]
 
 
-def test_builder_rejects_non_global_or_unmapped_variables() -> None:
+def test_builder_rejects_unsupported_localized_or_unmapped_variables() -> None:
     localized = VariationPlan(
         mode="swing",
         noise=(
@@ -174,7 +196,7 @@ def test_builder_rejects_non_global_or_unmapped_variables() -> None:
         n_runs=2,
     )
 
-    with pytest.raises(ContractViolationError, match="global perturbations"):
+    with pytest.raises(ContractViolationError, match="localized"):
         build_simulation_ensemble_request(localized, _base_config())
     with pytest.raises(ContractViolationError, match="not trace-capable"):
         build_simulation_ensemble_request(unsupported, _base_config())