fix(flow): a null matches only a null in rollback and qualification read-backs

packages/flow/openadapt_flow/qualify/campaign.py

@@ -560,6 +560,21 @@ def _records_reader(
     return None
 
 
+def _same_field(record: Mapping[str, Any], key: str, want: Any) -> bool:
+    """Whether ``record[key]`` equals ``want``, by text.
+
+    A null equals only a field that is there and null: ``str(None)`` is
+    ``"None"``, which would match the text ``"None"`` or a missing field.
+    """
+
+    if key not in record:
+        return False
+    have = record[key]
+    if have is None or want is None:
+        return have is None and want is None
+    return str(have) == str(want)
+
+
 def _canon(record: Mapping[str, Any]) -> str:
     return json.dumps(record, sort_keys=True, default=str)
 
@@ -623,7 +638,7 @@ def judge(self, oracle_kind: str, report: RunReport) -> dict[str, Any]:
         }
 
         def matches(rec: Mapping[str, Any], sel: Mapping[str, Any]) -> bool:
-            return all(str(rec.get(k)) == str(v) for k, v in sel.items())
+            return all(_same_field(rec, k, v) for k, v in sel.items())
 
         matched = [rec for rec in after if selector and matches(rec, selector)]
         after_set = {_canon(rec) for rec in after}
@@ -633,7 +648,7 @@ def matches(rec: Mapping[str, Any], sel: Mapping[str, Any]) -> bool:
             if effect.kind.value != "field_equals" or effect.field is None:
                 continue
             want = effect.value.literal if effect.value is not None else None
-            ok = len(matched) == 1 and str(matched[0].get(effect.field)) == str(want)
+            ok = len(matched) == 1 and _same_field(matched[0], effect.field, want)
             field_checks.append({"field": effect.field, "match": ok})
         expected = written.expected_count if written else 1
         exact = (

packages/flow/openadapt_flow/rollback.py

@@ -194,23 +194,52 @@ def fresh_presence(
             (s for s in iter_workflow_steps(workflow) if s.id == entry.step_id), None
         )
         effects = list(step.effects) if step is not None else []
+        # A null value counts only when declared or passed as None. An unset
+        # parameter has no value, and a supplied value its map doesn't name
+        # resolves to nothing: either way the step's new value can't be told.
         new_values = {
             e.field: e.value.resolve(params)
             for e in effects
-            if e.field is not None and e.value is not None
+            if e.field is not None
+            and e.value is not None
+            and (
+                e.value.resolve(params) is not None
+                or e.value.null
+                or (
+                    e.value.param is not None
+                    and e.value.param in params
+                    and params[e.value.param] is None
+                )
+            )
         }
         row = rows[0]
         fields = list(undo.restore.fields)
-        if all(str(row.get(f)) == str(values.get(f)) for f in fields):
+        if all(
+            f in row and f in values and _same_text(row[f], values[f]) for f in fields
+        ):
             return "absent", "the record still holds its old values"
         if fields and all(
-            f in new_values and str(row.get(f)) == str(new_values[f]) for f in fields
+            f in row and f in new_values and _same_text(row[f], new_values[f])
+            for f in fields
         ):
             return "present", "the record holds the run's new values"
         return "unknown", "the record holds neither the old nor the new values"
     return _judge_declared(entry, workflow, records, params)
 
 
+def _same_text(stored: Any, want: Any) -> bool:
+    """Whether a stored field equals a value, by text.
+
+    A null equals only a null: ``str(None)`` is ``"None"``, which would let a
+    stored null match the text ``"None"``, and a missing field or old value
+    match a null.
+    """
+
+    if stored is None or want is None:
+        return stored is None and want is None
+    return str(stored) == str(want)
+
+
 def _judge_declared(
     entry: ChangeLedgerEntry,
     workflow: Workflow,

packages/flow/openadapt_flow/runtime/effects/_common.py

@@ -33,8 +33,9 @@
 #: driver returns a number only for a numeric column). There a plain decimal
 #: expected as text compares with a stored number by value
 #: (``record_matches(text_numbers=True)``). A JSON body compares numbers by
-#: value only against a parameter passed as a number.
-TYPED_SUBSTRATES = frozenset({"sql"})
+#: value only against a parameter passed as a number. ``db`` is the reward
+#: worker's name for its SQL channel (a read through ``SqlRecordVerifier``).
+TYPED_SUBSTRATES = frozenset({"sql", "db"})
 
 
 def drop_record_fields(

packages/flow/openadapt_flow/runtime/effects/onscreen.py

@@ -940,9 +940,12 @@ def verify_current_state(
     @staticmethod
     def _expected_text(expected: Effect, params: dict) -> Optional[str]:
         if expected.value is not None:
+            # A declared value with no text (a null or unset parameter, or
+            # "") has nothing to read on screen. Don't fall through to a
+            # selector value, which would confirm the record is shown, not
+            # that the field holds the value.
             resolved = expected.value.resolve(params)
-            if resolved:
-                return resolved
+            return resolved or None
         for v in expected.match.values():
             resolved = v.resolve(params)
             if resolved:

packages/flow/tests/test_null_text_matching.py

@@ -0,0 +1,143 @@
+"""A null never equals the text "None" in the rollback and qualification
+read-backs, and the reward worker's SQL channel compares numbers by value."""
+
+from __future__ import annotations
+
+from openadapt_flow.qualify.campaign import _same_field
+from openadapt_flow.rollback import _same_text
+from openadapt_flow.runtime.effects._common import TYPED_SUBSTRATES, judge_records
+from openadapt_flow.runtime.effects.effect import (
+    Effect,
+    EffectKind,
+    EffectState,
+    Verdict,
+)
+
+
+def test_rollback_null_equals_only_null() -> None:
+    assert _same_text(None, None)
+    assert not _same_text(None, "None")
+    assert not _same_text("None", None)
+    assert not _same_text("", None)
+    assert _same_text(3, "3")
+    assert not _same_text("2.0", "2")
+
+
+def test_qualification_null_needs_a_present_null_field() -> None:
+    assert _same_field({"note": None}, "note", None)
+    assert not _same_field({}, "note", None)
+    assert not _same_field({"note": "None"}, "note", None)
+    assert not _same_field({"note": None}, "note", "None")
+    assert _same_field({"id": 7}, "id", "7")
+
+
+def test_reward_db_channel_is_typed() -> None:
+    assert "db" in TYPED_SUBSTRATES
+    effect = Effect(
+        kind=EffectKind.FIELD_EQUALS,
+        field="qty",
+        value="2.0",
+        match={"id": "1"},
+    )
+    after = [{"id": 1, "qty": 2}]
+    for substrate in ("sql", "db"):
+        before = EffectState(substrate=substrate, reachable=True, records=after)
+        verdict = judge_records(effect, before, after, substrate=substrate)
+        assert verdict.verdict is Verdict.CONFIRMED, substrate
+
+
+def _restore_presence(tmp_path, value, params, fresh_note):
+    """fresh_presence for a field edit of ``note`` whose old value was "old"."""
+    from openadapt_flow.after_stop import keep_prior_state
+    from openadapt_flow.ir import (
+        ActionKind,
+        ChangeLedgerEntry,
+        Step,
+        UndoRestore,
+        UndoStep,
+        Workflow,
+    )
+    from openadapt_flow.rollback import fresh_presence
+
+    edit = Step(
+        id="edit_note",
+        intent="edit the note",
+        action=ActionKind.KEY,
+        key="Enter",
+        risk="irreversible",
+        effects=[
+            Effect(
+                kind=EffectKind.FIELD_EQUALS,
+                match={"id": "r1"},
+                field="note",
+                value=value,
+                risk="irreversible",
+            )
+        ],
+    )
+    undo = UndoStep(
+        id="restore_note",
+        undoes="edit_note",
+        intent="put the old note back",
+        source="hand_written",
+        kind="restore",
+        restore=UndoRestore(match={"id": "r1"}, fields={"note": "old_note"}),
+        steps=[
+            Step(
+                id="put_back",
+                intent="type the old note",
+                action=ActionKind.KEY,
+                key="Enter",
+                effects=[
+                    Effect(
+                        kind=EffectKind.FIELD_EQUALS,
+                        match={"id": "r1"},
+                        field="note",
+                        value={"param": "old_note"},
+                    )
+                ],
+            )
+        ],
+    )
+    workflow = Workflow(name="w", steps=[edit], undo_steps=[undo])
+    kept = keep_prior_state(
+        workflow, edit, [{"id": "r1", "note": "old"}], params, tmp_path
+    )
+    assert kept is not None
+    entry = ChangeLedgerEntry(
+        position=1,
+        step_id="edit_note",
+        intent="edit the note",
+        attempt_state="delivery_uncertain",
+        status="unknown",
+        undo_id="restore_note",
+    )
+    return fresh_presence(
+        entry,
+        undo,
+        workflow,
+        run_dir=tmp_path,
+        params=params,
+        read_records=lambda: [{"id": "r1", "note": fresh_note}],
+    )
+
+
+def test_a_value_its_map_does_not_name_is_unknown_not_a_null(tmp_path) -> None:
+    mapped = {"param": "new", "map": {"known": "mapped"}}
+    # The supplied value is outside the map: it resolves to nothing, which
+    # says nothing about whether the write landed.
+    assert _restore_presence(tmp_path / "a", mapped, {"new": "unmapped"}, None)[0] == (
+        "unknown"
+    )
+    # Controls: a parameter passed as None is an explicit null ...
+    assert _restore_presence(tmp_path / "b", {"param": "new"}, {"new": None}, None)[
+        0
+    ] == ("present")
+    # ... an unset one has no value ...
+    assert _restore_presence(tmp_path / "c", {"param": "new"}, {}, None)[0] == (
+        "unknown"
+    )
+    # ... and a mapped value reads back as its code.
+    assert _restore_presence(tmp_path / "d", mapped, {"new": "known"}, "mapped")[0] == (
+        "present"
+    )

packages/flow/tests/test_onscreen_verifier.py

@@ -310,3 +310,21 @@ def test_single_glyph_difference_is_refuted_same_surface() -> None:
     v = _verifier([_Line("Saved note for patient 8")], min_ratio=0.9)
     verdict = v.read_back("Saved note for patient 3")
     assert verdict.verdict is Verdict.REFUTED
+
+
+def test_null_value_does_not_fall_back_to_the_selector() -> None:
+    """A field_equals whose value has no text reads nothing on screen.
+
+    The record's id is on screen, but that doesn't show the field holds the
+    value, so the check can't confirm.
+    """
+    v = _verifier([_Line("REC-7")])
+    effect = Effect(
+        kind=EffectKind.FIELD_EQUALS,
+        field="note",
+        value={"param": "note_text"},
+        match={"id": "REC-7"},
+    )
+    for params in ({"note_text": None}, {}):
+        verdict = v.verify(effect, v.capture_pre_state(), context=params)
+        assert verdict.verdict is Verdict.INDETERMINATE