From b1b67ce8e78b2b7caf66a92376f3188389372e61 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Sun, 26 Jul 2026 14:20:10 -0400 Subject: [PATCH] feat(runtime): explicit transaction outcome taxonomy + effect journal + idempotency (Section 3) Replace the ambiguous success/halt/failure model with a first-class terminal transaction outcome that states what is known about the BUSINESS EFFECT, plus a per-step effect journal and caller-supplied idempotency. Builds on PR #250's "verify uncertain delivery without retry" behavior rather than duplicating it. New openadapt_flow.transaction module (leaf; reads only typed report fields): - TransactionOutcome enum: VERIFIED, HALTED_BEFORE_EFFECT, RECONCILIATION_REQUIRED, FAILED_PLATFORM, CANCELED, REJECTED_POLICY, COMPLETED_UNVERIFIED (plus ROLLED_BACK carried through 1:1). - classify_transaction_outcome: refines the coarse execution_outcome. Any unresolved uncertain/conflicting delivery or persistence dominates and yields RECONCILIATION_REQUIRED (never claim "no effect", never blind-retry). A coarse HALTED otherwise splits into REJECTED_POLICY (governed/identity refusal) vs HALTED_BEFORE_EFFECT (verifier established absence); a coarse FAILED maps to CANCELED or FAILED_PLATFORM. - is_billable / is_production_success / is_platform_fault: FAILED_PLATFORM is never a billable success; COMPLETED_UNVERIFIED is never a production success. - build_effect_journal + EffectJournalEntry: PHI-free per-consequential-step ledger (intended contract hashes, attempt state, observed effect, verifier freshness, collateral reconciliation), persisted in report.effect_journal. - IdempotencyLedger: file-backed at-most-once store; a repeat under the same key is suppressed before any actuation. Honesty linchpin: EffectVerdict.observed_effect + a new observed_effect field on EffectVerificationEvidence distinguish a verifier-established absence (no write) from a conflicting/uncertain write, so HALTED_BEFORE_EFFECT is only claimed when nothing landed. Backward compatibility: execution_outcome, success, production_eligible, and outcome_envelope are unchanged; every new field is additive/optional and defaults preserve old behavior. stamp_execution_outcome additionally stamps the transaction fields. Replayer gains an optional idempotency_ledger + run() idempotency_key; with neither set the run path is unchanged. Scoped out (follow-ups, noted in docs): full saga compensation steps, the human-reconciliation-task UI, and Cloud/runner propagation of the taxonomy. Tests: tests/test_transaction_outcome.py (25) covers each outcome from its own condition, duplicate suppression, RECONCILIATION_REQUIRED no-auto-retry via the #250 path, and the non-billable/non-success flags. Full mypy + ruff + paper artifacts + consistency gates pass. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01NyCHrzA1psrKMFfroYbzaM --- README.md | 6 +- docs/EXECUTION_PROFILES.md | 44 +++ openadapt_flow/execution_profiles.py | 6 + openadapt_flow/ir.py | 130 +++++++ openadapt_flow/runtime/effects/effect.py | 29 +- openadapt_flow/runtime/replayer.py | 61 +++ openadapt_flow/transaction.py | 398 ++++++++++++++++++++ tests/test_transaction_outcome.py | 451 +++++++++++++++++++++++ 8 files changed, 1123 insertions(+), 2 deletions(-) create mode 100644 openadapt_flow/transaction.py create mode 100644 tests/test_transaction_outcome.py diff --git a/README.md b/README.md index f06382e3..771b005c 100644 --- a/README.md +++ b/README.md @@ -133,7 +133,11 @@ storage boundary may supply at-rest encryption, or `--profile demo` for an explicitly non-production run. Demo completions are `COMPLETED_UNVERIFIED`; Standard and Regulated return `VERIFIED` only when every consequential effect is confirmed at the workflow's -configured minimum evidence tier. +configured minimum evidence tier. Every run also carries a first-class +`transaction_outcome` that states what is known about the business effect +(`VERIFIED`, `HALTED_BEFORE_EFFECT`, `RECONCILIATION_REQUIRED`, `FAILED_PLATFORM`, +`CANCELED`, `REJECTED_POLICY`, `COMPLETED_UNVERIFIED`) plus a per-step effect +journal. See [execution profiles](docs/EXECUTION_PROFILES.md). **Secrets never get recorded.** An `input[type=password]` field (or any field diff --git a/docs/EXECUTION_PROFILES.md b/docs/EXECUTION_PROFILES.md index cf1eb7c8..c23a750a 100644 --- a/docs/EXECUTION_PROFILES.md +++ b/docs/EXECUTION_PROFILES.md @@ -50,3 +50,47 @@ Reports retain the legacy `success` field for compatibility and add `execution_profile`, `execution_outcome`, and `production_eligible`. Production callers must use `execution_outcome`; Standard and Regulated treat `COMPLETED_UNVERIFIED` as a non-success exit. + +## Transaction outcomes (Section 3) + +The coarse `execution_outcome` (`VERIFIED` / `COMPLETED_UNVERIFIED` / `HALTED` / +`FAILED` / `ROLLED_BACK`) is refined into a first-class **terminal transaction +outcome** that states what is known about the BUSINESS EFFECT. It is additive: +`execution_outcome`, `success`, `production_eligible`, and `outcome_envelope` +are unchanged, and the new `transaction_outcome` is derived from the same typed +evidence (see `openadapt_flow.transaction`). + +| `transaction_outcome` | Meaning | Billable | Production success | +| --- | --- | --- | --- | +| `VERIFIED` | Every declared effect (and collateral-effect check) passed at/above the required tier. | yes | yes | +| `HALTED_BEFORE_EFFECT` | The run stopped AND the verifier established that no business effect occurred. | no | no | +| `RECONCILIATION_REQUIRED` | Delivery/persistence is uncertain, conflicting, or temporarily unverifiable. The runtime does NOT blind-retry; resuming must reconcile current state first. | no | no | +| `FAILED_PLATFORM` | An OpenAdapt/platform failure before any possible effect. | no (`transaction_platform_fault=true`) | no | +| `CANCELED` | Canceled before any business effect. | no | no | +| `REJECTED_POLICY` | Authorization / identity / qualification / environment refused execution before any effect. | no | no | +| `COMPLETED_UNVERIFIED` | Demo-only completion with no production-grade effect evidence. | never | never | +| `ROLLED_BACK` | A detected duplicate / collateral write was compensated and re-verified (legacy compensation path). | no | no | + +Mapping from the coarse outcome: `VERIFIED` and `ROLLED_BACK` map through +1:1; a coarse `HALTED` splits into `REJECTED_POLICY` (a governed pre-execution +refusal or identity refusal), `HALTED_BEFORE_EFFECT` (verifier-established +absence), or `RECONCILIATION_REQUIRED` (any uncertain/conflicting delivery or +persistence, which always dominates); a coarse `FAILED` maps to `CANCELED` (when +the run was canceled) or `FAILED_PLATFORM`. + +Each run also persists an **effect journal** (`effect_journal`): one PHI-free +entry per consequential step recording the intended effect (by contract hash), +the actuation attempt state, the verifier's observed effect, verifier freshness, +and any collateral reconciliation. Callers that meter usage should read +`transaction_billable` / `transaction_platform_fault`: a `FAILED_PLATFORM` is +never a billable success and a `COMPLETED_UNVERIFIED` is never a production +success. + +**Idempotency.** A caller may pass a run-level `idempotency_key`; when the +`Replayer` is built with an `idempotency_ledger`, a repeat under the same key is +suppressed before any actuation (`idempotent_replay=true`, no consequential +action re-performed) rather than blind-retried. + +Scoped out of the runtime and tracked as follow-ups: full saga compensation +steps, the human-reconciliation-task UI, and Cloud/runner propagation of the +transaction taxonomy. diff --git a/openadapt_flow/execution_profiles.py b/openadapt_flow/execution_profiles.py index 32a6f0ab..c5f65384 100644 --- a/openadapt_flow/execution_profiles.py +++ b/openadapt_flow/execution_profiles.py @@ -280,6 +280,12 @@ def stamp_execution_outcome( elif outcome is ExecutionOutcome.ROLLED_BACK: report.success = False report.outcome_envelope = build_outcome_envelope(report, workflow) + # Section 3: refine the coarse outcome into a first-class terminal + # transaction outcome + effect journal. Additive -- reads the fields set + # above and never mutates them (leaf import; see openadapt_flow.transaction). + from openadapt_flow.transaction import stamp_transaction_outcome + + stamp_transaction_outcome(report, workflow) return outcome diff --git a/openadapt_flow/ir.py b/openadapt_flow/ir.py index 51fb12da..34f7d251 100644 --- a/openadapt_flow/ir.py +++ b/openadapt_flow/ir.py @@ -1908,6 +1908,17 @@ class EffectVerificationEvidence(BaseModel): final_verdict: Literal["confirmed", "refuted", "indeterminate"] reconciliation_completed: bool = False reconciliation_actions: int = Field(default=0, ge=0) + #: What the verifier OBSERVED about the business effect in the system of + #: record, independent of the pass/fail verdict. This is what lets a halted + #: run state honestly what is known about the write: "absent" (the verifier + #: established NO record was written -> HALTED_BEFORE_EFFECT), "conflicting" + #: (a record WAS written but is a duplicate / wrong value -> a business + #: effect that must be RECONCILED), "unknown" (the record could not be read, + #: or a refutation carried no count, so absence cannot be claimed -> + #: RECONCILIATION_REQUIRED, fail-safe), or "present" (accompanies a CONFIRMED + #: or reconciled effect). Additive; defaults to the fail-safe "unknown" so an + #: evidence record that predates this field never asserts "no effect". + observed_effect: Literal["present", "absent", "conflicting", "unknown"] = "unknown" class StepResult(BaseModel): @@ -2132,6 +2143,53 @@ def _validate_evidence_contract(self) -> "ExecutionOutcomeEnvelope": return self +class EffectJournalEntry(BaseModel): + """One consequential step's transaction-effect record (PHI-free). + + The effect journal is the per-step ledger the reconciliation model reads. + For each consequential step it states the INTENDED effect (by stable + contract hash), the ATTEMPT state (delivered / uncertain / actuated / not + actuated), the OBSERVED effect the independent verifier read from the + system of record, how FRESH that verification was, and any COLLATERAL delta + (reconciling compensations applied). It carries no record values, + parameters, or free text -- only typed evidence already present on the + :class:`StepResult`, so it can be persisted in the run's evidence output + without leaking PHI. + """ + + step_id: str + intent: str + consequential: bool = True + #: Stable, non-secret digests of the effect contracts THIS step intended to + #: apply (mirrors ``StepResult.effect_contract_hashes``). + intended_effect_contract_hashes: list[str] = Field(default_factory=list) + #: How far actuation got before verification: the write was never actuated, + #: was delivered through the GUI/native ladder, was delivered via the API + #: tier, or the action API raised AFTER delivery may have begun (the #250 + #: uncertain-delivery path -- never blind-retried). + attempt_state: Literal[ + "not_actuated", "delivered", "actuated_api", "delivery_uncertain" + ] = "not_actuated" + #: The verifier's independent read of the business effect, taken as the worst + #: case across this step's declared effects (unknown < conflicting < absent + #: < present is NOT the ordering; see the classifier -- "conflicting" and + #: "unknown" both force reconciliation, "absent" proves no effect). + observed_effect: Literal["present", "absent", "conflicting", "unknown"] = "unknown" + #: True when every declared effect was CONFIRMED at/above the required tier. + effect_verified: Optional[bool] = None + #: Whether this was an approved-but-unverified GUI write (risk accepted). + approved_unverified: bool = False + #: Verifier freshness: whether independent verification actually ran this + #: run, and the ISO-8601 instant a delivery-uncertainty was observed (the + #: tightest freshness signal we retain, present only on the #250 path). + verification_performed: bool = False + observed_at: Optional[str] = None + #: Collateral delta: count of reconciling compensation actions completed for + #: this step (0 when none). Nonzero means the system of record was mutated to + #: undo a detected duplicate / collateral write. + collateral_reconciliation_actions: int = Field(default=0, ge=0) + + class RunReport(BaseModel): workflow_name: str started_at: str @@ -2178,6 +2236,78 @@ class RunReport(BaseModel): "model calls, and external-network-call observability." ), ) + # -- Section 3: explicit transaction / reconciliation semantics ------------ + # The coarse ``execution_outcome`` (VERIFIED/HALTED/FAILED/...) is refined + # into a first-class TERMINAL TRANSACTION outcome that states what is known + # about the BUSINESS EFFECT. Additive and derived from the same typed + # evidence: ``execution_outcome`` and ``outcome_envelope`` are unchanged, so + # every existing consumer keeps working (see openadapt_flow.transaction). + transaction_outcome: Optional[ + Literal[ + "VERIFIED", + "HALTED_BEFORE_EFFECT", + "RECONCILIATION_REQUIRED", + "FAILED_PLATFORM", + "CANCELED", + "REJECTED_POLICY", + "COMPLETED_UNVERIFIED", + "ROLLED_BACK", + ] + ] = Field( + default=None, + description=( + "Terminal transaction outcome describing what is known about the " + "business effect. Refines the coarse execution_outcome without " + "replacing it." + ), + ) + transaction_billable: Optional[bool] = Field( + default=None, + description=( + "Whether this run represents a chargeable business outcome. A " + "FAILED_PLATFORM (OpenAdapt/platform fault) and a Demo-only " + "COMPLETED_UNVERIFIED are never billable." + ), + ) + transaction_platform_fault: Optional[bool] = Field( + default=None, + description=( + "True only for FAILED_PLATFORM: an OpenAdapt/platform failure that " + "occurred before any possible business effect, distinct from a " + "customer/governed outcome." + ), + ) + effect_journal: list[EffectJournalEntry] = Field( + default_factory=list, + description=( + "Per consequential-step transaction ledger (intended effect, " + "attempt state, observed effect, verifier freshness, collateral " + "delta). PHI-free; the substrate the reconciliation model reads." + ), + ) + idempotency_key: Optional[str] = Field( + default=None, + description=( + "Caller-supplied at-most-once key for this run. When an " + "idempotency ledger is configured, a repeat with the same key is " + "suppressed rather than re-actuated." + ), + ) + idempotent_replay: bool = Field( + default=False, + description=( + "True when this run was SUPPRESSED as a duplicate of a prior run " + "that already actuated under the same idempotency key. No " + "consequential action was re-performed." + ), + ) + canceled: bool = Field( + default=False, + description=( + "True when the run was canceled before any business effect could " + "occur. Drives the CANCELED terminal transaction outcome." + ), + ) execution_target_kind: Optional[ExecutionTargetKind] = Field( default=None, description=( diff --git a/openadapt_flow/runtime/effects/effect.py b/openadapt_flow/runtime/effects/effect.py index d51688c5..1b4288f1 100644 --- a/openadapt_flow/runtime/effects/effect.py +++ b/openadapt_flow/runtime/effects/effect.py @@ -42,7 +42,7 @@ from collections.abc import Mapping from datetime import datetime, timezone from enum import Enum -from typing import Any, Optional, Protocol, runtime_checkable +from typing import Any, Literal, Optional, Protocol, runtime_checkable from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator @@ -475,6 +475,33 @@ def should_halt(self) -> bool: def confirmed(self) -> bool: return self.verdict is Verdict.CONFIRMED + @property + def observed_effect(self) -> Literal["present", "absent", "conflicting", "unknown"]: + """Honest read of the BUSINESS EFFECT behind this verdict. + + Independent of pass/fail, this states what the system of record showed + so a halted run can distinguish "no write happened" from "a write may + have happened": + + - ``present``: CONFIRMED -- the effect landed and is correct. + - ``absent``: REFUTED with an observed count of zero -- the verifier + established that NO record was written (a screen may claim success, + but nothing landed). This is what makes HALTED_BEFORE_EFFECT honest. + - ``conflicting``: REFUTED with a nonzero observed count -- a record WAS + written but is a duplicate / partial / wrong value. A business effect + occurred and must be reconciled. + - ``unknown``: INDETERMINATE (system of record unreachable/unreadable), + or a REFUTED verdict that carried no count. Absence cannot be claimed; + fail safe to reconciliation. + """ + if self.verdict is Verdict.CONFIRMED: + return "present" + if self.verdict is Verdict.INDETERMINATE: + return "unknown" + if self.observed_count is None: + return "unknown" + return "absent" if self.observed_count == 0 else "conflicting" + @runtime_checkable class EffectVerifier(Protocol): diff --git a/openadapt_flow/runtime/replayer.py b/openadapt_flow/runtime/replayer.py index b5dcd5dd..4156d3ff 100644 --- a/openadapt_flow/runtime/replayer.py +++ b/openadapt_flow/runtime/replayer.py @@ -123,6 +123,7 @@ if TYPE_CHECKING: from openadapt_flow.runtime.control_overlay import RuntimeControlOverlayEmitter + from openadapt_flow.transaction import IdempotencyLedger # REGION_STABLE template check: how far the expected content may shift from # the recorded region (real apps re-layout by a few pixels between runs), @@ -364,6 +365,7 @@ def __init__( settle_readiness_timeout_s: float = 10.0, interstitials: Optional[list[Interstitial]] = None, control_overlay: Optional["RuntimeControlOverlayEmitter"] = None, + idempotency_ledger: Optional["IdempotencyLedger"] = None, ) -> None: if vision is None: import openadapt_flow.vision as vision # lazy: heavy OCR deps @@ -523,6 +525,11 @@ def __init__( self.control_overlay_error: Optional[str] = None self._control_overlay_disabled = False self._control_overlay_last_observation_png: Optional[bytes] = None + # At-most-once ledger (Section 3). When set together with a + # caller-supplied ``idempotency_key`` on run(), a repeat under the same + # key is SUPPRESSED before any actuation (never blind-retried). None + # (default) leaves the run path byte-for-byte unchanged. + self.idempotency_ledger = idempotency_ledger # -- public API ---------------------------------------------------------- @@ -538,6 +545,7 @@ def run( resume_from: Optional[int] = None, resume_program: Optional[ProgramCheckpoint] = None, run_id: Optional[str] = None, + idempotency_key: Optional[str] = None, execution_target_kind: Optional[ExecutionTargetKind] = None, execution_origin: Optional[str] = None, execution_entry_url: Optional[str] = None, @@ -592,6 +600,13 @@ def run( New runs generate one; resumed legs preserve the manifest value so attended capabilities and effect idempotency remain bound to the same logical run. + idempotency_key: Caller-supplied at-most-once key for this run + (Section 3). When the Replayer was built with an + ``idempotency_ledger``, the key is RESERVED before any + actuation; a later run that supplies the same key is SUPPRESSED + (``idempotent_replay=True``, no consequential action + re-performed) rather than blind-retried. None (default) leaves + behavior unchanged. execution_target_kind: Resolved backend token (web, windows, macos, linux, rdp, or citrix). Runtime validation binds this report field into the signed substrate-aware attestation. @@ -682,8 +697,39 @@ def run( screenshots_may_leave_box=( self._screenshots_may_leave_box or prior_screenshots_may_leave_box ), + idempotency_key=idempotency_key, ) self._begin_control_overlay(report.execution_profile) + # At-most-once idempotency (Section 3): a repeat under an already-seen + # key is SUPPRESSED before any actuation. A durable RESUME carries the + # same logical run_id and legitimately re-enters with the same key, so it + # is exempt (its own completed-effect ledger already prevents re-writes). + if ( + idempotency_key is not None + and self.idempotency_ledger is not None + and resume_from is None + and resume_program is None + ): + if self.idempotency_ledger.seen(idempotency_key): + report.results.append( + StepResult( + step_id="", + intent="suppress duplicate actuation for idempotency key", + ok=False, + failure_category="governed_refusal", + error=( + "a run with this idempotency key already actuated; " + "refusing to re-actuate the consequential write " + "-- reconcile current state before resuming" + ), + ) + ) + report.idempotent_replay = True + report.success = False + return self._finalize_report(report, workflow, run_dir) + # Reserve BEFORE any actuation: if this run crashes after the write, + # a naive retry is blocked and must reconcile rather than double-act. + self.idempotency_ledger.reserve(idempotency_key, run_id=self._run_id) if self.governed_authorization is not None: profile_refusal = self._profile_runtime_refusal(workflow) if profile_refusal is not None: @@ -1035,6 +1081,17 @@ def _finalize_report( """Persist exact evidence first, then project its terminal outcome.""" self._stamp_execution_outcome(report, workflow) + # Record the terminal transaction outcome against the reserved key so a + # later duplicate (suppressed above) can surface what already happened. + # The suppressed replay itself never overwrites the original outcome. + if ( + self.idempotency_ledger is not None + and report.idempotency_key is not None + and not report.idempotent_replay + ): + self.idempotency_ledger.record_outcome( + report.idempotency_key, report.transaction_outcome + ) report.save(run_dir) self._emit_control_overlay_terminal(report.execution_outcome) return report @@ -3773,6 +3830,7 @@ def _verify_effects( verification_tier=(int(tier) if tier is not None else None), initial_verdict=verdict.verdict.value, final_verdict=verdict.verdict.value, + observed_effect=verdict.observed_effect, ) ) result.effect_results.append( @@ -3804,6 +3862,7 @@ def _verify_effects( verification_tier=(int(tier) if tier is not None else None), initial_verdict=verdict.verdict.value, final_verdict=final_verdict.verdict.value, + observed_effect=final_verdict.observed_effect, reconciliation_completed=True, reconciliation_actions=comp.actions_taken, ) @@ -3823,6 +3882,7 @@ def _verify_effects( verification_tier=(int(tier) if tier is not None else None), initial_verdict=verdict.verdict.value, final_verdict=final_verdict.verdict.value, + observed_effect=final_verdict.observed_effect, reconciliation_actions=comp.actions_taken, ) ) @@ -3848,6 +3908,7 @@ def _verify_effects( verification_tier=(int(tier) if tier is not None else None), initial_verdict=verdict.verdict.value, final_verdict=verdict.verdict.value, + observed_effect=verdict.observed_effect, ) ) result.effect_results.append( diff --git a/openadapt_flow/transaction.py b/openadapt_flow/transaction.py new file mode 100644 index 00000000..773110cb --- /dev/null +++ b/openadapt_flow/transaction.py @@ -0,0 +1,398 @@ +"""Explicit transaction + reconciliation semantics (roadmap Section 3). + +The runtime already produces a COARSE lifecycle -- ``success`` / ``halt`` / +``failure`` -- and a first evidence-qualified projection +(``execution_outcome``: VERIFIED / COMPLETED_UNVERIFIED / HALTED / FAILED / +ROLLED_BACK). Neither of those states what is known about the BUSINESS EFFECT +when a run stops: a governed halt where nothing was written and a stop where a +consequential write may have half-landed both collapse to "HALTED / not +success". + +This module refines that coarse outcome into a first-class TERMINAL TRANSACTION +outcome (:class:`TransactionOutcome`) that describes what the evidence proves +about the effect, records a per-step :class:`~openadapt_flow.ir.EffectJournalEntry` +ledger, and provides a caller-supplied :class:`IdempotencyLedger` so a repeat +under the same key does not re-actuate. It is a LEAF: it reads only typed +fields already present on the ``RunReport`` (mirroring +``execution_profiles.build_outcome_envelope``), so it adds no import weight and +leaks no PHI. ``execution_outcome`` and ``outcome_envelope`` are unchanged -- +every existing consumer keeps working; new consumers read +``transaction_outcome``. + +Scoped OUT of this first PR (follow-ups): full saga compensation steps, the +human-reconciliation-task UI, and Cloud/runner propagation of the taxonomy. +""" + +from __future__ import annotations + +import json +import os +import tempfile +from datetime import datetime, timezone +from enum import Enum +from pathlib import Path +from typing import TYPE_CHECKING, Literal, Optional + +if TYPE_CHECKING: + from openadapt_flow.ir import EffectJournalEntry, RunReport, StepResult, Workflow + + +class TransactionOutcome(str, Enum): + """Terminal transaction outcome: what is known about the business effect. + + Every run ends in exactly one of these. The value is a superset of the + Section 3 taxonomy: ``ROLLED_BACK`` is carried through from the existing + compensation path so the legacy outcome maps 1:1 without losing + information. + """ + + #: Every declared effect (and collateral-effect check) passed at/above the + #: required tier under a production profile. The only production success. + VERIFIED = "VERIFIED" + #: The run stopped AND the verifier established that NO business effect + #: occurred (every consequential effect was observed absent, or the run + #: halted before any consequential action ran). + HALTED_BEFORE_EFFECT = "HALTED_BEFORE_EFFECT" + #: Delivery or persistence is uncertain, conflicting, or temporarily + #: unverifiable. The runtime must NOT blind-retry the consequential write; + #: resuming must reconcile current state first (builds on #250). + RECONCILIATION_REQUIRED = "RECONCILIATION_REQUIRED" + #: An OpenAdapt / platform failure before any possible business effect. + #: Never a successful billable run. + FAILED_PLATFORM = "FAILED_PLATFORM" + #: Canceled before any business effect could occur. + CANCELED = "CANCELED" + #: Authorization / identity / qualification / environment refused execution + #: before any business effect. + REJECTED_POLICY = "REJECTED_POLICY" + #: Demo-only completion with no production-grade effect evidence. MUST be + #: treated as never-billable and never a production success. + COMPLETED_UNVERIFIED = "COMPLETED_UNVERIFIED" + #: A detected duplicate / collateral write was compensated and re-verified. + #: Carried through from the existing ROLLED_BACK lifecycle; non-success. + ROLLED_BACK = "ROLLED_BACK" + + @property + def is_production_success(self) -> bool: + """Only VERIFIED counts as a production success.""" + + return self is TransactionOutcome.VERIFIED + + @property + def is_platform_fault(self) -> bool: + """True only for a platform-caused failure, distinct from a + customer/governed outcome (policy rejection, governed halt) and from an + uncertain delivery.""" + + return self is TransactionOutcome.FAILED_PLATFORM + + @property + def is_billable(self) -> bool: + """Whether this run is a chargeable business outcome. + + Conservative by design: only a VERIFIED business outcome is billable in + the runtime. A platform fault, a cancellation, a policy rejection, an + uncertain-and-unreconciled delivery, and a Demo-only + COMPLETED_UNVERIFIED are all non-billable. Broader billing policy + (e.g. charging for governed halts) is a Cloud/billing decision that is + scoped OUT of this PR; keeping the runtime default narrow guarantees a + FAILED_PLATFORM and a COMPLETED_UNVERIFIED can never be mistaken for a + billable success. + """ + + return self is TransactionOutcome.VERIFIED + + +# The gate pseudo-steps the runtime appends when it refuses BEFORE acting. +# These are policy/authorization/environment refusals, not platform faults. +_POLICY_GATE_STEP_IDS = frozenset( + {"", "", "", ""} +) + + +def _has_unresolved_uncertainty(report: RunReport) -> bool: + """True when any step's delivery or persistence is uncertain/conflicting. + + This is the signal that must dominate every non-VERIFIED coarse bucket: if a + consequential write MAY have landed (an uncertain delivery the complete + contract did not resolve, a duplicate/partial write, or an unreadable system + of record) the run can neither claim "no effect" nor be blind-retried. + """ + + for result in report.results: + uncertainty = result.delivery_uncertainty + if uncertainty is not None and not uncertainty.resolved_by_contract: + return True + for evidence in result.effect_evidence: + if evidence.final_verdict == "confirmed": + # A confirmed (or reconciled-to-confirmed) effect is settled. + continue + if evidence.final_verdict == "indeterminate": + return True + # final_verdict == "refuted": only a verifier-established ABSENCE is + # safe to treat as "no effect"; anything else may have written. + if evidence.observed_effect != "absent": + return True + return False + + +def _policy_rejected(report: RunReport) -> bool: + """True when authorization/identity/qualification/environment refused. + + A governed refusal at a pre-execution gate, or a pre-click identity check + that did not verify, stopped the run BEFORE any business effect. (The + uncertain-delivery path also carries ``failure_category='governed_refusal'`` + but is handled earlier by :func:`_has_unresolved_uncertainty`, so reaching + here means no write may have landed.) + """ + + for result in report.results: + if result.step_id in _POLICY_GATE_STEP_IDS: + return True + if ( + not result.ok + and result.identity is not None + and result.identity.status != "verified" + ): + return True + return False + + +def classify_transaction_outcome(report: RunReport) -> TransactionOutcome: + """Refine the coarse ``execution_outcome`` into a transaction outcome. + + Reads only ``report.execution_outcome`` (already stamped) plus typed step + evidence. Precedence is deliberate: a settled success/rollback first, then + any unresolved uncertainty (never claim "no effect", never blind-retry), + then a completed-but-unverified Demo, then the reason a run stopped before + any effect (policy rejection, cancellation, governed halt, platform fault). + """ + + coarse = report.execution_outcome + if coarse == "VERIFIED": + return TransactionOutcome.VERIFIED + if coarse == "ROLLED_BACK": + return TransactionOutcome.ROLLED_BACK + + # Uncertain / conflicting delivery or persistence dominates every remaining + # coarse bucket. This is where #250's no-blind-retry behavior surfaces as a + # first-class terminal outcome. + if _has_unresolved_uncertainty(report): + return TransactionOutcome.RECONCILIATION_REQUIRED + + if coarse == "COMPLETED_UNVERIFIED": + return TransactionOutcome.COMPLETED_UNVERIFIED + + # Remaining: the run did not complete and nothing may have landed. + if _policy_rejected(report): + return TransactionOutcome.REJECTED_POLICY + if report.canceled: + return TransactionOutcome.CANCELED + if coarse == "HALTED": + # A governed halt with the verifier having established no effect. + return TransactionOutcome.HALTED_BEFORE_EFFECT + # coarse == "FAILED": a platform failure before any possible effect. + return TransactionOutcome.FAILED_PLATFORM + + +def _attempt_state( + result: StepResult, +) -> Literal["not_actuated", "delivered", "actuated_api", "delivery_uncertain"]: + """Derive the PHI-free actuation attempt state for one step.""" + + if result.delivery_uncertainty is not None: + return "delivery_uncertain" + if result.actuation == "api": + return "actuated_api" + if result.delivery_receipt is not None: + return "delivered" + if result.effect_contract_hashes or result.effect_evidence: + # A consequential step whose write was actuated through the GUI ladder + # (no native delivery receipt) still reached actuation if it produced an + # effect contract or a verdict. + return "delivered" + return "not_actuated" + + +def _worst_observed_effect( + result: StepResult, +) -> Literal["present", "absent", "conflicting", "unknown"]: + """Collapse this step's effect evidence to the least-settled observation. + + ``unknown`` and ``conflicting`` (a write may have landed) outrank ``absent`` + (proven no write) which outranks ``present`` (confirmed). With no evidence + the observation is ``unknown``. + """ + + order = {"present": 0, "absent": 1, "conflicting": 2, "unknown": 3} + worst: Literal["present", "absent", "conflicting", "unknown"] = "present" + seen = False + for evidence in result.effect_evidence: + seen = True + if order[evidence.observed_effect] > order[worst]: + worst = evidence.observed_effect + return worst if seen else "unknown" + + +def build_effect_journal( + report: RunReport, workflow: Workflow +) -> list[EffectJournalEntry]: + """Build the per-consequential-step effect journal from typed evidence. + + One entry per executed consequential step (a step that declared effects, or + that the run gate classifies consequential). Carries only hashes, enums, + counts, and timestamps -- no record values, parameters, or free text. + """ + + from openadapt_flow.ir import EffectJournalEntry + from openadapt_flow.run_gate import is_consequential + from openadapt_flow.traversal import iter_workflow_steps + + steps_by_id = {step.id: step for step in iter_workflow_steps(workflow)} + consequential_ids = { + step.id for step in iter_workflow_steps(workflow) if is_consequential(step) + } + + journal: list[EffectJournalEntry] = [] + for result in report.results: + if result.skipped: + continue + step = steps_by_id.get(result.step_id) + declared_effects = bool( + result.effect_contract_hashes + or result.effect_evidence + or ( + step is not None + and ( + step.effects + or ( + step.api_binding is not None and step.api_binding.effects + ) + ) + ) + ) + is_conseq = result.step_id in consequential_ids or declared_effects + if not is_conseq: + continue + collateral = sum( + evidence.reconciliation_actions + for evidence in result.effect_evidence + if evidence.reconciliation_completed + ) + observed_at: Optional[str] = None + if result.delivery_uncertainty is not None: + observed_at = result.delivery_uncertainty.observed_at + journal.append( + EffectJournalEntry( + step_id=result.step_id, + intent=result.intent, + consequential=True, + intended_effect_contract_hashes=list(result.effect_contract_hashes), + attempt_state=_attempt_state(result), + observed_effect=_worst_observed_effect(result), + effect_verified=result.effect_verified, + approved_unverified=result.effect_approved_unverified, + verification_performed=bool(result.effect_evidence), + observed_at=observed_at, + collateral_reconciliation_actions=collateral, + ) + ) + return journal + + +def stamp_transaction_outcome(report: RunReport, workflow: Workflow) -> TransactionOutcome: + """Write the transaction outcome, billing metadata, and effect journal. + + Called after the coarse ``execution_outcome`` is stamped (see + ``execution_profiles.stamp_execution_outcome``). Never mutates + ``execution_outcome``, ``success``, ``production_eligible``, or the + ``outcome_envelope`` -- it only ADDS the Section 3 fields. + """ + + outcome = classify_transaction_outcome(report) + report.transaction_outcome = outcome.value + report.transaction_billable = outcome.is_billable + report.transaction_platform_fault = outcome.is_platform_fault + report.effect_journal = build_effect_journal(report, workflow) + return outcome + + +class DuplicateActuation(Exception): + """A run reserved an idempotency key that was already actuated.""" + + +class IdempotencyLedger: + """At-most-once ledger keyed by a caller-supplied idempotency key. + + A tiny append-only JSON store mapping ``key -> {run_id, reserved_at, + outcome}``. A run RESERVES its key before any actuation; a repeat with an + already-reserved key is SUPPRESSED (never re-actuated) -- the safe + at-most-once posture, since a crash after reservation must reconcile rather + than blind-retry the consequential write. + + Concurrency: single-writer, last-write-wins JSON (adequate for the local + runtime). A shared/locked store is a Cloud follow-up and is scoped out. + """ + + def __init__(self, path: Optional[Path | str] = None) -> None: + #: None keeps the ledger purely in memory (tests / ephemeral runs). + self.path: Optional[Path] = Path(path) if path is not None else None + self._records: dict[str, dict[str, Optional[str]]] = {} + if self.path is not None and self.path.exists(): + try: + loaded = json.loads(self.path.read_text(encoding="utf-8")) + if isinstance(loaded, dict): + self._records = loaded + except (json.JSONDecodeError, OSError): + # A corrupt ledger must not silently permit re-actuation; fail + # loud rather than treat every key as unseen. + raise + + def lookup(self, key: str) -> Optional[dict[str, Optional[str]]]: + """Return the stored record for ``key``, or None when unseen.""" + + return self._records.get(key) + + def seen(self, key: str) -> bool: + return key in self._records + + def reserve(self, key: str, *, run_id: str) -> None: + """Reserve ``key`` before actuation. Raise if already reserved.""" + + if key in self._records: + raise DuplicateActuation( + f"idempotency key {key!r} already reserved by run " + f"{self._records[key].get('run_id')!r}" + ) + self._records[key] = { + "run_id": run_id, + "reserved_at": datetime.now(timezone.utc).isoformat(), + "outcome": None, + } + self._flush() + + def record_outcome(self, key: str, outcome: Optional[str]) -> None: + """Record the terminal outcome for a previously reserved ``key``.""" + + record = self._records.get(key) + if record is None: + return + record["outcome"] = outcome + self._flush() + + def _flush(self) -> None: + if self.path is None: + return + self.path.parent.mkdir(parents=True, exist_ok=True) + # Atomic replace so a concurrent reader never sees a half-written file. + fd, tmp = tempfile.mkstemp(dir=str(self.path.parent), suffix=".tmp") + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(self._records, handle, indent=2, sort_keys=True) + os.replace(tmp, self.path) + except BaseException: + try: + os.unlink(tmp) + except OSError: + pass + raise diff --git a/tests/test_transaction_outcome.py b/tests/test_transaction_outcome.py new file mode 100644 index 00000000..cf31b120 --- /dev/null +++ b/tests/test_transaction_outcome.py @@ -0,0 +1,451 @@ +"""Section 3: explicit transaction / reconciliation outcome taxonomy. + +Covers the terminal-outcome classifier (each outcome from its own condition), +the effect journal, caller-supplied idempotency (duplicate suppression + no +blind retry after uncertain delivery), and the billing/success flags. +""" + +from __future__ import annotations + +import json + +from openadapt_flow.execution_profiles import ExecutionProfile, stamp_execution_outcome +from openadapt_flow.ir import ( + ActionKind, + EffectVerificationEvidence, + RunReport, + Step, + StepResult, + Workflow, +) +from openadapt_flow.runtime.effects import Verdict +from openadapt_flow.runtime.effects.effect import EffectKind, EffectVerdict +from openadapt_flow.runtime.replayer import Replayer +from openadapt_flow.transaction import ( + IdempotencyLedger, + TransactionOutcome, + build_effect_journal, + classify_transaction_outcome, +) +from tests.test_replayer import FakeBackend, FakeVision, Match, click_step, make_png +from tests.test_uncertain_delivery import _run + +_HASH = "sha256:" + "a" * 64 + + +def _report(coarse: str, results, *, canceled: bool = False) -> RunReport: + report = RunReport( + workflow_name="wf", + started_at="2026-07-26T00:00:00Z", + execution_outcome=coarse, + canceled=canceled, + results=results, + ) + return report + + +def _refuted_evidence(observed_effect: str) -> EffectVerificationEvidence: + return EffectVerificationEvidence( + effect_contract_hash=_HASH, + substrate="test", + initial_verdict="refuted", + final_verdict="refuted", + observed_effect=observed_effect, + ) + + +# -- classifier: each outcome from its own condition ------------------------- + + +def test_verified_maps_to_verified(): + report = _report("VERIFIED", [StepResult(step_id="s1", intent="x", ok=True)]) + assert classify_transaction_outcome(report) is TransactionOutcome.VERIFIED + + +def test_rolled_back_maps_to_rolled_back(): + report = _report("ROLLED_BACK", [StepResult(step_id="s1", intent="x", ok=True)]) + assert classify_transaction_outcome(report) is TransactionOutcome.ROLLED_BACK + + +def test_halted_before_effect_requires_verifier_established_absence(): + # A governed halt where the verifier proved NO record was written. + result = StepResult( + step_id="s1", + intent="write", + ok=False, + safety_halt=True, + effect_verified=False, + effect_contract_hashes=[_HASH], + effect_evidence=[_refuted_evidence("absent")], + ) + report = _report("HALTED", [result]) + assert ( + classify_transaction_outcome(report) + is TransactionOutcome.HALTED_BEFORE_EFFECT + ) + + +def test_indeterminate_effect_forces_reconciliation(): + result = StepResult( + step_id="s1", + intent="write", + ok=False, + effect_verified=False, + effect_contract_hashes=[_HASH], + effect_evidence=[ + EffectVerificationEvidence( + effect_contract_hash=_HASH, + substrate="test", + initial_verdict="indeterminate", + final_verdict="indeterminate", + observed_effect="unknown", + ) + ], + ) + report = _report("HALTED", [result]) + assert ( + classify_transaction_outcome(report) + is TransactionOutcome.RECONCILIATION_REQUIRED + ) + + +def test_conflicting_effect_forces_reconciliation(): + # A refuted-but-present write (duplicate / wrong value) may have landed. + result = StepResult( + step_id="s1", + intent="write", + ok=False, + effect_verified=False, + effect_contract_hashes=[_HASH], + effect_evidence=[_refuted_evidence("conflicting")], + ) + report = _report("HALTED", [result]) + assert ( + classify_transaction_outcome(report) + is TransactionOutcome.RECONCILIATION_REQUIRED + ) + + +def test_refuted_without_count_is_treated_as_unknown_and_reconciled(): + # Fail safe: a refutation that cannot prove absence must not claim "no + # effect". + result = StepResult( + step_id="s1", + intent="write", + ok=False, + effect_verified=False, + effect_evidence=[_refuted_evidence("unknown")], + ) + report = _report("HALTED", [result]) + assert ( + classify_transaction_outcome(report) + is TransactionOutcome.RECONCILIATION_REQUIRED + ) + + +def test_failed_platform_before_any_effect(): + result = StepResult(step_id="s1", intent="click", ok=False, error="backend crash") + report = _report("FAILED", [result]) + assert classify_transaction_outcome(report) is TransactionOutcome.FAILED_PLATFORM + + +def test_canceled_before_effect(): + result = StepResult(step_id="s1", intent="click", ok=False) + report = _report("FAILED", [result], canceled=True) + assert classify_transaction_outcome(report) is TransactionOutcome.CANCELED + + +def test_rejected_policy_from_authorization_gate(): + result = StepResult( + step_id="", + intent="validate governed run authorization", + ok=False, + failure_category="governed_refusal", + error="not admitted", + ) + report = _report("HALTED", [result]) + assert classify_transaction_outcome(report) is TransactionOutcome.REJECTED_POLICY + + +def test_rejected_policy_from_identity_refusal(): + from openadapt_flow.ir import IdentityCheck + + result = StepResult( + step_id="s1", + intent="click patient row", + ok=False, + identity=IdentityCheck(status="mismatch"), + ) + report = _report("HALTED", [result]) + assert classify_transaction_outcome(report) is TransactionOutcome.REJECTED_POLICY + + +def test_completed_unverified_maps_through(): + report = _report( + "COMPLETED_UNVERIFIED", [StepResult(step_id="s1", intent="x", ok=True)] + ) + assert ( + classify_transaction_outcome(report) + is TransactionOutcome.COMPLETED_UNVERIFIED + ) + + +def test_uncertainty_dominates_a_concurrent_policy_refusal(): + # An unresolved uncertain delivery must never be downgraded to "no effect" + # even when a governed-refusal category is also present. + from openadapt_flow.ir import ActionDeliveryUncertainty + + result = StepResult( + step_id="s1", + intent="write", + ok=False, + failure_category="governed_refusal", + delivery_uncertainty=ActionDeliveryUncertainty( + operation="guarded_coordinate_click", + native=False, + observed_at="2026-07-26T00:00:00.000000+00:00", + cause_type="ConnectionResetError", + resolved_by_contract=False, + ), + ) + report = _report("HALTED", [result]) + assert ( + classify_transaction_outcome(report) + is TransactionOutcome.RECONCILIATION_REQUIRED + ) + + +# -- billing / success flags ------------------------------------------------- + + +def test_failed_platform_is_not_billable_and_is_a_platform_fault(): + outcome = TransactionOutcome.FAILED_PLATFORM + assert outcome.is_billable is False + assert outcome.is_production_success is False + assert outcome.is_platform_fault is True + + +def test_completed_unverified_is_never_billable_or_production_success(): + outcome = TransactionOutcome.COMPLETED_UNVERIFIED + assert outcome.is_billable is False + assert outcome.is_production_success is False + assert outcome.is_platform_fault is False + + +def test_only_verified_is_billable_production_success(): + assert TransactionOutcome.VERIFIED.is_billable is True + assert TransactionOutcome.VERIFIED.is_production_success is True + assert TransactionOutcome.VERIFIED.is_platform_fault is False + for other in ( + TransactionOutcome.HALTED_BEFORE_EFFECT, + TransactionOutcome.RECONCILIATION_REQUIRED, + TransactionOutcome.CANCELED, + TransactionOutcome.REJECTED_POLICY, + TransactionOutcome.ROLLED_BACK, + ): + assert other.is_billable is False + assert other.is_production_success is False + + +# -- effect journal ---------------------------------------------------------- + + +def test_effect_journal_records_intended_attempt_and_observed(): + workflow = Workflow( + name="wf", + steps=[Step(id="save", intent="save", action=ActionKind.CLICK)], + ) + result = StepResult( + step_id="save", + intent="save", + ok=False, + effect_verified=False, + effect_contract_hashes=[_HASH], + effect_evidence=[_refuted_evidence("absent")], + ) + report = _report("HALTED", [result]) + journal = build_effect_journal(report, workflow) + assert len(journal) == 1 + entry = journal[0] + assert entry.step_id == "save" + assert entry.intended_effect_contract_hashes == [_HASH] + assert entry.attempt_state == "delivered" + assert entry.observed_effect == "absent" + assert entry.effect_verified is False + assert entry.verification_performed is True + assert entry.collateral_reconciliation_actions == 0 + + +def test_effect_journal_skips_non_consequential_steps(): + workflow = Workflow( + name="wf", + steps=[Step(id="nav", intent="scroll", action=ActionKind.SCROLL)], + ) + result = StepResult(step_id="nav", intent="scroll", ok=True) + report = _report("COMPLETED_UNVERIFIED", [result]) + assert build_effect_journal(report, workflow) == [] + + +# -- stamping (end-to-end through execution_profiles) ------------------------ + + +def test_stamp_adds_transaction_fields_without_touching_execution_outcome(): + workflow = Workflow( + name="wf", + steps=[Step(id="save", intent="save", action=ActionKind.CLICK)], + ) + report = RunReport( + workflow_name="wf", + started_at="2026-07-26T00:00:00Z", + success=True, + results=[StepResult(step_id="save", intent="save", ok=True)], + ) + stamp_execution_outcome(report, workflow, ExecutionProfile.DEMO) + # Coarse outcome is unchanged; the refined transaction outcome is added. + assert report.execution_outcome == "COMPLETED_UNVERIFIED" + assert report.transaction_outcome == "COMPLETED_UNVERIFIED" + assert report.transaction_billable is False + assert report.transaction_platform_fault is False + # Round-trips through the persisted report shape. + reloaded = RunReport.model_validate(json.loads(report.model_dump_json())) + assert reloaded.transaction_outcome == "COMPLETED_UNVERIFIED" + + +# -- idempotency: duplicate suppression, no re-actuation --------------------- + + +def _simple_click_run(backend, ledger, run_dir, bundle, key): + vision = FakeVision() + vision.template_results = [ + Match(point=(110, 105), region=(100, 100, 50, 20), confidence=0.95) + ] + replayer = Replayer( + backend, + vision=vision, + poll_interval_s=0.01, + idempotency_ledger=ledger, + ) + return replayer.run( + Workflow(name="wf", steps=[click_step(risk="reversible")]), + bundle_dir=bundle, + run_dir=run_dir, + idempotency_key=key, + ) + + +def test_idempotency_key_suppresses_duplicate_actuation(tmp_path): + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + (bundle / "templates" / "btn.png").write_bytes(make_png((50, 20))) + ledger = IdempotencyLedger(tmp_path / "ledger.json") + + first_backend = FakeBackend() + first = _simple_click_run( + first_backend, ledger, tmp_path / "run1", bundle, "order-42" + ) + assert first.idempotent_replay is False + assert first_backend.actions == [("click", 110, 105, False)] + assert ledger.seen("order-42") + + # A fresh backend proves the second run cannot have actuated. + second_backend = FakeBackend() + second = _simple_click_run( + second_backend, ledger, tmp_path / "run2", bundle, "order-42" + ) + assert second.idempotent_replay is True + assert second_backend.actions == [] + assert second.transaction_outcome == "REJECTED_POLICY" + assert second.success is False + + +def test_idempotency_reservation_survives_new_ledger_instance(tmp_path): + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + (bundle / "templates" / "btn.png").write_bytes(make_png((50, 20))) + + ledger = IdempotencyLedger(tmp_path / "ledger.json") + _simple_click_run(FakeBackend(), ledger, tmp_path / "run1", bundle, "k") + + # A process restart re-reads the persisted ledger and still suppresses. + reopened = IdempotencyLedger(tmp_path / "ledger.json") + backend = FakeBackend() + report = _simple_click_run(backend, reopened, tmp_path / "run2", bundle, "k") + assert report.idempotent_replay is True + assert backend.actions == [] + + +def test_different_keys_do_not_suppress(tmp_path): + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + (bundle / "templates" / "btn.png").write_bytes(make_png((50, 20))) + ledger = IdempotencyLedger(tmp_path / "ledger.json") + + backend = FakeBackend() + _simple_click_run(backend, ledger, tmp_path / "run1", bundle, "a") + other = FakeBackend() + report = _simple_click_run(other, ledger, tmp_path / "run2", bundle, "b") + assert report.idempotent_replay is False + assert other.actions == [("click", 110, 105, False)] + + +# -- no blind retry after uncertain delivery (builds on #250) ---------------- + + +def test_reconciliation_required_does_not_auto_retry(tmp_path): + # A refuted uncertain delivery: the write was attempted exactly once and the + # runtime does NOT re-actuate. Section 3 surfaces this as + # RECONCILIATION_REQUIRED (delivery uncertain, not resolved by the contract). + report, backend, verifier, _workflow, _bundle = _run(tmp_path, Verdict.REFUTED) + assert backend.actuation_count == 1 + assert verifier.verify_calls == 1 + assert report.results[0].delivery_uncertainty.resolved_by_contract is False + assert report.transaction_outcome == "RECONCILIATION_REQUIRED" + assert report.transaction_billable is False + + +def test_indeterminate_uncertain_delivery_is_reconciliation_required(tmp_path): + report, backend, _verifier, _workflow, _bundle = _run( + tmp_path, Verdict.INDETERMINATE + ) + assert backend.actuation_count == 1 + assert report.transaction_outcome == "RECONCILIATION_REQUIRED" + + +def test_resolved_uncertain_delivery_is_verified(tmp_path): + report, backend, _verifier, _workflow, _bundle = _run(tmp_path, Verdict.CONFIRMED) + assert backend.actuation_count == 1 + assert report.transaction_outcome == "VERIFIED" + assert report.transaction_billable is True + + +def test_effect_verdict_observed_effect_mapping(): + assert ( + EffectVerdict( + verdict=Verdict.CONFIRMED, kind=EffectKind.RECORD_WRITTEN, observed_count=1 + ).observed_effect + == "present" + ) + assert ( + EffectVerdict( + verdict=Verdict.REFUTED, kind=EffectKind.RECORD_WRITTEN, observed_count=0 + ).observed_effect + == "absent" + ) + assert ( + EffectVerdict( + verdict=Verdict.REFUTED, kind=EffectKind.RECORD_WRITTEN, observed_count=3 + ).observed_effect + == "conflicting" + ) + assert ( + EffectVerdict( + verdict=Verdict.INDETERMINATE, kind=EffectKind.RECORD_WRITTEN + ).observed_effect + == "unknown" + ) + assert ( + EffectVerdict( + verdict=Verdict.REFUTED, kind=EffectKind.RECORD_WRITTEN + ).observed_effect + == "unknown" + )