Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion packages/flow/docs/EFFECT_KIT.md
Original file line number Diff line number Diff line change
Expand Up @@ -362,7 +362,9 @@ The command looks for a read that returns a record holding them:
a recorded value (bound as `{param: NAME}`, so each run is checked against
its own values) and on fields the save sent with a fixed value under the
same name (`type: Triage`). `forbid_collateral_loss` stops the run if a
record that was there before is gone.
record that was there before is gone. `forbid_collateral_change` (off by
default in a run, on in `verify-run`) also stops it if such a record is
still there but changed.
- Recorded values in the read's URL become placeholders (`path_params`,
FHIR `search_param_exprs`), so the read is scoped to the run's record. A
path segment that looks like one record's id (an MRN, a number, a UUID)
Expand Down
24 changes: 23 additions & 1 deletion packages/flow/docs/VERIFY_RUN.md
Original file line number Diff line number Diff line change
Expand Up @@ -78,14 +78,36 @@ Optional flags:
Both link ids must be opaque (8 to 128 of `A-Za-z0-9._:-`). They are absent
unless given, and when present they are signed with the rest of the receipt.

## Records outside the target

With a baseline, a `record_written` check also refutes when a record that
was there before the agent started and is not the target has changed
(`forbid_collateral_change`), not only when it is gone
(`forbid_collateral_loss`). The agent's write can be right while it also
edited another patient's record in place, and only the baseline shows that.
verify-run turns this on for every `record_written` effect when it judges.
The contract in the file is unchanged, so its digest is still the one qualify
reports, and `evidence.readback.effects[].collateral_change_checked` says the
check ran. To skip it for one effect, for example when the read covers a
table other people edit during the run, set `forbid_collateral_change: false`
on that effect in `effects:`. Fields that change on their own, such as a
modification time, belong in the verifier's `records_drop_fields`.

A changed record is not proof that the executor changed it: another user may
have edited it during the run. The verdict says so, and the outcome is
`reconciliation_required`, so a person checks the record before anything is
retried or undone. Values compare by type. A `Decimal` or `datetime` read
from a database equals the JSON form the baseline stored it in, while two
strings compare exactly, so `"02139"` changed to `"2139"` is a change.

## Verdicts

| What the store shows | `effect_verdict` | `outcome` |
|---|---|---|
| The declared record, exactly once, through a system-of-record read | `CONFIRMED` | `verified` |
| The same, but the executor approved its own write | `CONFIRMED` | `rejected_policy` |
| Nothing that matches | `REFUTED` | `halted_before_effect` |
| A duplicate, a wrong value, or only some of the effects | `REFUTED` | `reconciliation_required` |
| A duplicate, a wrong value, only some of the effects, or another record that vanished or changed | `REFUTED` | `reconciliation_required` |
| The store could not be read, no baseline, or a separate-session read | `INDETERMINATE` or `CONFIRMED` | `reconciliation_required` |

Only a read through the system of record produces `verified`. A screen
Expand Down
10 changes: 10 additions & 0 deletions packages/flow/openadapt_flow/bundle_validation.py
Original file line number Diff line number Diff line change
Expand Up @@ -277,6 +277,16 @@ class _SealedCanonicalOmission:
"the whole over-write contract and remains sealed content"
),
),
_SealedCanonicalOmission(
owner=Effect,
field_name="forbid_collateral_change",
omitted_values=(False,),
reason=(
"an effect that does not check unchanged non-target records is "
"what every earlier bundle held; a contract that turns the "
"collateral-change check on remains sealed content"
),
),
_SealedCanonicalOmission(
owner=Effect,
field_name="identity_field",
Expand Down
99 changes: 98 additions & 1 deletion packages/flow/openadapt_flow/runtime/effects/_common.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@

from __future__ import annotations

from datetime import date, datetime, time
from decimal import Decimal, InvalidOperation
from typing import Any, Iterable, Optional

from openadapt_flow.runtime.effects.effect import (
Expand Down Expand Up @@ -261,6 +263,72 @@ def _judge_exact_new_set(
)


_NUMBER = (int, float, Decimal)
_WHEN = (datetime, date, time)


def _as_number(value: Any) -> Optional[Decimal]:
try:
return Decimal(str(value))
except (InvalidOperation, ValueError):
return None


def _as_when(value: Any) -> Optional[Any]:
if isinstance(value, _WHEN):
return value
if isinstance(value, str):
for parse in (datetime.fromisoformat, date.fromisoformat, time.fromisoformat):
try:
return parse(value)
except ValueError:
continue
return None


def _same_value(before: Any, after: Any) -> bool:
"""Whether a record field is unchanged between the baseline and now.

The baseline may have been written to JSON and read back (verify-run's
baseline file), while the current read has the driver's own types. So a
``Decimal('1.50')`` read now equals the ``"1.50"`` or ``1.5`` it was
stored as, a ``datetime`` equals its ISO string (either separator), and
``bytes`` equal their hex or UTF-8 text. Nested values compare element by
element, so key order doesn't matter. Two strings compare exactly: a
text field that changed from ``"02139"`` to ``"2139"`` is a change.
"""

if isinstance(before, dict) and isinstance(after, dict):
return set(before) == set(after) and all(
_same_value(before[k], after[k]) for k in before
)
if isinstance(before, (list, tuple)) and isinstance(after, (list, tuple)):
return len(before) == len(after) and all(
_same_value(b, a) for b, a in zip(before, after)
)
if before is None or after is None:
return before is None and after is None
if isinstance(before, bool) or isinstance(after, bool):
return before == after
if isinstance(before, (bytes, bytearray)) or isinstance(after, (bytes, bytearray)):
raw, other = (
(before, after)
if isinstance(before, (bytes, bytearray))
else (after, before)
)
if isinstance(other, (bytes, bytearray)):
return bytes(raw) == bytes(other)
text = str(other)
return text == bytes(raw).hex() or text == bytes(raw).decode("utf-8", "replace")
if isinstance(before, _NUMBER) or isinstance(after, _NUMBER):
b, a = _as_number(before), _as_number(after)
return b is not None and a is not None and b == a
if isinstance(before, _WHEN) or isinstance(after, _WHEN):
b, a = _as_when(before), _as_when(after)
return b is not None and a is not None and b == a
return before == after


def _judge_record_written(
effect: Effect,
before: EffectState,
Expand All @@ -285,7 +353,26 @@ def _judge_record_written(
if stable_id(r) not in current_ids:
collateral_lost.append(r)

if count_ok and not collateral_lost:
# Collateral change: a non-target record that is still there but whose
# fields differ from the pre-state (an in-place write to the wrong row).
collateral_changed: list[tuple[dict[str, Any], list[str]]] = []
if effect.forbid_collateral_change and before.reachable:
now = {stable_id(r): r for r in current}
for r in before.records:
if record_matches(r, effect.match):
continue
after = now.get(stable_id(r))
if after is None:
continue # vanished: the collateral-loss rule's case
fields = sorted(
k
for k in set(r) | set(after)
if not _same_value(r.get(k), after.get(k))
)
if fields:
collateral_changed.append((r, fields))

if count_ok and not collateral_lost and not collateral_changed:
return EffectVerdict(
verdict=Verdict.CONFIRMED,
kind=effect.kind,
Expand Down Expand Up @@ -325,6 +412,16 @@ def _judge_record_written(
f"collateral loss (stale / concurrent last-write-wins overwrote "
f"another actor's row)"
)
if collateral_changed:
names = sorted({f for _r, fields in collateral_changed for f in fields})
reasons.append(
f"{len(collateral_changed)} pre-existing record(s) outside the "
f"target changed -- collateral change (field(s): "
f"{', '.join(names[:_SAMPLE_LIMIT])}"
f"{'' if len(names) <= _SAMPLE_LIMIT else ', ...'}); this may be "
f"another user's concurrent edit, not this action -- a person "
f"should reconcile it before anything is retried or undone"
)
return EffectVerdict(
verdict=Verdict.REFUTED,
kind=effect.kind,
Expand Down
23 changes: 23 additions & 0 deletions packages/flow/openadapt_flow/runtime/effects/effect.py
Original file line number Diff line number Diff line change
Expand Up @@ -656,6 +656,16 @@ class Effect(BaseModel):
#: vanished is the collateral loss (an exact-set claim about additions
#: must not quietly tolerate removals inside the same scope).
forbid_collateral_loss: bool = True
#: ``record_written`` only: also REFUTE when a record that existed in the
#: pre-state, does NOT match :attr:`match`, and is still there (same
#: :func:`stable_id`) has CHANGED -- collateral change. The vanished-record
#: rule above misses a write that edits another record in place (a wrong
#: row updated alongside the right one). Records without an ``id`` are
#: identified by their whole content, so for them a change already reads
#: as a vanished record. Default False: a shared table that other actors
#: edit would otherwise refute every save; ``verify-run`` turns it on.
#: Bound into the contract hash only when set, so existing digests hold.
forbid_collateral_change: bool = False
#: Consequential-write flag (mirrors ``Step.risk`` / RFC ``State.risk``).
#: Compensation (``effects.compensation``) only fires for irreversible
#: effects; a reversible effect just halts.
Expand Down Expand Up @@ -743,6 +753,15 @@ def _count_new_only_scope(self) -> "Effect":
)
return self

@model_validator(mode="after")
def _collateral_change_scope(self) -> "Effect":
"""``forbid_collateral_change`` is judged on ``record_written`` only."""
if self.forbid_collateral_change and self.kind is not EffectKind.RECORD_WRITTEN:
raise ValueError(
"forbid_collateral_change applies only to record_written effects"
)
return self

@model_validator(mode="after")
def _exact_new_set_shape(self) -> "Effect":
"""Refuse an ``exact_new_set`` whose declared set is unusable, and
Expand Down Expand Up @@ -883,6 +902,8 @@ def value(expr: ValueExpr | None) -> object:
}
if self.count_new_only:
payload["count_new_only"] = True
if self.forbid_collateral_change:
payload["forbid_collateral_change"] = True
# Bound ONLY on the new kind, so every pre-existing effect's digest
# (and every ledger entry that pins it) stays byte-identical.
if self.kind is EffectKind.EXACT_NEW_SET:
Expand Down Expand Up @@ -953,6 +974,8 @@ def _contract_payload(self) -> dict[str, object]:
# its hash — PR #129) keeps its exact digest.
if self.count_new_only:
payload["count_new_only"] = True
if self.forbid_collateral_change:
payload["forbid_collateral_change"] = True
if self.kind is EffectKind.EXACT_NEW_SET:
payload["new_records"] = [
{k: str(v) for k, v in sorted(selector.items())}
Expand Down
34 changes: 31 additions & 3 deletions packages/flow/openadapt_flow/verify_run.py
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@
from openadapt_flow.runtime.effects.adapter import verifier_identity
from openadapt_flow.runtime.effects.effect import (
Effect,
EffectKind,
EffectState,
EffectVerdict,
Verdict,
Expand Down Expand Up @@ -256,7 +257,13 @@ def parse_verify_config(data: Any, *, source: str = "verify config") -> VerifyCo
raise VerifyRunError(
"give the save's checks as either effects or record, not both"
)
data["effects"] = [e.model_dump(mode="json") for e in _record_effects(record)]
# Only the fields record_effects set, so an effect's model_fields_set
# says what the file chose (``_audit_effect`` reads it); the values,
# and so the digests, are the same as qualify's.
data["effects"] = [
e.model_dump(mode="json", exclude_unset=True)
for e in _record_effects(record)
]
try:
cfg = VerifyConfig.model_validate(data)
except Exception as exc:
Expand Down Expand Up @@ -291,6 +298,25 @@ def parse_verify_config(data: Any, *, source: str = "verify config") -> VerifyCo
return cfg


def _audit_effect(effect: Effect, bound: Effect) -> Effect:
"""The effect as verify-run judges it: an audit after the fact.

A ``record_written`` check also refutes when a record outside the target
changed (``Effect.forbid_collateral_change``), not only when one vanished,
unless the file sets ``forbid_collateral_change: false`` on that effect.
The contract itself is left as the file states it, so its digest stays
the one qualify reports; the receipt's evidence says the check ran.
"""

if (
bound.kind is not EffectKind.RECORD_WRITTEN
or not bound.forbid_collateral_loss
or "forbid_collateral_change" in effect.model_fields_set
):
return bound
return bound.model_copy(update={"forbid_collateral_change": True})


# -- hashing ------------------------------------------------------------------


Expand Down Expand Up @@ -716,9 +742,10 @@ def verify_run(
)
except (KeyError, ValueError) as exc:
raise VerifyRunError(f"effect contract cannot bind params: {exc}") from exc
tier = verifier_effect_tier(verifier, bound)
judged = _audit_effect(effect, bound)
tier = verifier_effect_tier(verifier, judged)
try:
verdict = verify_effect_without_mutation(verifier, bound, before)
verdict = verify_effect_without_mutation(verifier, judged, before)
except EffectContractMutationError as exc:
raise VerifyRunError(str(exc)) from exc
if tier is None and verdict.verdict is Verdict.CONFIRMED:
Expand Down Expand Up @@ -746,6 +773,7 @@ def verify_run(
"expected_count": verdict.expected_count,
"unavailable": verdict.unavailable,
"stale": verdict.stale,
"collateral_change_checked": judged.forbid_collateral_change,
}
)

Expand Down
3 changes: 3 additions & 0 deletions packages/flow/tests/test_bundle_schema_v2.py
Original file line number Diff line number Diff line change
Expand Up @@ -208,6 +208,9 @@ def _pre_frame_path_content(wf: Workflow, shape: str) -> dict:
# key name -- see the note above.
assert effect.pop("new_records") == []
assert effect.pop("identity_field") == "id"
# Reviewed v2 omission: the collateral-change check postdates these
# seals (#1375).
assert effect.pop("forbid_collateral_change") is False
return content


Expand Down
Loading
Loading