problems_only switch policy: input problems leave it on, a person re-arms with one signed tap

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`

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
             ):

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,13 +258,161 @@ 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 == "<params>":
+        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",
     *,
     r1: CheckOutcome,
     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.