fix(rc05): preserve multi-operation canary progress

yuqei-infra/scripts/multiscene_call_ledger.py

@@ -95,6 +95,11 @@ class CallLedger:
         self.candidate_binding = _json_mapping(candidate_binding, "candidate binding")
         self.release_controls = _release_controls(release_controls, self.case_limits)
         self.controls_enabled = self.release_controls is not None
+        # This is deliberately process-local.  The same runner instance may
+        # advance a declared multi-operation case between immutable provider
+        # finishes.  A restarted or concurrent runner has no recoverable
+        # in-memory workflow state and must therefore treat that gap as unknown.
+        self._active_case_ids: set[str] = set()
         if registry_root is not None:
             if not real_provider or historical_usage is None:
                 raise ValueError("global registry is only valid for a receipt-bound real run")
@@ -320,7 +325,12 @@ class CallLedger:
             else:
                 _write_x(path, payload)
 
-    def _materialize_attempt_breakers_unlocked(self) -> None:
+    def _materialize_attempt_breakers_unlocked(
+        self,
+        *,
+        progress_case_id: str | None = None,
+        progress_operation: str | None = None,
+    ) -> None:
         if not self.controls_enabled:
             return
         records = self._records()
@@ -328,6 +338,10 @@ class CallLedger:
         for case_id in cases:
             relevant = [record for (current, _operation), record in records.items()
                         if current == case_id]
+            expected_operations = self.release_controls["expected_operations"][case_id]
+            actual_operations = {
+                operation for current, operation in records if current == case_id
+            }
             unfinished = any(record["finish"] is None for record in relevant)
             unsuccessful = next((record["finish"]["status"] for record in relevant
                                  if record["finish"] is not None
@@ -336,6 +350,17 @@ class CallLedger:
                 status, reason = "unknown", "ledger has an unfinished provider attempt"
             elif unsuccessful is not None:
                 status, reason = unsuccessful, f"provider attempt finished as {unsuccessful}"
+            elif actual_operations - set(expected_operations):
+                status, reason = "unknown", "case contains an operation outside frozen controls"
+            elif (case_id == progress_case_id
+                  and progress_operation in expected_operations
+                  and progress_operation not in actual_operations
+                  and not (self.case_outcomes / f"{case_id}.json").exists()):
+                # Only the same live runner instance may cross the small
+                # finish->next-claim gap.  A restart cannot reconstruct the
+                # isolated case workflow, so it must materialize an unknown
+                # breaker rather than silently continue or repeat a call.
+                continue
             else:
                 try:
                     self._validated_persisted_outcome_unlocked(
@@ -431,11 +456,17 @@ class CallLedger:
             raise CallLedgerError("case is not in the frozen ledger limits")
         if operation not in OPERATIONS or operation not in self.operation_limits:
             raise CallLedgerError("operation is not enabled for this ledger")
+        if (self.controls_enabled
+                and operation not in self.release_controls["expected_operations"][case_id]):
+            raise CallLedgerError("operation is outside frozen controls for this case")
         request_sha256 = _sha(request_sha256, "request_sha256")
         stem = f"{case_id}--{operation}"
         with self._locked():
             if self.controls_enabled:
-                self._materialize_attempt_breakers_unlocked()
+                self._materialize_attempt_breakers_unlocked(
+                    progress_case_id=(case_id if case_id in self._active_case_ids else None),
+                    progress_operation=operation,
+                )
                 state = self._control_snapshot_unlocked()
                 failure_class = self.release_controls["failure_classes"][case_id]
                 if failure_class in state["blocked_failure_classes"]:
@@ -484,6 +515,7 @@ class CallLedger:
                     "request_sha256": request_sha256,
                 })
             _write_x(self.attempts / f"{stem}.claim.json", claim_payload)
+            self._active_case_ids.add(case_id)
             return claim_id
 
     def finish(self, token: str, status: str, response_sha256: str | None = None) -> None:
@@ -730,6 +762,9 @@ class CallLedger:
         outcome_sha256 = _sha(outcome_sha256, "case outcome SHA-256")
         evidence = _evidence_hashes(evidence_sha256s)
         expected = _operation_projection(expected_operations)
+        if (self.controls_enabled
+                and expected != self.release_controls["expected_operations"][case_id]):
+            raise CallLedgerError("case outcome operations differ from frozen controls")
         with self._locked():
             records = self._records()
             projection = _case_attempt_projection(records, case_id)
@@ -747,6 +782,7 @@ class CallLedger:
                 "evidence_sha256s": evidence,
             }
             _write_x(self.case_outcomes / f"{case_id}.json", payload)
+            self._active_case_ids.discard(case_id)
 
     def verify_case_outcome(
         self,
@@ -763,6 +799,10 @@ class CallLedger:
         outcome_sha256 = _sha(outcome_sha256, "case outcome SHA-256")
         evidence = _evidence_hashes(evidence_sha256s)
         declared_operations = _operation_projection(expected_operations)
+        if (self.controls_enabled
+                and declared_operations
+                != self.release_controls["expected_operations"][case_id]):
+            raise CallLedgerError("case outcome operations differ from frozen controls")
         with self._locked():
             path = self.case_outcomes / f"{case_id}.json"
             _reject_link_or_reparse(path, "case outcome binding")
@@ -825,6 +865,10 @@ class CallLedger:
             raise CallLedgerError("case outcome binding file is invalid")
         outcome_sha = _sha(payload.get("outcome_sha256"), "case outcome SHA-256")
         expected_operations = _operation_projection(payload.get("expected_operations"))
+        if (self.controls_enabled
+                and expected_operations
+                != self.release_controls["expected_operations"][case_id]):
+            raise CallLedgerError("persisted case operations differ from frozen controls")
         evidence = _evidence_hashes(payload.get("evidence_sha256s"))
         projection = _case_attempt_projection(records, case_id)
         expected = {
@@ -1113,21 +1157,33 @@ def _release_controls(value: Mapping[str, object] | None,
     normalized = _json_mapping(value, "release controls")
     if normalized is None:
         return None
-    if set(normalized) != {"schema_version", "canary_case_ids", "failure_classes"}:
+    if set(normalized) != {
+        "schema_version", "canary_case_ids", "failure_classes", "expected_operations",
+    }:
         raise ValueError("release controls shape is invalid")
-    if normalized["schema_version"] != "multiscene-release-controls.v1":
+    if normalized["schema_version"] != "multiscene-release-controls.v2":
         raise ValueError("release controls schema is unsupported")
     canaries = normalized["canary_case_ids"]
     classes = normalized["failure_classes"]
+    expected_operations = normalized["expected_operations"]
     if (not isinstance(canaries, list) or not canaries
             or len(canaries) != len(set(canaries))
             or any(case_id not in case_limits for case_id in canaries)):
         raise ValueError("release control canary inventory is invalid")
     if not isinstance(classes, dict) or set(classes) != set(case_limits):
         raise ValueError("release control failure-class inventory is invalid")
+    if (not isinstance(expected_operations, dict)
+            or set(expected_operations) != set(case_limits)):
+        raise ValueError("release control operation inventory is invalid")
     for case_id, failure_class in classes.items():
         _identifier(case_id, "release control case ID")
         _identifier(failure_class, "release control failure class")
+        operations = expected_operations[case_id]
+        if (not isinstance(operations, list) or not operations
+                or len(operations) != len(set(operations))
+                or len(operations) > case_limits[case_id]
+                or any(operation not in OPERATIONS for operation in operations)):
+            raise ValueError("release control expected operations are invalid")
     return normalized
 
 

yuqei-infra/scripts/multiscene_eval_plan.py

@@ -25,7 +25,7 @@ from multiscene_supplemental_cases import build_supplemental_cases
 ROOT = Path(__file__).resolve().parents[2]
 SCHEMA_VERSION = "multiscene-evaluation-plan.v2"
 CANDIDATE_SCHEMA_VERSION = "multiscene-candidate-binding.v1"
-RELEASE_CONTROLS_SCHEMA_VERSION = "multiscene-release-controls.v1"
+RELEASE_CONTROLS_SCHEMA_VERSION = "multiscene-release-controls.v2"
 REQUIRED_IMAGE_ROLES = (
     "contract-web", "contract-api", "ai-platform-api", "ai-worker",
 )
@@ -479,6 +479,9 @@ def build_plan(runner, candidate_binding: Mapping[str, Any]) -> dict[str, Any]:
             "schema_version": RELEASE_CONTROLS_SCHEMA_VERSION,
             "canary_case_ids": list(CANARY_CASE_IDS),
             "failure_classes": {case["case_id"]: case["failure_class"] for case in cases},
+            "expected_operations": {
+                case["case_id"]: list(case["operations"]) for case in cases
+            },
         },
         "source_pins": _source_pins(runner), "cases": cases,
         "limitations": ["合成隔离评测，不是生产发布或法律批准。",

yuqei-infra/tests/test_multiscene_call_ledger.py

@@ -72,11 +72,15 @@ def ledger(tmp_path, **overrides):
     return LEDGER.CallLedger(**values)
 
 
-def _controls(case_classes=None, canaries=None):
+def _controls(case_classes=None, canaries=None, expected_operations=None):
     case_classes = case_classes or {"purchase": "contract", "service": "specialist"}
-    return {"schema_version": "multiscene-release-controls.v1",
+    expected_operations = expected_operations or {
+        case_id: ["draft"] for case_id in case_classes
+    }
+    return {"schema_version": "multiscene-release-controls.v2",
             "canary_case_ids": canaries or ["purchase"],
-            "failure_classes": case_classes}
+            "failure_classes": case_classes,
+            "expected_operations": expected_operations}
 
 
 def _write_json(path, value):
@@ -285,7 +289,9 @@ def test_persisted_canary_transition_is_strict_and_revalidated(tmp_path, mutatio
 def test_canary_missing_mandatory_operation_cannot_authorize_full(tmp_path):
     book = ledger(tmp_path, operation_limits={"draft": 1, "semantic_review": 1},
                   case_limits={"purchase": 2},
-                  release_controls=_controls({"purchase": "contract"}, ["purchase"]))
+                  release_controls=_controls(
+                      {"purchase": "contract"}, ["purchase"],
+                      {"purchase": ["draft", "semantic_review"]}))
     book.authorize_selection(["purchase"], resume=False, explicit_selection=True)
     token = book.claim("purchase", "draft", SHA_A)
     book.finish(token, "succeeded", SHA_B)
@@ -297,6 +303,58 @@ def test_canary_missing_mandatory_operation_cannot_authorize_full(tmp_path):
     assert book.snapshot()["unknown"] == 1
 
 
+def test_multi_operation_case_advances_between_declared_operations(tmp_path):
+    controls = _controls(
+        {"purchase": "contract", "service": "specialist"},
+        ["purchase"],
+        {"purchase": ["draft", "semantic_review"], "service": ["draft"]},
+    )
+    values = dict(
+        operation_limits={"draft": 2, "semantic_review": 1},
+        case_limits={"purchase": 2, "service": 1},
+        release_controls=controls,
+    )
+    first = ledger(tmp_path, **values)
+    first.authorize_selection(["purchase"], resume=False, explicit_selection=True)
+    token = first.claim("purchase", "draft", SHA_A)
+    first.finish(token, "succeeded", SHA_B)
+
+    token = first.claim("purchase", "semantic_review", SHA_A)
+    first.finish(token, "succeeded", SHA_B)
+    outcome_sha = _bind_success(
+        first, tmp_path, "purchase", ["draft", "semantic_review"])
+    first.complete_canary({"purchase": outcome_sha})
+    assert first.control_snapshot()["phase"] == "full_authorized"
+
+
+def test_restart_between_multi_operation_steps_fails_closed(tmp_path):
+    controls = _controls(
+        {"purchase": "contract", "service": "specialist"},
+        ["purchase"],
+        {"purchase": ["draft", "semantic_review"], "service": ["draft"]},
+    )
+    values = dict(
+        operation_limits={"draft": 2, "semantic_review": 1},
+        case_limits={"purchase": 2, "service": 1},
+        release_controls=controls,
+    )
+    first = ledger(tmp_path, **values)
+    first.authorize_selection(["purchase"], resume=False, explicit_selection=True)
+    token = first.claim("purchase", "draft", SHA_A)
+    first.finish(token, "succeeded", SHA_B)
+
+    resumed = ledger(tmp_path, **values)
+    with pytest.raises(LEDGER.CallLedgerError, match="blocked.*failure class"):
+        resumed.authorize_selection(["purchase"], resume=True, explicit_selection=True)
+    assert resumed.control_snapshot()["blocked_failure_classes"] == ["contract"]
+
+
+def test_case_cannot_claim_operation_outside_frozen_controls(tmp_path):
+    book = ledger(tmp_path, release_controls=_controls())
+    with pytest.raises(LEDGER.CallLedgerError, match="outside frozen controls"):
+        book.claim("purchase", "semantic_review", SHA_A)
+
+
 @pytest.mark.parametrize("mutation", ["result", "evidence", "binding"])
 def test_resume_materializes_invalid_success_outcome_as_unknown_breaker(
     tmp_path, mutation,

yuqei-infra/tests/test_multiscene_eval_plan.py

@@ -38,6 +38,10 @@ def test_plan_freezes_144_cases_280_calls_and_real_request_projections(plan):
     assert plan["schema_version"] == "multiscene-evaluation-plan.v2"
     assert plan["candidate"] == _candidate()
     assert plan["release_controls"]["canary_case_ids"] == list(plan_module.CANARY_CASE_IDS)
+    assert plan["release_controls"]["schema_version"] == "multiscene-release-controls.v2"
+    assert plan["release_controls"]["expected_operations"] == {
+        case["case_id"]: case["operations"] for case in plan["cases"]
+    }
     assert plan["case_count"] == 144
     assert plan["total_provider_call_budget"] == 280
     assert plan["operation_limits"] == plan_module.EXPECTED_OPERATIONS

yuqei-infra/tests/test_multiscene_eval_runner.py

@@ -62,9 +62,12 @@ def _plan(tmp_path, count=3):
         "operation_limits": {"draft": count},
         "source_pins": {"source.py": _sha(source.read_bytes())},
         "candidate": candidate,
-        "release_controls": {"schema_version": "multiscene-release-controls.v1",
+        "release_controls": {"schema_version": "multiscene-release-controls.v2",
             "canary_case_ids": ["case-0"],
-            "failure_classes": {case["case_id"]: "draft" for case in cases}},
+            "failure_classes": {case["case_id"]: "draft" for case in cases},
+            "expected_operations": {
+                case["case_id"]: list(case["operations"]) for case in cases
+            }},
     }
 
 
@@ -446,6 +449,9 @@ def test_real_canary_does_not_transition_before_operation_and_evidence_gates(
         provider_call_budget=2,
         operations=["draft", "semantic_review"],
     )
+    plan["release_controls"]["expected_operations"]["case-0"] = [
+        "draft", "semantic_review",
+    ]
     plan["operation_limits"] = {"draft": 1, "semantic_review": 1}
     calls = _install_modules(monkeypatch, plan)
     registry = tmp_path.parent / f"{tmp_path.name}-rc05-canary-registry"