diff --git a/packages/flow/docs/APPROVAL_MODES.md b/packages/flow/docs/APPROVAL_MODES.md index 963612848..9880806ca 100644 --- a/packages/flow/docs/APPROVAL_MODES.md +++ b/packages/flow/docs/APPROVAL_MODES.md @@ -217,11 +217,30 @@ The gate writes a `switched_off` event to the runner's hash-chained switch log, which applies before the next save, offline too. Cloud can't undo it, and a missing or broken log also reads as off. +### Which problems turn it off + +Every item that needs a person is classed by its cause +(`approval_mode/post.py`, `problem_class`): + +- **An item-input problem leaves the switch on.** The item goes to a person + and doesn't count toward `too_many_problems`. It is one of these, caught + before anything was written: + - a value refused before the first step: a missing required value, or a + typed value that isn't valid, such as a date that isn't an ISO date; + - an input item check (`inputs_supplied`, `inputs_strict`, `channel`); + - a real candidate in the app's possible-duplicate list + ([`DUPLICATE_CHECK.md`](DUPLICATE_CHECK.md)). +- **A program or environment problem turns it off at once.** For example a + wrong write the record check caught, a failed identity check, a + postcondition that failed after delivery, a changed screen, or the wrong + user signed in. Any cause not named above counts here, and so does any + outcome that doesn't consistently read "nothing was written". + | Reason | Trigger in Flow | | --- | --- | | `uncertain_result` | The item ended "Check the record", "Finished, not checked", rolled back, had an unknown or inconsistent outcome, or an earlier item was interrupted. Also when the gate let a save through and its input was delivered, but the outcome says nothing was written: the read-back showed only that the expected record is absent. | -| `too_many_problems` | 3 counted problems in a row, or 5 in the last 20 counted items. Items sent to approval by the item checks or by a stale status don't count. | -| `screen_changed` | P4 or R2 failed. | +| `too_many_problems` | 3 counted problems in a row, or 5 in the last 20 counted items. Items sent to approval by the item checks or by a stale status, and item-input problems, don't count. Every program problem now turns the switch off by itself, so this is a backstop. | +| `screen_changed` | P4 or R2 failed, a failed identity check, or a stop before any effect with no input cause. | | `record_check_lost` | P5 failed, or the engine's own record-check preflight refused the save. | | `version_changed` / `environment_changed` | The bundle, the environment, the runner or the dedicated user changed. | | Any reason in a Cloud "off" status | Cloud turned the switch off. | @@ -233,6 +252,37 @@ While the switch reads off, each item waits for a person to approve it before the save, as in `each_item`, when the run has an approval store. Without one, the item stops before the save and nothing is saved. +### Re-arming takes one signed tap + +A switched-off switch never turns itself back on, under any profile. A person +may re-arm it once these hold: + +- **10 consecutive verified items since the switch went off.** These are + items saved without a person, or approved by one, and confirmed. Any other + finished item resets the count. +- **The person is shown what happened:** the switch-off, its reason, the + recorded cause, and those items. Their approver key signs it + (`approval_mode/switch.py`, `rearm_request`, `sign_rearm`, `rearm`). + +The log's `rearmed` event records: + +- who signed, and the key; +- what they saw; +- the off event it answers. + +A gate honours it only when the signing key is an approver key it trusts. +By default, those are the approvers on the signed approver roster that the +run's pinned root approver vouches for (#1486; the root is pinned by the +deployment config, `--approver-root` or `OPENADAPT_APPROVER_ROOT`). Desktop's +approver key is one of them. With no pinned root, or a roster that doesn't +verify, no re-arm counts. `ApprovalGate(pinned_approver_keys=...)` names the +keys explicitly instead. +Desktop can sign with a key that never leaves its keychain: `rearm_to_sign` +returns the bytes to sign, and `rearm_evidence` wraps the signature. Without the pinned key, a forged, +tampered or early re-arm is ignored, and the switch stays off. Each reader +checks the re-arm again against the log before it. A later problem turns the +switch off again, and the count starts over. + ## Requirements `evaluate_requirements` in `openadapt_flow/approval_mode/requirements.py` diff --git a/packages/flow/openadapt_flow/approval_mode/gate.py b/packages/flow/openadapt_flow/approval_mode/gate.py index a081e698e..da6062d32 100644 --- a/packages/flow/openadapt_flow/approval_mode/gate.py +++ b/packages/flow/openadapt_flow/approval_mode/gate.py @@ -296,6 +296,7 @@ def __init__( approver_policy: Optional["ApproverPolicyV1"] = None, approver: Optional["Approver"] = None, policy_name: Optional[str] = None, + pinned_approver_keys: Optional[Mapping[str, str]] = None, ) -> None: self.mode = resolve_approval_mode(mode) if approval_store is None or isinstance(approval_store, ItemApprovalStore): @@ -324,6 +325,18 @@ def __init__( else: self.log = SwitchLog(log) self._pinned_keys = dict(pinned_status_keys or {}) + # Approver keys a person's signed re-arm may be signed with: the ones + # given here (key id to a raw Ed25519 public key, hex or base64), or + # by default the approvers on the signed approver roster that the + # run's pinned root vouches for (#1486). That default is deliberate: + # the roster is signed down from a root a person pins, and lists only + # people. With none, a switch that turned off stays off. + if pinned_approver_keys is not None: + self._approver_keys = dict(pinned_approver_keys) + else: + from openadapt_flow.approval_mode.switch import registry_approver_keys + + self._approver_keys = registry_approver_keys() self._fetch_status = fetch_status self._environment_digest = environment_digest self._runner_id = runner_id @@ -592,7 +605,7 @@ def _switch_reads_off(self) -> bool: if self.record is None or self.log is None: return True try: - return bool(self.log.state().is_off) + return bool(self.log.state(approver_keys=self._approver_keys).is_off) except (SwitchLogError, OSError): return True @@ -991,7 +1004,7 @@ def fail( if record is None or self.log is None: return fail("no_switch_record") try: - state = self.log.state() + state = self.log.state(approver_keys=self._approver_keys) except (SwitchLogError, OSError): return fail("switch_log_unreadable") if ( @@ -1117,7 +1130,7 @@ def _log_switch_off(self, reason: OffReason) -> None: if self._off_logged is not None: return try: - state = self.log.state() + state = self.log.state(approver_keys=self._approver_keys) if not state.is_off: self.log.append("switched_off", reason=reason.value) self._off_logged = reason @@ -1532,9 +1545,10 @@ def _unapproved_delivered_saves( continue step = steps.get(result.step_id) if step is not None and workflow is not None: - from openadapt_flow.run_gate import is_consequential - - consequential = is_consequential(step, workflow) + # The same classifier as the gate's own saves, with the + # certifying policy: a step the certified qualification + # classified as not saving is not an unapproved save. + consequential = self._is_consequential(step, workflow) else: from openadapt_flow.approval_mode.post import _consequential @@ -1625,12 +1639,14 @@ def _finish_problems_only_with_person( ), ) if reason == "save_not_approved": + # A save was delivered that no person approved: a program + # problem, so the switch turns off. decision = ItemDecision( kind=ItemDecisionKind.PROBLEM, reason=reason, needs_person=True, counts_toward_problems=True, - switch_off=decision.switch_off, + switch_off=decision.switch_off or OffReason.UNCERTAIN_RESULT, ) recorded, off = self._log_finish(decision) evidence = ApprovalModeItemEvidence( @@ -1673,7 +1689,7 @@ def _log_finish(self, decision: ItemDecision) -> tuple[bool, Optional[OffReason] # Only a counted problem can tip the problem rate. return True, off try: - state = self.log.state() + state = self.log.state(approver_keys=self._approver_keys) if not state.is_off and too_many_problems( state.finished, self.record.thresholds ): diff --git a/packages/flow/openadapt_flow/approval_mode/post.py b/packages/flow/openadapt_flow/approval_mode/post.py index 44510ebc8..ea7601565 100644 --- a/packages/flow/openadapt_flow/approval_mode/post.py +++ b/packages/flow/openadapt_flow/approval_mode/post.py @@ -18,6 +18,7 @@ from __future__ import annotations +import dataclasses from collections import Counter from collections.abc import Sequence from dataclasses import dataclass @@ -257,6 +258,131 @@ def engine_refusal_decision( return None +#: A problem with the item's own inputs: the program and the app both worked, +#: and this item needs a person. It leaves the switch on. +INPUT_PROBLEM = "input" +#: A problem with the program or its environment. It turns the switch off. +PROGRAM_PROBLEM = "program" +#: Item checks that judge the item's inputs, not the program. +_INPUT_CHECK_CODES = frozenset({"inputs_supplied", "inputs_strict", "channel"}) +#: ``StepResult.duplicate_check`` values (set by the duplicate check): a real +#: candidate was found, or the list or record channel couldn't be read. +#: The record channel's own decisions name an exact or close match. +_DUPLICATE_FOUND = frozenset({"candidate", "exact", "close"}) +_DUPLICATE_UNREADABLE = "unreadable" + + +def problem_class(report: "RunReport", reason: Optional[str] = None) -> str: + """``input`` or ``program`` for a finished item that needs a person. + + An input problem is one the program caught in the item's own values + before anything was written: + + - a value refused before the first step: a missing required value, or a + typed value that isn't valid (a date that isn't an ISO date); + - an item check on the inputs (``inputs_supplied``, ``inputs_strict``); + - a real duplicate candidate in the app's possible-duplicate list. + + Everything else is a program or environment problem: a wrong write the + record check caught, a failed identity check, a postcondition that failed + after delivery, a changed screen, and any cause not named here. An + unknown cause is a program problem (fail closed), and so is any outcome + that doesn't read "nothing was written" consistently. + """ + + if report.transaction_outcome not in NOTHING_WRITTEN or outcome_is_uncertain( + report + ): + return PROGRAM_PROBLEM + if reason in _INPUT_CHECK_CODES: + return INPUT_PROBLEM + halted = next( + (r for r in reversed(report.results) if not r.ok and not r.skipped), None + ) + if halted is None: + return PROGRAM_PROBLEM + if halted.step_id == "": + return INPUT_PROBLEM + # The duplicate check says why it failed in a structured field; the error + # text is never read. A check that couldn't read its list, and any + # duplicate stop that doesn't say, is a program problem. + if ( + report.transaction_outcome == "HALTED_BEFORE_EFFECT" + and getattr(halted, "duplicate_check", None) in _DUPLICATE_FOUND + and not any( + r.delivery_uncertainty is not None + or (r.delivery_attempted is True and _consequential(r)) + for r in report.results + ) + ): + return INPUT_PROBLEM + return PROGRAM_PROBLEM + + +#: Gate pre-save checks, by the closed off reason their failure means. +_GATE_OFF_REASONS = { + "signed_in_user": OffReason.ENVIRONMENT_CHANGED, + "right_record": OffReason.SCREEN_CHANGED, + "same_screens": OffReason.SCREEN_CHANGED, + "record_check_ready": OffReason.RECORD_CHECK_LOST, +} + + +def _program_off_reason(report: "RunReport", reason: Optional[str] = None) -> OffReason: + """The closed off reason for a program problem that named none.""" + + if reason in _GATE_OFF_REASONS: + return _GATE_OFF_REASONS[reason] + if any( + getattr(result, "duplicate_check", None) == _DUPLICATE_UNREADABLE + for result in report.results + ): + return OffReason.RECORD_CHECK_LOST + + for result in report.results: + evidence = result.safety_refusal_evidence + if evidence is not None and evidence.stage == "identity_verification": + return OffReason.SCREEN_CHANGED + if report.transaction_outcome == "HALTED_BEFORE_EFFECT" and not ( + outcome_is_uncertain(report) + ): + # The program stopped before any effect: the app no longer looks the + # way the program was admitted for. + return OffReason.SCREEN_CHANGED + return OffReason.UNCERTAIN_RESULT + + +def classify_decision(report: "RunReport", decision: ItemDecision) -> ItemDecision: + """Apply the switch policy to a decision that counts as a problem. + + An input problem goes to a person and leaves the switch on: it no longer + counts toward ``too_many_problems``. A program problem turns the switch + off, with the decision's own reason or a closed one for its cause. A + decision that isn't a problem (a person is waiting to approve, the item + checks sent it to a person) is unchanged. + """ + + if not decision.needs_person or not ( + decision.counts_toward_problems or decision.switch_off is not None + ): + return decision + if decision.switch_off is not None: + # The gate, R2 or the engine already turned the switch off for a + # named reason. Nothing classes that down to an input problem. + return decision + if problem_class(report, decision.reason) == INPUT_PROBLEM: + return dataclasses.replace( + decision, counts_toward_problems=False, switch_off=None + ) + if decision.switch_off is None: + return dataclasses.replace( + decision, + counts_toward_problems=True, + switch_off=_program_off_reason(report, decision.reason), + ) + return decision + + def decide_problems_only( report: "RunReport", *, @@ -264,6 +390,29 @@ def decide_problems_only( r2: CheckOutcome, gate_decision: Optional[ItemDecision], save_delivered: bool = False, +) -> ItemDecision: + """The item's decision under ``problems_only``, with the switch policy + applied (:func:`classify_decision`).""" + + return classify_decision( + report, + _decide_problems_only( + report, + r1=r1, + r2=r2, + gate_decision=gate_decision, + save_delivered=save_delivered, + ), + ) + + +def _decide_problems_only( + report: "RunReport", + *, + r1: CheckOutcome, + r2: CheckOutcome, + gate_decision: Optional[ItemDecision], + save_delivered: bool = False, ) -> ItemDecision: """Combine the gate's pre-save decision with R1, R2 and the outcome. diff --git a/packages/flow/openadapt_flow/approval_mode/switch.py b/packages/flow/openadapt_flow/approval_mode/switch.py index c48ad8186..6a863ad1a 100644 --- a/packages/flow/openadapt_flow/approval_mode/switch.py +++ b/packages/flow/openadapt_flow/approval_mode/switch.py @@ -32,7 +32,15 @@ from pathlib import Path from typing import Any, Literal, Optional -from pydantic import BaseModel, ConfigDict, Field, ValidationError, model_validator +from pydantic import ( + BaseModel, + ConfigDict, + Field, + ValidationError, + field_validator, + model_serializer, + model_validator, +) from openadapt_flow.approval_mode.model import ( CODE_PATTERN, @@ -58,6 +66,12 @@ _STATUS_DOMAIN = b"openadapt.approval-mode-status/v1\n" _RECORD_DOMAIN = b"openadapt.approval-mode-switch/v1\n" _LOG_DOMAIN = b"openadapt.approval-mode-log/v1\n" +_REARM_DOMAIN = b"openadapt.approval-mode-rearm/v1\n" +#: Consecutive verified items since the switch went off before a person may +#: re-arm it. Never automatic. +REARM_CLEAN_ITEMS = 10 +#: Item decisions that count as a verified item toward a re-arm. +_CLEAN_DECISIONS = frozenset({"no_person", "approved"}) GENESIS_SHA256 = "sha256:" + "0" * 64 #: Resolution rungs from strongest to weakest (``ir.Rung``). @@ -297,9 +311,41 @@ def verify_status( "switched_off", "status_unconfirmed", "status_restored", + "rearmed", ] +class RearmEvidence(BaseModel): + """A person's signed re-arm of a switched-off switch (one tap). + + Signed with the person's approver key over everything they were shown: + the switch, the off event and its reason, the recorded cause, and the run + refs of the consecutive verified items since the switch went off. + """ + + model_config = ConfigDict(extra="forbid", frozen=True) + + off_seq: int = Field(ge=1) + off_sha256: str = Field(pattern=PREFIXED_SHA256_PATTERN) + off_reason: str = Field(pattern=CODE_PATTERN) + #: What the person was told caused the switch-off, in a sentence. + cause: str = Field(min_length=1, max_length=500) + clean_run_refs: tuple[str, ...] = Field(min_length=REARM_CLEAN_ITEMS) + approver_id: str = Field(pattern=OPAQUE_ID_PATTERN) + key_id: str = Field(min_length=1, max_length=128) + signed_at: str = Field(min_length=1, max_length=64) + signature: str = Field(min_length=1, max_length=256) + + @field_validator("clean_run_refs") + @classmethod + def _refs(cls, value: tuple[str, ...]) -> tuple[str, ...]: + import re as _re + + if any(not _re.fullmatch(HEX_SHA256_PATTERN, ref) for ref in value): + raise ValueError("a clean run ref is a hex SHA-256") + return value + + class SwitchLogEvent(BaseModel): """One hash-chained event in the runner's local switch log.""" @@ -319,9 +365,21 @@ class SwitchLogEvent(BaseModel): decision: Optional[ItemDecisionKind] = None reason: Optional[str] = Field(default=None, pattern=CODE_PATTERN) counts_toward_problems: Optional[bool] = None + #: ``rearmed`` only: the person's signed re-arm. + rearm: Optional[RearmEvidence] = None prev_sha256: str = Field(pattern=PREFIXED_SHA256_PATTERN) sha256: str = Field(pattern=PREFIXED_SHA256_PATTERN) + @model_serializer(mode="wrap") + def _serialize_compatible(self, handler: Any) -> dict[str, Any]: + """Leave ``rearm`` out when unset, so every other event (and every + existing log line and hash) serializes exactly as before.""" + + data: dict[str, Any] = handler(self) + if self.rearm is None: + data.pop("rearm", None) + return data + @model_validator(mode="after") def _shape_per_kind(self) -> "SwitchLogEvent": kind = self.kind @@ -338,6 +396,8 @@ def _shape_per_kind(self) -> "SwitchLogEvent": if kind == "switched_off": if self.reason is None or self.reason not in {r.value for r in OffReason}: raise ValueError("a 'switched_off' event names a closed reason") + if (kind == "rearmed") != (self.rearm is not None): + raise ValueError("only a 'rearmed' event carries a signed re-arm") return self @@ -368,8 +428,18 @@ def is_off(self) -> bool: return self.off_reason is not None -def summarize_events(events: list[SwitchLogEvent]) -> SwitchLogState: - """Fold a verified event list into its state.""" +def summarize_events( + events: list[SwitchLogEvent], + *, + approver_keys: Optional[Mapping[str, str]] = None, +) -> SwitchLogState: + """Fold a verified event list into its state. + + A ``rearmed`` event turns the switch back on only when ``approver_keys`` + (key id to base64 raw Ed25519 public key, pinned for this runner) verify + it and it meets the re-arm rule against the log before it + (:func:`rearm_refusal`). Otherwise it is ignored and the switch stays off. + """ if not events or events[0].kind != "opened": raise SwitchLogError("the switch log does not start with 'opened'") @@ -383,12 +453,18 @@ def summarize_events(events: list[SwitchLogEvent]) -> SwitchLogState: ) started: dict[str, bool] = {} latest_finished: dict[str, SwitchLogEvent] = {} - for event in events[1:]: + for index, event in enumerate(events[1:], start=1): if event.kind == "opened": raise SwitchLogError("the switch log was opened twice") if event.kind == "switched_off" and state.off_reason is None: assert event.reason is not None state.off_reason = OffReason(event.reason) + elif event.kind == "rearmed" and state.off_reason is not None: + assert event.rearm is not None + if approver_keys and ( + rearm_refusal(events[:index], event.rearm, approver_keys) is None + ): + state.off_reason = None elif event.kind == "status_unconfirmed": state.status_unconfirmed = True elif event.kind == "status_restored": @@ -492,8 +568,10 @@ def read(self) -> list[SwitchLogEvent]: raise SwitchLogError("the switch log is empty") return events - def state(self) -> SwitchLogState: - return summarize_events(self.read()) + def state( + self, *, approver_keys: Optional[Mapping[str, str]] = None + ) -> SwitchLogState: + return summarize_events(self.read(), approver_keys=approver_keys) # -- appending ----------------------------------------------------------- def append( @@ -505,6 +583,7 @@ def append( decision: Optional[ItemDecisionKind] = None, reason: Optional[str] = None, counts_toward_problems: Optional[bool] = None, + rearm: Optional[RearmEvidence] = None, ) -> SwitchLogEvent: if kind == "opened": raise SwitchLogError("use SwitchLog.open_new to open a log") @@ -524,6 +603,8 @@ def append( "counts_toward_problems": counts_toward_problems, "prev_sha256": last.sha256, } + if rearm is not None: + fields["rearm"] = rearm.model_dump(mode="json") return self._write_event(fields) def _write_event(self, fields: dict[str, Any]) -> SwitchLogEvent: @@ -596,3 +677,276 @@ def turn_off( """ return log.append("switched_off", reason=reason.value, at=at) + + +# -- re-arming a switched-off switch ------------------------------------------- + + +def _last_off( + events: list[SwitchLogEvent], + approver_keys: Optional[Mapping[str, str]] = None, +) -> Optional[SwitchLogEvent]: + """The ``switched_off`` event that holds the switch off now, if any. + + Only a re-arm that verifies against the log before it (with + ``approver_keys``) ends an off period; a forged or early one doesn't. + """ + + off: Optional[SwitchLogEvent] = None + for index, event in enumerate(events): + if event.kind == "switched_off" and off is None: + off = event + elif ( + event.kind == "rearmed" + and off is not None + and event.rearm is not None + and approver_keys + and rearm_refusal(events[:index], event.rearm, approver_keys) is None + ): + off = None + return off + + +def clean_streak( + events: list[SwitchLogEvent], + approver_keys: Optional[Mapping[str, str]] = None, +) -> tuple[str, ...]: + """Run refs of the consecutive verified items since the switch went off. + + The trailing run of ``item_finished`` events after the off event whose + decision is a verified item (``no_person`` or ``approved``). Any other + finished item resets it; an item still open after the last one does too. + """ + + off = _last_off(events, approver_keys) + if off is None: + return () + streak: list[str] = [] + started: set[str] = set() + for event in events[off.seq :]: + if event.kind == "item_started" and event.run_ref: + started.add(event.run_ref) + elif event.kind == "item_finished" and event.run_ref: + started.discard(event.run_ref) + if event.decision is not None and event.decision.value in _CLEAN_DECISIONS: + streak.append(event.run_ref) + else: + streak = [] + return () if started else tuple(streak) + + +def _public_key_bytes(encoded: str) -> bytes: + """A raw Ed25519 public key, given as hex (the principal registry's form) + or base64 (the pinned-key form status keys use).""" + + import re as _re + + if _re.fullmatch(r"[0-9a-f]{64}", encoded): + return bytes.fromhex(encoded) + return base64.b64decode(encoded, validate=True) + + +def registry_approver_keys(key_dir: Optional[Path] = None) -> dict[str, str]: + """The approver keys a run trusts: the signed approver roster's entries + that the run's pinned root vouches for. + + The roster (``approval_mode.approver_roster``, #1486) lists each person's + approver key, signed down from a root approver; the run pins that root + (the deployment config, ``--approver-root`` or ``OPENADAPT_APPROVER_ROOT``). + Returns ``{name: base64 public key}``, and nothing when no root is pinned, + the roster doesn't verify, or this Flow has no roster yet. Then no re-arm + counts. + """ + + import importlib + + try: + # Imported by name: the roster arrives with #1486. + approver_roster: Any = importlib.import_module( + "openadapt_flow.approval_mode.approver_roster" + ) + from openadapt_flow.jit.lifecycle import default_key_dir + except ImportError: + return {} + try: + keys = approver_roster.approver_keys_b64( + key_dir or default_key_dir(), approver_roster.pinned_root(None) + ) + except Exception: # noqa: BLE001 - an unverifiable roster trusts no one + return {} + return {str(name): str(key) for name, key in dict(keys).items()} + + +def rearm_signing_payload( + switch_id: str, switch_record_sha256: str, fields: Mapping[str, Any] +) -> bytes: + """The bytes a person's approver key signs to re-arm a switch.""" + + unsigned = {k: v for k, v in fields.items() if k != "signature"} + return _REARM_DOMAIN + canonical_json( + { + "switch_id": switch_id, + "switch_record_sha256": switch_record_sha256, + **unsigned, + } + ) + + +def rearm_request( + log: SwitchLog, + *, + cause: str, + approver_keys: Optional[Mapping[str, str]] = None, +) -> dict[str, Any]: + """What a person is shown before re-arming, or raise when it isn't offered. + + Offered only while the switch is off and after + :data:`REARM_CLEAN_ITEMS` consecutive verified items since it went off. + ``approver_keys`` are the keys the gate trusts (by default the registered + approvers, :func:`registry_approver_keys`): a re-arm already in the log + counts only if it verifies with them. Returns the unsigned fields the + person's key signs (without ``approver_id``, ``key_id``, ``signed_at``). + """ + + keys = registry_approver_keys() if approver_keys is None else approver_keys + events = log.read() + off = _last_off(events, keys) + if off is None: + raise SwitchLogError("the switch is on; there is nothing to re-arm") + streak = clean_streak(events, keys) + if len(streak) < REARM_CLEAN_ITEMS: + raise SwitchLogError( + f"{len(streak)} consecutive verified items since the switch went " + f"off; a re-arm needs {REARM_CLEAN_ITEMS}" + ) + return { + "off_seq": off.seq, + "off_sha256": off.sha256, + "off_reason": off.reason, + "cause": cause, + "clean_run_refs": list(streak[-REARM_CLEAN_ITEMS:]), + } + + +def rearm_to_sign( + log: SwitchLog, + request: Mapping[str, Any], + *, + approver_id: str, + key_id: str, + signed_at: Optional[str] = None, +) -> tuple[dict[str, Any], bytes]: + """The fields to sign and their exact bytes, for a key held elsewhere. + + For a key that never leaves a keychain: sign the returned bytes there, + then pass the fields and the base64 signature to :func:`rearm_evidence`. + """ + + first = log.read()[0] + assert first.switch_record_sha256 is not None + fields = { + **dict(request), + "approver_id": approver_id, + "key_id": key_id, + "signed_at": signed_at or _utc_now(), + } + return fields, rearm_signing_payload( + first.switch_id, first.switch_record_sha256, fields + ) + + +def rearm_evidence(fields: Mapping[str, Any], signature_b64: str) -> RearmEvidence: + """The re-arm from :func:`rearm_to_sign`'s fields and a signature made + elsewhere. :func:`rearm` verifies it; nothing is trusted here.""" + + return RearmEvidence.model_validate({**dict(fields), "signature": signature_b64}) + + +def sign_rearm( + log: SwitchLog, + request: Mapping[str, Any], + *, + approver_id: str, + key_id: str, + private_key: Any, + signed_at: Optional[str] = None, +) -> RearmEvidence: + """A person's approver key signs the re-arm they were shown.""" + + fields, payload = rearm_to_sign( + log, request, approver_id=approver_id, key_id=key_id, signed_at=signed_at + ) + signature = private_key.sign(payload) + return rearm_evidence(fields, base64.b64encode(signature).decode("ascii")) + + +def rearm_refusal( + events: list[SwitchLogEvent], + rearm: RearmEvidence, + approver_keys: Mapping[str, str], +) -> Optional[str]: + """Why ``rearm`` may not turn the switch on after ``events``, or None.""" + + from cryptography.exceptions import InvalidSignature + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + + if not events or events[0].switch_record_sha256 is None: + return "the switch log is not readable" + if rearm.approver_id != rearm.key_id: + # The key is registered under the approver's own name; a re-arm that + # names one person and is signed with another's key proves nothing. + return "the re-arm's approver is not the key's registered approver" + off = _last_off(events, approver_keys) + if off is None: + return "the switch is on; there is nothing to re-arm" + if rearm.off_seq != off.seq or rearm.off_sha256 != off.sha256: + return "the re-arm names another switch-off" + if rearm.off_reason != off.reason: + return "the re-arm names another off reason" + streak = clean_streak(events, approver_keys) + if len(streak) < REARM_CLEAN_ITEMS: + return ( + f"{len(streak)} consecutive verified items since the switch went off; " + f"a re-arm needs {REARM_CLEAN_ITEMS}" + ) + if tuple(rearm.clean_run_refs) != tuple(streak[-REARM_CLEAN_ITEMS:]): + return "the re-arm names other verified items than the log" + encoded = approver_keys.get(rearm.key_id) + if not encoded: + return "the re-arm is not signed by a registered approver key" + try: + public_key = Ed25519PublicKey.from_public_bytes(_public_key_bytes(encoded)) + public_key.verify( + base64.b64decode(rearm.signature, validate=True), + rearm_signing_payload( + events[0].switch_id, + events[0].switch_record_sha256, + rearm.model_dump(mode="json"), + ), + ) + except (InvalidSignature, ValueError, TypeError, binascii.Error): + return "the re-arm signature does not verify" + return None + + +def rearm( + log: SwitchLog, + evidence: RearmEvidence, + *, + approver_keys: Mapping[str, str], + at: Optional[str] = None, +) -> SwitchLogEvent: + """Record a person's signed re-arm, or raise :class:`SwitchLogError`. + + Never automatic: it needs a signature by a pinned approver key over the + switch-off, its cause and the :data:`REARM_CLEAN_ITEMS` consecutive + verified items since. The event records who signed and what they saw. + """ + + refusal = rearm_refusal(log.read(), evidence, approver_keys) + if refusal is not None: + raise SwitchLogError(refusal) + # Every reader checks the re-arm again against the log before it, so an + # event that lands between this check and the append can only make the + # re-arm count for nothing, never turn the switch on early. + return log.append("rearmed", rearm=evidence, at=at) diff --git a/packages/flow/tests/test_approval_mode_gate.py b/packages/flow/tests/test_approval_mode_gate.py index 119260781..f34363c76 100644 --- a/packages/flow/tests/test_approval_mode_gate.py +++ b/packages/flow/tests/test_approval_mode_gate.py @@ -818,19 +818,20 @@ def test_a_stop_before_saving_is_finished_as_a_counted_problem(tmp_path): evidence = h.gate.finish_item(_halted_report("HALTED_BEFORE_EFFECT")) assert evidence.decision is ItemDecisionKind.STOP_BEFORE_SAVING assert evidence.counts_toward_problems - assert evidence.switch_off_reason is None + # The wrong user signed in: an environment problem, so the switch turns + # off at once (not after three in a row). + assert evidence.switch_off_reason is OffReason.ENVIRONMENT_CHANGED assert evidence.gated_step_id == "save" -def test_three_counted_problems_in_a_row_switch_the_mode_off(tmp_path): +def test_a_program_problem_switches_the_mode_off_at_once(tmp_path): h = Harness(tmp_path) h.user.verdict = "mismatch" - for index in range(3): - h.begin(run_id=f"run-{index}") - assert not h.before_save().proceed - evidence = h.gate.finish_item(_halted_report("HALTED_BEFORE_EFFECT")) - assert evidence.switch_off_reason is OffReason.TOO_MANY_PROBLEMS - assert h.log.state().off_reason is OffReason.TOO_MANY_PROBLEMS + h.begin(run_id="run-0") + assert not h.before_save().proceed + evidence = h.gate.finish_item(_halted_report("HALTED_BEFORE_EFFECT")) + assert evidence.switch_off_reason is OffReason.ENVIRONMENT_CHANGED + assert h.log.state().off_reason is OffReason.ENVIRONMENT_CHANGED def test_items_sent_to_approval_by_a_stale_status_never_switch_off(tmp_path): @@ -869,8 +870,9 @@ def test_each_item_finish_always_needs_a_person(tmp_path): def test_a_reused_gate_never_carries_one_items_context_to_the_next(tmp_path): h = Harness(tmp_path) h.begin(run_id="run-a") - assert h.before_save().proceed - h.gate.finish_item(_verified_report()) + saved = _gated_save() + assert h.before_save(current=saved).proceed + h.gate.finish_item(_verified_report(saved)) # The next item without a fresh context gets the fail-safe default. h.begin(run_id="run-b", fresh_item=False) verdict = h.before_save() diff --git a/packages/flow/tests/test_approval_mode_post.py b/packages/flow/tests/test_approval_mode_post.py index dd0571fd5..a1bf3c293 100644 --- a/packages/flow/tests/test_approval_mode_post.py +++ b/packages/flow/tests/test_approval_mode_post.py @@ -372,7 +372,9 @@ def test_check_the_record_switches_off_with_uncertain_result(): assert decision.reason == "reconciliation_required" -def test_a_stop_that_wrote_nothing_is_a_counted_problem_without_switch_off(): +def test_a_stop_that_wrote_nothing_for_an_unnamed_cause_switches_off(): + # A halt before any effect that isn't an input problem: the app no longer + # looks as the program was admitted for. A program problem, so off. report = _halted_report("HALTED_BEFORE_EFFECT") decision = decide_problems_only( report, @@ -382,7 +384,7 @@ def test_a_stop_that_wrote_nothing_is_a_counted_problem_without_switch_off(): ) assert decision.kind is ItemDecisionKind.PROBLEM assert decision.counts_toward_problems - assert decision.switch_off is None + assert decision.switch_off is OffReason.SCREEN_CHANGED def test_gate_decision_wins_and_an_uncertain_earlier_save_still_switches_off(): diff --git a/packages/flow/tests/test_approval_mode_switch_policy.py b/packages/flow/tests/test_approval_mode_switch_policy.py new file mode 100644 index 000000000..7884a9953 --- /dev/null +++ b/packages/flow/tests/test_approval_mode_switch_policy.py @@ -0,0 +1,464 @@ +"""Which problems turn the problems_only switch off, and how a person re-arms it. + +Policy (coordinator decision, 2026-10-24): + +- An item-input problem caught before anything was written (a missing or + invalid value refused before the first step, an input item check, a real + candidate in the app's duplicate list) sends that item to a person and + leaves the switch on. +- A program or environment problem (a wrong write the record check caught, a + failed identity check, a changed screen, any unnamed cause) turns it off. + An unknown cause counts as a program problem. +- Re-arming takes one signed tap from a person with a pinned approver key, + offered only after 10 consecutive verified items since the switch went off. + It is never automatic, and the log records who signed and what they saw. +""" + +from __future__ import annotations + +import base64 + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat + +from openadapt_flow.approval_mode.model import ( + CheckOutcome, + ItemDecisionKind, + OffReason, +) +from openadapt_flow.approval_mode.post import ( + INPUT_PROBLEM, + PROGRAM_PROBLEM, + decide_problems_only, + problem_class, +) +from openadapt_flow.approval_mode.switch import ( + REARM_CLEAN_ITEMS, + SwitchLogError, + rearm, + rearm_evidence, + rearm_request, + rearm_to_sign, + sign_rearm, + turn_off, +) +from openadapt_flow.ir import RunReport, StepResult +from tests._approval_mode_support import BUNDLE_DIGEST, make_record, open_log + +FAIL = CheckOutcome(code="verified", passed=False, detail="not_verified") + + +@pytest.fixture(autouse=True) +def _no_registered_approvers(monkeypatch): + """Hermetic: never read the developer's own approver roster.""" + + monkeypatch.delenv("OPENADAPT_APPROVER_ROOT", raising=False) + + +def _with_duplicate_check(result: StepResult, value: str) -> StepResult: + """The duplicate check's structured outcome on a halted step. + + ``StepResult.duplicate_check`` is added and set by the record-channel + duplicate check (#1516); until then it is set here directly. + """ + + result = result.model_copy() + object.__setattr__(result, "duplicate_check", value) + return result + + +PASS_R2 = CheckOutcome(code="whole_run_clean", passed=True) + + +def _report( + transaction: str, halted: StepResult, execution: str = "HALTED" +) -> RunReport: + return RunReport( + workflow_name="referral", + started_at="2026-10-24T10:00:00+00:00", + execution_profile="regulated", + execution_outcome=execution, + execution_completed=False, + transaction_outcome=transaction, + bundle_content_digest=BUNDLE_DIGEST, + results=[halted], + ) + + +def _decide(report: RunReport): + return decide_problems_only(report, r1=FAIL, r2=PASS_R2, gate_decision=None) + + +INVALID_DATE = StepResult( + step_id="", + intent="validate typed workflow parameters", + ok=False, + failure_category="governed_refusal", + error="Typed workflow parameter(s) are not valid: parameter 'DOB' is not an " + "ISO date — refusing to run; nothing was typed; run aborted", +) +DUPLICATE = StepResult( + step_id="open_duplicate_check", + intent="click 'Create New Patient'", + ok=False, + delivery_attempted=True, + error="Postconditions failed for step 'open_duplicate_check': expected screen " + "state not reached (semantic drift) — failed: no_duplicate_candidate (no " + "listed row has this run's lname, DOB) — run aborted", +) +NOT_FOUND = StepResult( + step_id="save", + intent="click 'Save'", + ok=False, + delivery_attempted=False, + error="Target not found for step 'save'", +) +CAUGHT_WRONG_WRITE = StepResult( + step_id="save", + intent="click 'Save'", + ok=False, + risk="irreversible", + delivery_attempted=True, + effect_verified=False, + error="System-of-record effect verification HALTED step 'save'", +) + + +@pytest.mark.parametrize( + "transaction, halted", + [ + ("REJECTED_POLICY", INVALID_DATE), + ("HALTED_BEFORE_EFFECT", INVALID_DATE), + ("HALTED_BEFORE_EFFECT", _with_duplicate_check(DUPLICATE, "candidate")), + ("HALTED_BEFORE_EFFECT", _with_duplicate_check(DUPLICATE, "exact")), + ("HALTED_BEFORE_EFFECT", _with_duplicate_check(DUPLICATE, "close")), + ], +) +def test_an_input_problem_leaves_the_switch_on(transaction, halted) -> None: + report = _report(transaction, halted) + assert problem_class(report) == INPUT_PROBLEM + decision = _decide(report) + assert decision.needs_person + assert not decision.counts_toward_problems + assert decision.switch_off is None + + +def test_a_caught_wrong_write_turns_the_switch_off() -> None: + report = _report("RECONCILIATION_REQUIRED", CAUGHT_WRONG_WRITE) + assert problem_class(report) == PROGRAM_PROBLEM + decision = _decide(report) + assert decision.counts_toward_problems + assert decision.switch_off is OffReason.UNCERTAIN_RESULT + + +def test_an_unnamed_stop_before_any_effect_is_a_program_problem() -> None: + report = _report("HALTED_BEFORE_EFFECT", NOT_FOUND) + assert problem_class(report) == PROGRAM_PROBLEM + assert _decide(report).switch_off is OffReason.SCREEN_CHANGED + + +def test_a_duplicate_stop_after_a_delivered_save_is_a_program_problem() -> None: + # The duplicate-list failure counts as an input problem only when nothing + # that saves was delivered. + report = _report( + "HALTED_BEFORE_EFFECT", _with_duplicate_check(DUPLICATE, "candidate") + ) + report.results.insert(0, CAUGHT_WRONG_WRITE.model_copy(update={"ok": True})) + assert problem_class(report) == PROGRAM_PROBLEM + + +def test_an_inconsistent_outcome_is_never_an_input_problem() -> None: + # "Nothing was written" with an execution outcome that says otherwise. + report = _report("HALTED_BEFORE_EFFECT", INVALID_DATE, execution="VERIFIED") + assert problem_class(report) == PROGRAM_PROBLEM + + +# -- re-arming ------------------------------------------------------------------- + + +def _keypair() -> tuple[Ed25519PrivateKey, dict[str, str]]: + private = Ed25519PrivateKey.generate() + public = private.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw) + return private, {"approver-1": base64.b64encode(public).decode("ascii")} + + +def _finish(log, index: int, decision: ItemDecisionKind) -> None: + ref = f"{index:064x}" + log.append("item_started", run_ref=ref) + log.append( + "item_finished", + run_ref=ref, + decision=decision, + counts_toward_problems=decision is ItemDecisionKind.PROBLEM, + ) + + +def _switched_off_log(tmp_path, clean: int): + log = open_log(tmp_path, make_record()) + turn_off(log, OffReason.UNCERTAIN_RESULT) + for index in range(clean): + decision = ( + ItemDecisionKind.APPROVED if index % 2 else ItemDecisionKind.NO_PERSON + ) + _finish(log, index, decision) + return log + + +def _signed(log, private) -> object: + request = rearm_request( + log, cause="a malformed date of birth was saved wrong", approver_keys={} + ) + return sign_rearm( + log, + request, + approver_id="approver-1", + key_id="approver-1", + private_key=private, + ) + + +def test_a_signed_rearm_after_ten_verified_items_turns_the_switch_on(tmp_path) -> None: + private, keys = _keypair() + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + event = rearm(log, _signed(log, private), approver_keys=keys) + assert event.kind == "rearmed" and event.rearm is not None + assert event.rearm.approver_id == "approver-1" + assert event.rearm.cause.startswith("a malformed date") + assert len(event.rearm.clean_run_refs) == REARM_CLEAN_ITEMS + assert not log.state(approver_keys=keys).is_off + # A reader without the pinned approver keys never honours it. + assert log.state().is_off + + +def test_a_rearm_before_ten_verified_items_is_refused(tmp_path) -> None: + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS - 1) + with pytest.raises(SwitchLogError, match="needs 10"): + rearm_request(log, cause="x") + + +def test_a_problem_resets_the_verified_streak(tmp_path) -> None: + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + _finish(log, 99, ItemDecisionKind.PROBLEM) + with pytest.raises(SwitchLogError, match="0 consecutive"): + rearm_request(log, cause="x") + + +def test_a_rearm_without_a_valid_signature_is_refused(tmp_path) -> None: + private, keys = _keypair() + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + signed = _signed(log, private) + # Signed by a key that isn't pinned. + other, _ = _keypair() + with pytest.raises(SwitchLogError, match="registered approver key"): + rearm(log, _signed(log, other), approver_keys={}) + # Tampered after signing: a different cause than the person saw. + tampered = signed.model_copy(update={"cause": "nothing happened"}) + with pytest.raises(SwitchLogError, match="does not verify"): + rearm(log, tampered, approver_keys=keys) + # Signed by another person's key under this key id. + forged = _signed(log, other) + with pytest.raises(SwitchLogError, match="does not verify"): + rearm(log, forged, approver_keys=keys) + assert log.state(approver_keys=keys).is_off + + +def test_a_rearm_is_never_automatic_and_a_later_problem_turns_it_off_again( + tmp_path, +) -> None: + private, keys = _keypair() + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS + 5) + # Fifteen clean items alone change nothing. + assert log.state(approver_keys=keys).is_off + rearm(log, _signed(log, private), approver_keys=keys) + assert not log.state(approver_keys=keys).is_off + turn_off(log, OffReason.SCREEN_CHANGED) + state = log.state(approver_keys=keys) + assert state.off_reason is OffReason.SCREEN_CHANGED + with pytest.raises(SwitchLogError, match="0 consecutive"): + rearm_request(log, cause="x", approver_keys=keys) + + +def test_an_appended_rearm_event_with_a_bad_signature_is_ignored(tmp_path) -> None: + # Someone with write access to the log appends a forged re-arm directly. + private, keys = _keypair() + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + other, _ = _keypair() + log.append("rearmed", rearm=_signed(log, other)) + assert log.state(approver_keys=keys).is_off + + +def test_a_key_held_in_a_keychain_signs_the_bytes_without_leaving_it(tmp_path) -> None: + # Desktop signs inside its keychain: it gets the bytes, returns a signature. + private, keys = _keypair() + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + request = rearm_request(log, cause="a malformed date of birth was saved wrong") + fields, payload = rearm_to_sign( + log, request, approver_id="approver-1", key_id="approver-1" + ) + signature = base64.b64encode(private.sign(payload)).decode("ascii") + rearm(log, rearm_evidence(fields, signature), approver_keys=keys) + assert not log.state(approver_keys=keys).is_off + + +def _fake_roster(monkeypatch, keys_or_error): + """Stand in for #1486's approver roster module (not on main yet).""" + + import sys + from types import ModuleType + + roster = ModuleType("openadapt_flow.approval_mode.approver_roster") + calls: list = [] + + def pinned_root(explicit): + return "root-key" + + def approver_keys_b64(key_dir, root): + calls.append((key_dir, root)) + if isinstance(keys_or_error, Exception): + raise keys_or_error + return keys_or_error + + roster.pinned_root = pinned_root + roster.approver_keys_b64 = approver_keys_b64 + monkeypatch.setitem(sys.modules, roster.__name__, roster) + import openadapt_flow.approval_mode as package + + monkeypatch.setattr(package, "approver_roster", roster, raising=False) + return calls + + +def test_roster_approver_keys_verify_a_rearm(tmp_path, monkeypatch) -> None: + """The pinned roster's approvers are the gate's default keys.""" + + from openadapt_flow.approval_mode import switch + + private, keys = _keypair() + calls = _fake_roster(monkeypatch, keys) + trusted = switch.registry_approver_keys(tmp_path) + assert trusted == keys and calls == [(tmp_path, "root-key")] + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + rearm(log, _signed(log, private), approver_keys=trusted) + assert not log.state(approver_keys=trusted).is_off + + +def test_hex_keys_still_verify_a_rearm(tmp_path) -> None: + private, keys = _keypair() + hex_keys = {k: base64.b64decode(v).hex() for k, v in keys.items()} + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + rearm(log, _signed(log, private), approver_keys=hex_keys) + assert not log.state(approver_keys=hex_keys).is_off + + +def test_an_unverifiable_roster_gives_no_approver_keys(tmp_path, monkeypatch) -> None: + from openadapt_flow.approval_mode import switch + + _fake_roster(monkeypatch, ValueError("the roster does not verify")) + assert switch.registry_approver_keys(tmp_path) == {} + + +def test_a_duplicate_stop_is_classed_only_by_its_structured_outcome() -> None: + # The error text alone says nothing: a duplicate stop that doesn't say + # why is a program problem. + plain = _report("HALTED_BEFORE_EFFECT", DUPLICATE) + assert problem_class(plain) == PROGRAM_PROBLEM + # A list or record channel that couldn't be read turns the switch off. + unreadable = _report( + "HALTED_BEFORE_EFFECT", _with_duplicate_check(DUPLICATE, "unreadable") + ) + assert problem_class(unreadable) == PROGRAM_PROBLEM + assert _decide(unreadable).switch_off is OffReason.RECORD_CHECK_LOST + + +def test_an_explicit_switch_off_is_never_classed_down() -> None: + from openadapt_flow.approval_mode.post import ItemDecision + + gate = ItemDecision( + kind=ItemDecisionKind.STOP_BEFORE_SAVING, + reason="record_check_ready", + needs_person=True, + counts_toward_problems=True, + switch_off=OffReason.RECORD_CHECK_LOST, + ) + decision = decide_problems_only( + _report("REJECTED_POLICY", INVALID_DATE), + r1=FAIL, + r2=PASS_R2, + gate_decision=gate, + ) + assert decision.switch_off is OffReason.RECORD_CHECK_LOST + assert decision.counts_toward_problems + + +def test_a_replayed_old_rearm_after_a_second_switch_off_is_refused(tmp_path) -> None: + private, keys = _keypair() + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + first = _signed(log, private) + rearm(log, first, approver_keys=keys) + turn_off(log, OffReason.SCREEN_CHANGED) + for index in range(100, 100 + REARM_CLEAN_ITEMS): + _finish(log, index, ItemDecisionKind.NO_PERSON) + with pytest.raises(SwitchLogError, match="another switch-off"): + rearm(log, first, approver_keys=keys) + assert log.state(approver_keys=keys).is_off + + +def test_a_rearm_for_another_switch_log_is_refused(tmp_path) -> None: + private, keys = _keypair() + log = _switched_off_log(tmp_path / "a", REARM_CLEAN_ITEMS) + other = open_log(tmp_path / "b", make_record(switch_id="switch-other-0001")) + turn_off(other, OffReason.UNCERTAIN_RESULT) + for index in range(REARM_CLEAN_ITEMS): + _finish(other, index, ItemDecisionKind.NO_PERSON) + evidence = _signed(log, private) + with pytest.raises(SwitchLogError): + rearm(other, evidence, approver_keys=keys) + assert other.state(approver_keys=keys).is_off + + +def test_a_forged_rearm_then_a_real_one(tmp_path) -> None: + private, keys = _keypair() + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + forger, _ = _keypair() + log.append("rearmed", rearm=_signed(log, forger)) + # The forged event ends no off period: the same switch-off and streak + # are still offered, and a real re-arm of them counts. + request = rearm_request(log, cause="x", approver_keys=keys) + assert len(request["clean_run_refs"]) == REARM_CLEAN_ITEMS + rearm(log, _signed(log, private), approver_keys=keys) + assert not log.state(approver_keys=keys).is_off + + +def test_a_rearm_naming_another_approver_than_its_key_is_refused(tmp_path) -> None: + private, keys = _keypair() + log = _switched_off_log(tmp_path, REARM_CLEAN_ITEMS) + request = rearm_request(log, cause="x", approver_keys=keys) + mismatched = sign_rearm( + log, + request, + approver_id="someone-else", + key_id="approver-1", + private_key=private, + ) + with pytest.raises(SwitchLogError, match="registered approver"): + rearm(log, mismatched, approver_keys=keys) + + +def test_the_gate_reads_on_after_a_valid_rearm(tmp_path) -> None: + from openadapt_flow.approval_mode.gate import ApprovalGate + + private, keys = _keypair() + record = make_record() + log = open_log(tmp_path, record) + turn_off(log, OffReason.UNCERTAIN_RESULT) + for index in range(REARM_CLEAN_ITEMS): + _finish(log, index, ItemDecisionKind.APPROVED) + trusting = ApprovalGate( + "problems_only", record=record, log=log, pinned_approver_keys=keys + ) + untrusting = ApprovalGate("problems_only", record=record, log=log) + assert trusting._switch_reads_off() and untrusting._switch_reads_off() + rearm(log, _signed(log, private), approver_keys=keys) + assert not trusting._switch_reads_off() + # A gate that doesn't trust the key still reads off. + assert untrusting._switch_reads_off()