Fix locked exposures and repeated 1x captures; prepare beta.19

coolscanpy/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "coolscanpy"
-version = "0.7.10"
+version = "0.7.11"
 description = "Direct-USB acquisition library for Nikon Coolscan film scanners with a python-sane-style API — roll scanning, RGBI capture, IR plane, scan receipts"
 readme = "README.md"
 license = "GPL-3.0-only"

coolscanpy/src/coolscanpy/_roll.py

@@ -51,6 +51,7 @@
     SinglePassFinalizationResult,
     SinglePassSession,
     SinglePassWorkflowError,
+    _exposure_authority_is_bound,
 )
 from coolscanpy.exceptions import (
     AdapterUnsupported,
@@ -2651,24 +2652,13 @@ def _read_exact_analyzer_source(
         raise BatchIntegrityError(
             "settled third meter pass is not bound to the accepted controller result"
         )
-    # Since the guarded nikon-parity solve became the RGB command authority,
-    # the commanded contract is bound to the active controller's accepted
-    # solve THROUGH the journaled authority record: active solve -> authority
-    # -> commanded contract, with infrared passing through unchanged.
-    authority = journal.get("active_exposure_authority")
-    if (
-        type(authority) is not dict
-        or authority.get("rgb_source") != "nikon-parity-guarded-v2"
-        or authority.get("ir_source") != "active-controller"
-        or authority.get("commanded_channels_raw_10ns") != final_controller
-        or authority.get("active_controller_channels_raw_10ns")
-        != controller.get("final_exposures_raw_10ns")
-        or type(controller.get("final_exposures_raw_10ns")) is not dict
-        or final_controller.get("IR")
-        != controller["final_exposures_raw_10ns"].get("IR")
+    if not _exposure_authority_is_bound(
+        journal,
+        final_controller,
+        controller.get("final_exposures_raw_10ns"),
     ):
         raise BatchIntegrityError(
-            "commanded exposure contract is not bound to the parity authority "
+            "commanded exposure contract is not bound to the declared authority "
             "and the accepted controller result"
         )
 

coolscanpy/src/coolscanpy/capture/single_pass_workflow.py

@@ -60,6 +60,7 @@
     decode_full_records,
     decode_records,
 )
+from coolscanpy.protocol.ls5000_single_pass.meter import EXPOSURE_MAX, EXPOSURE_MIN
 from coolscanpy.protocol.ls5000_single_pass.streaming_sidecar import (
     CANONICAL_RGB_AVERAGE,
     STREAM_DATA_SUFFIX,
@@ -121,6 +122,48 @@
 )
 
 
+def _exposure_authority_is_bound(
+    journal: Mapping[str, Any],
+    commanded: Mapping[str, int],
+    active_solve: object,
+) -> bool:
+    authority = journal.get("active_exposure_authority")
+    if (
+        type(authority) is not dict
+        or authority.get("ir_source") != "active-controller"
+        or authority.get("commanded_channels_raw_10ns") != commanded
+        or type(active_solve) is not dict
+        or authority.get("active_controller_channels_raw_10ns") != active_solve
+        or commanded.get("IR") != active_solve.get("IR")
+    ):
+        return False
+    source = authority.get("rgb_source")
+    if source == "nikon-parity-guarded-v2":
+        return True
+    if source != "explicit-rgb-override":
+        return False
+    override = journal.get("exposure_override")
+    forced = override.get("forced_10ns") if type(override) is dict else None
+    if (
+        type(override) is not dict
+        or override.get("applied") is not True
+        or type(forced) is not dict
+        or set(forced) != {"red", "green", "blue"}
+        or authority.get("device_exposure_bounds_raw_10ns")
+        != [EXPOSURE_MIN, EXPOSURE_MAX]
+    ):
+        return False
+    for name, channel in (("red", "R"), ("green", "G"), ("blue", "B")):
+        value = forced[name]
+        if (
+            type(value) is not int
+            or not EXPOSURE_MIN <= value <= EXPOSURE_MAX
+            or value != commanded.get(channel)
+        ):
+            return False
+    return True
+
+
 class SinglePassWorkflowError(RuntimeError):
     """A capture cannot safely become a committed NegPy artifact set."""
 
@@ -1476,24 +1519,15 @@ def _validate_completed_capture(
         }
         if controller_exposures != expected_controller:
             raise SinglePassIntegrityError("meter controller exposure echo is inconsistent")
-        # Guarded nikon-parity is the RGB command authority: the commanded
-        # contract is bound to the active controller's accepted solve THROUGH
-        # the journaled authority record (active solve -> authority ->
-        # commanded wire echo), with infrared passing through unchanged.
-        # Mirrors the roll publication consumer's binding check.
         authority = journal.get("active_exposure_authority")
         active_solve = controller.get("final_exposures_raw_10ns")
-        if (
-            not isinstance(authority, dict)
-            or authority.get("rgb_source") != "nikon-parity-guarded-v2"
-            or authority.get("ir_source") != "active-controller"
-            or authority.get("commanded_channels_raw_10ns") != expected_controller
-            or not isinstance(active_solve, dict)
-            or authority.get("active_controller_channels_raw_10ns") != active_solve
-            or expected_controller.get("IR") != active_solve.get("IR")
+        if not _exposure_authority_is_bound(
+            journal,
+            expected_controller,
+            active_solve,
         ):
             raise SinglePassIntegrityError(
-                "commanded exposure contract is not bound to the parity authority "
+                "commanded exposure contract is not bound to the declared authority "
                 "and the accepted controller result"
             )
 

coolscanpy/src/coolscanpy/protocol/ls5000_single_pass/bundle.py

@@ -28,6 +28,7 @@
 # density-source cap-0x10d/f03 exposures, the proven 97-dpi reservation-preview
 # evidence, runtime arithmetic gate, and exact per-frame ownership receipt.
 CAPTURE_BUNDLE_COMPONENT_SHA256 = {
+    # Resealed 2026-09-08: resumed batches validate the original capture plan.
     # Resealed 2026-09-08: held-child non-motion film-status rendezvous.
     # 0.7.9 preview: held metering/explicit refusal skips plus retained local
     # timing, required scanner identity and adapter-conditional replay.
@@ -63,7 +64,7 @@
     # Resealed 2026-08-11: capture descendants remain in the bridge-owned
     # process group and inherit its exclusive process-ownership fence.
     "capture_process.py": "e66a50d07a4704a10f38f3d477432e0e2b3853c23c43a70e740950bd2dccb2ea",
-    "worker.py": "41588f69eb75ac67c2f8a2bb86f8bf175ae9949109c06ca9256115a7362ecd41",
+    "worker.py": "e8bd899f166042ca7d6cc2678df7f585a47d2e7ac38d131f3c65f9ffdeec502c",
     "manual_frames.py": "8fc4ba82c177e1b7ecd6943354db33930b468ef3b4b82c712202b8caea54c9bb",
     "usb_backend.py": "afb5b3cbb57404b758f4f8d8795f4307c07c8f6d01bbeccb3ced38026787fd62",
     "density.py": "c2c47de2886bc4b60197d2721b6d72050a76f1095760590fa7bb34a728b9da76",

coolscanpy/src/coolscanpy/protocol/ls5000_single_pass/worker.py

@@ -8148,7 +8148,7 @@ def run_live_capture(
                 derive_equivalent_continuation_blocks(continuation_plan)
                 batch_job = next_batch_job
                 batch_selections = _derive_live_batch_selections(
-                    active_plan,
+                    plan,
                     preview_bytes,
                     live_sub_8e_table,
                     batch_job.frames,

coolscanpy/tests/capture/test_ls5000_single_pass_workflow.py

@@ -888,25 +888,25 @@ def assessor(rgb: np.ndarray, *, dpi: int) -> StoppedTransportSmearAssessment:
         ),
         (
             lambda journal: journal.__delitem__("active_exposure_authority"),
-            "parity authority",
+            "declared authority",
         ),
         (
             lambda journal: journal["active_exposure_authority"][
                 "commanded_channels_raw_10ns"
             ].__setitem__("G", 1),
-            "parity authority",
+            "declared authority",
         ),
         (
             lambda journal: journal["active_exposure_authority"][
                 "active_controller_channels_raw_10ns"
             ].__setitem__("R", 1),
-            "parity authority",
+            "declared authority",
         ),
         (
             lambda journal: journal["meter_controller_final_result"][
                 "final_exposures_raw_10ns"
             ].__setitem__("IR", 1),
-            "parity authority",
+            "declared authority",
         ),
     ],
 )
@@ -973,6 +973,67 @@ def test_parity_commanded_exposures_differing_from_active_solve_finalize(
     assert authority["active_controller_channels_raw_10ns"] == active
 
 
+def _record_explicit_override(journal: dict[str, Any]) -> None:
+    commanded = journal["active_exposure_authority"][
+        "commanded_channels_raw_10ns"
+    ]
+    journal["active_exposure_authority"]["rgb_source"] = "explicit-rgb-override"
+    journal["exposure_override"] = {
+        "applied": True,
+        "forced_10ns": {
+            "red": commanded["R"],
+            "green": commanded["G"],
+            "blue": commanded["B"],
+        },
+        "metered_10ns": {
+            "red": commanded["R"],
+            "green": commanded["G"],
+            "blue": commanded["B"],
+        },
+    }
+
+
+def test_explicit_override_authority_finalizes_when_exactly_bound(
+    tmp_path: Path,
+) -> None:
+    attempt, _stream = _attempt(tmp_path)
+    journal = json.loads(attempt.journal_path.read_text(encoding="utf-8"))
+    _record_explicit_override(journal)
+    attempt.journal_path.write_text(json.dumps(journal), encoding="utf-8")
+
+    result = _workflow().finalize_attempt(attempt, delete_scratch=False)
+
+    authority = result.manifest["exposure_evidence"]["active_exposure_authority"]
+    assert authority["rgb_source"] == "explicit-rgb-override"
+
+
+def test_explicit_override_authority_refuses_tampered_forced_ticks(
+    tmp_path: Path,
+) -> None:
+    attempt, stream = _attempt(tmp_path)
+    journal = json.loads(attempt.journal_path.read_text(encoding="utf-8"))
+    _record_explicit_override(journal)
+    journal["exposure_override"]["forced_10ns"]["red"] += 1
+    attempt.journal_path.write_text(json.dumps(journal), encoding="utf-8")
+
+    with pytest.raises(SinglePassIntegrityError, match="authority"):
+        _workflow().finalize_attempt(attempt)
+
+    assert stream.is_file()
+
+
+def test_legacy_parity_labeled_override_still_finalizes(tmp_path: Path) -> None:
+    attempt, _stream = _attempt(tmp_path)
+    journal = json.loads(attempt.journal_path.read_text(encoding="utf-8"))
+    _record_explicit_override(journal)
+    journal["active_exposure_authority"]["rgb_source"] = (
+        "nikon-parity-guarded-v2"
+    )
+    attempt.journal_path.write_text(json.dumps(journal), encoding="utf-8")
+
+    _workflow().finalize_attempt(attempt, delete_scratch=False)
+
+
 def test_stream_hash_mismatch_retains_scratch_before_decode(tmp_path: Path) -> None:
     attempt, stream = _attempt(tmp_path)
     stream.write_bytes(b"87654321")

coolscanpy/tests/protocol/ls5000_single_pass/test_worker.py

@@ -7062,14 +7062,15 @@ def dispose_resources(_device: object) -> None:
             pass
 
     def derive_batch(
-        _plan: list[dict],
+        selection_plan: list[dict],
         preview: bytes,
         table: bytes,
         frames: tuple,
         *,
         reviewed_fingerprint: ReviewedRollFingerprint,
         manual_boundary_rows: tuple[int, ...] | None = None,
     ) -> tuple:
+        worker_module.validate_plan(selection_plan)
         assert len(preview) == len(worker_module.PREVIEW_READ_SEQUENCES)
         assert table == header_8e
         assert len(frames) == 1

coolscanpy/tests/protocol/ls5000_single_pass/test_worker_streaming_integration.py

@@ -403,15 +403,8 @@ def test_continuation_loop_wires_hook_and_binds_finish_to_raw_sha(
     assert log["finish_calls"] == [(0, hashlib.sha256(b"x").hexdigest(), 1)]
 
 
-def test_continuation_meter_evidence_is_accepted_by_publication_consumer(
-    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
-) -> None:
-    journal, second = _drive_continuation(
-        tmp_path,
-        monkeypatch,
-        open_session=lambda *_a: None,
-        full_meter_payload=True,
-    )
+def _publication_consumer_inputs(tmp_path: Path, journal, second):
+    second.journal.write_text(json.dumps(journal), encoding="utf-8")
     journal_payload = second.journal.read_bytes()
     attempt = SinglePassAttempt(
         session=SinglePassSession(root=tmp_path, session_id="continuation-consumer"),
@@ -441,6 +434,19 @@ def test_continuation_meter_evidence_is_accepted_by_publication_consumer(
         resumed=False,
         scratch_deleted=False,
     )
+    return attempt, finalization
+
+
+def test_continuation_meter_evidence_is_accepted_by_publication_consumer(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    journal, second = _drive_continuation(
+        tmp_path,
+        monkeypatch,
+        open_session=lambda *_a: None,
+        full_meter_payload=True,
+    )
+    attempt, finalization = _publication_consumer_inputs(tmp_path, journal, second)
 
     meter_rgbi, final_rgb = roll_module._read_exact_analyzer_source(
         attempt, finalization
@@ -460,6 +466,63 @@ def test_continuation_meter_evidence_is_accepted_by_publication_consumer(
     )
 
 
+def test_publication_consumer_accepts_bound_explicit_override(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    journal, second = _drive_continuation(
+        tmp_path,
+        monkeypatch,
+        open_session=lambda *_a: None,
+        full_meter_payload=True,
+    )
+    commanded = journal["active_exposure_authority"][
+        "commanded_channels_raw_10ns"
+    ]
+    journal["active_exposure_authority"]["rgb_source"] = "explicit-rgb-override"
+    journal["exposure_override"] = {
+        "applied": True,
+        "forced_10ns": {
+            "red": commanded["R"],
+            "green": commanded["G"],
+            "blue": commanded["B"],
+        },
+    }
+    attempt, finalization = _publication_consumer_inputs(tmp_path, journal, second)
+
+    _meter_rgbi, final_rgb = roll_module._read_exact_analyzer_source(
+        attempt, finalization
+    )
+
+    assert final_rgb == (commanded["R"], commanded["G"], commanded["B"])
+
+
+def test_publication_consumer_refuses_tampered_explicit_override(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    journal, second = _drive_continuation(
+        tmp_path,
+        monkeypatch,
+        open_session=lambda *_a: None,
+        full_meter_payload=True,
+    )
+    commanded = journal["active_exposure_authority"][
+        "commanded_channels_raw_10ns"
+    ]
+    journal["active_exposure_authority"]["rgb_source"] = "explicit-rgb-override"
+    journal["exposure_override"] = {
+        "applied": True,
+        "forced_10ns": {
+            "red": commanded["R"] + 1,
+            "green": commanded["G"],
+            "blue": commanded["B"],
+        },
+    }
+    attempt, finalization = _publication_consumer_inputs(tmp_path, journal, second)
+
+    with pytest.raises(roll_module.BatchIntegrityError, match="authority"):
+        roll_module._read_exact_analyzer_source(attempt, finalization)
+
+
 def test_synchronous_decoder_exception_never_aborts_continuation_capture(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch
 ) -> None: