diff --git a/CAPABILITIES.md b/CAPABILITIES.md index b29347101..0877806c6 100644 --- a/CAPABILITIES.md +++ b/CAPABILITIES.md @@ -185,6 +185,7 @@ improvising or retrying when something is off. | Run an API step in a durable run: checkpoint the span after the save is sent, never resend or redo a save that may have landed, and reconcile it by reading the record back | In review ([#1358](https://github.com/OpenAdaptAI/OpenAdapt/pull/1358)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#in-a-durable-run) | | Run an admitted API step under a governed run authorization and the Standard or Regulated profile: the authorization names the API step by digest, the person who admits it names the right-record parameter, and the outcome reads `VERIFIED` | In review ([#1361](https://github.com/OpenAdaptAI/OpenAdapt/pull/1361)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#under-governance) | | Defer an API save's read-back with pipelined read-back, as for a GUI save, and keep self-repair's probes and resumed run on the repaired GUI step | In review ([#1362](https://github.com/OpenAdaptAI/OpenAdapt/pull/1362)) | Synthetic | [API_TIER.md](packages/flow/docs/API_TIER.md#with-parallel-workers-pipelined-read-back-and-self-repair) | +| Say, for every feature under every run profile, whether it works or is refused and why, checked against the runtime in CI | In review ([#1364](https://github.com/OpenAdaptAI/OpenAdapt/pull/1364)) | Synthetic | [COMPATIBILITY.md](packages/flow/docs/COMPATIBILITY.md) | | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](packages/flow/docs/THROUGHPUT.md) | diff --git a/llms-full.txt b/llms-full.txt index 2fcd25df9..9c07878c5 100644 --- a/llms-full.txt +++ b/llms-full.txt @@ -541,6 +541,7 @@ improvising or retrying when something is off. | Run an API step in a durable run: checkpoint the span after the save is sent, never resend or redo a save that may have landed, and reconcile it by reading the record back | In review ([#1358](https://github.com/OpenAdaptAI/OpenAdapt/pull/1358)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#in-a-durable-run) | | Run an admitted API step under a governed run authorization and the Standard or Regulated profile: the authorization names the API step by digest, the person who admits it names the right-record parameter, and the outcome reads `VERIFIED` | In review ([#1361](https://github.com/OpenAdaptAI/OpenAdapt/pull/1361)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#under-governance) | | Defer an API save's read-back with pipelined read-back, as for a GUI save, and keep self-repair's probes and resumed run on the repaired GUI step | In review ([#1362](https://github.com/OpenAdaptAI/OpenAdapt/pull/1362)) | Synthetic | [API_TIER.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/API_TIER.md#with-parallel-workers-pipelined-read-back-and-self-repair) | +| Say, for every feature under every run profile, whether it works or is refused and why, checked against the runtime in CI | In review ([#1364](https://github.com/OpenAdaptAI/OpenAdapt/pull/1364)) | Synthetic | [COMPATIBILITY.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/COMPATIBILITY.md) | | Run a worklist on several workers at once, one signed-in session each, each item claimed once, with a stop on any worker stopping all | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic patients on a local OpenEMR test copy | `--workers`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | | Read a save back while the next item runs (pipelined read-back, REST, FHIR, GraphQL or SQL record checks) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `--readback-window`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | | A faster speed profile that keeps every check (one insert per plain text field, shorter page-quiet wait) | In review ([#1296](https://github.com/OpenAdaptAI/OpenAdapt/pull/1296)) | Synthetic | `runtime.speed: fast`, [THROUGHPUT.md](https://github.com/OpenAdaptAI/OpenAdapt/blob/main/packages/flow/docs/THROUGHPUT.md) | diff --git a/packages/flow/docs/API_TIER.md b/packages/flow/docs/API_TIER.md index 029781876..2f56c5a93 100644 --- a/packages/flow/docs/API_TIER.md +++ b/packages/flow/docs/API_TIER.md @@ -220,6 +220,9 @@ Not yet covered, so the GUI steps run with that reason: probes and resumed run use the GUI steps, because the repair and the person's approval are about the GUI step ([SELF_REPAIR.md](SELF_REPAIR.md)). +Every feature under every run profile, and the API step with each other +feature, is in [COMPATIBILITY.md](COMPATIBILITY.md). + ## When it doesn't run The API step runs only in a plain linear run. It doesn't run, and the GUI diff --git a/packages/flow/docs/COMPATIBILITY.md b/packages/flow/docs/COMPATIBILITY.md new file mode 100644 index 000000000..9b4b5c395 --- /dev/null +++ b/packages/flow/docs/COMPATIBILITY.md @@ -0,0 +1,180 @@ +# Compatibility: features and run profiles + + + +Each feature either works under a run profile or is refused with a +named reason. Nothing falls back silently. When a run takes the slower or +narrower path, it records why: an API step's `api_fallback`, a +pipelined read-back's `inline_reason`, or the command's refusal before +anything runs. `tests/test_compatibility_matrix.py` checks every cell +against the runtime, and fails when a cell that should work falls back +or a refusal gives another reason. + +## The widest combination today + +A Regulated, durable, governed run with per-item approval, on parallel +workers, using the API step for an admitted save, is the combination +where the most work together: each worker's item sends one request a +person approved, checks it with the system-of-record read-back, and +reads `VERIFIED`. In that run the read-back runs inline (durable), and +self-repair runs through the Python API under the Standard contract. +Under a signed qualification admission (what `openadapt-flow run` +requires for Standard and Regulated actuation) the API step and +self-repair are still refused, each with its reason below. + +## Profiles + +- **Default (gated).** `openadapt-flow run` with no `--profile`: a run authorization from the run gate, and per-item approval (`each_item`), the default approval mode. +- **Durable.** `--durable`: a checkpoint after every verified step, and pause and resume from the last one. +- **Governed.** The Standard or Regulated profile contract with a run authorization from the run gate: durable, an identity check on every save, and a record check that reads the system of record. This is how the Python API and the tutorial run it. +- **Pinned.** `--pin-digest`: the run refuses any bundle but the one with that sealed content digest. +- **Qualified.** A signed production qualification admission (`--qualification-authority-file`), which `openadapt-flow run` requires for Standard and Regulated actuation. +- **Managed.** A Cloud-dispatched run (`--managed-dispatch-file`), whose run authorization comes from Cloud. + +## Features + +- **API step** ([API_TIER.md](API_TIER.md)). An admitted save compiled from network capture, run as one request in place of its GUI steps. +- **Pipelined read-back** ([THROUGHPUT.md](THROUGHPUT.md)). `--readback-window`: an item's record check runs while the next item starts. +- **Self-repair** ([SELF_REPAIR.md](SELF_REPAIR.md)). `--self-repair`: an agent proposes a fix for a halt on a changed screen; probes and a person's approval decide it. +- **Standard changes** ([SELF_REPAIR.md](SELF_REPAIR.md#standard-changes-repairs-without-a-person)). Self-repair fixes of a class a person approved in advance, in a signed repair policy. +- **Parallel workers** ([THROUGHPUT.md](THROUGHPUT.md)). `--workers N`: worklist items on several signed-in sessions at once. +- **Multi-surface** ([MULTI_SURFACE.md](MULTI_SURFACE.md)). One program across several windows, tabs or apps. + +## At a glance + +| Feature | Default (gated) | Durable | Governed | Pinned | Qualified | Managed | +| --- | --- | --- | --- | --- | --- | --- | +| API step | Works | Works | Works | Works | Refused | Refused | +| Pipelined read-back | Works | Refused | Refused | Works | Refused | Refused | +| Self-repair | Works | Works | Works | Refused | Refused | Refused | +| Standard changes | Works | Works | Works | Refused | Refused | Refused | +| Parallel workers | Works | Works | Works | Works | Works | Refused | +| Multi-surface | Works | Works | Works | Works | Works | Works | + +Every refusal names its reason (below), and a run or command that +refuses says it in those words. + +## Features by profile + +### API step + +| Profile | Result | +| --- | --- | +| Default (gated) | Works: the person approves the exact request | +| Durable | Works: a save that may have landed is never resent | +| Governed | Works: needs `api admit --identity` | +| Pinned | Works: the pin is checked before the run | +| Qualified | Refused: a production qualification admission doesn't cover API steps yet, so the qualified GUI path runs | +| Managed | Refused: a managed dispatch doesn't bind API steps yet | + +### Pipelined read-back + +| Profile | Result | +| --- | --- | +| Default (gated) | Works: GUI and API saves | +| Durable | Refused: durable runs verify inline (the record check runs before the next item) | +| Governed | Refused: durable runs verify inline (Standard and Regulated are durable) | +| Pinned | Works: when the run isn't durable | +| Qualified | Refused: durable runs verify inline (Standard and Regulated are durable) | +| Managed | Refused: --workers and --readback-window apply to local runs, not a managed dispatch or a qualification case | + +### Self-repair + +| Profile | Result | +| --- | --- | +| Default (gated) | Works | +| Durable | Works: self-repair always runs durable | +| Governed | Works: Standard contract through the Python API | +| Pinned | Refused: --pin-digest pins the original bundle | +| Qualified | Refused: a signed qualification admission binds the original bundle; the approved bundle needs its own admission | +| Managed | Refused: a managed dispatch binds the original bundle | + +### Standard changes + +| Profile | Result | +| --- | --- | +| Default (gated) | Works: needs a signed repair policy | +| Durable | Works: needs a signed repair policy | +| Governed | Works: needs a signed repair policy | +| Pinned | Refused: --pin-digest pins the original bundle | +| Qualified | Refused: a signed qualification admission binds the original bundle; the approved bundle needs its own admission | +| Managed | Refused: a managed dispatch binds the original bundle | + +### Parallel workers + +| Profile | Result | +| --- | --- | +| Default (gated) | Works | +| Durable | Works: each item has its own durable run | +| Governed | Works: each item gets its own single-use authorization | +| Pinned | Works | +| Qualified | Works: the admission is checked for every item | +| Managed | Refused: --workers and --readback-window apply to local runs, not a managed dispatch or a qualification case | + +### Multi-surface + +| Profile | Result | +| --- | --- | +| Default (gated) | Works: an API step inside it runs on the GUI (below) | +| Durable | Works: resume re-binds each surface by identity | +| Governed | Works | +| Pinned | Works | +| Qualified | Works: one environment boundary per surface | +| Managed | Works: every backend surface must be bound | + +## The API step with the other features + +| With | Result | +| --- | --- | +| Parallel workers | Works: each item runs its own API step | +| Pipelined read-back | Works: when the run isn't durable | +| Self-repair | Refused: self-repair's probes and resumed run exercise the repaired GUI step, and the person approved that step, so its API step doesn't run (the first, ordinary run can use the API step) | +| Standard changes | Refused: self-repair's probes and resumed run exercise the repaired GUI step, and the person approved that step, so its API step doesn't run (the repaired bundle's test-record run is GUI-only too) | +| Multi-surface | Refused: a run across several windows or tabs uses the GUI steps | + +## Evidence + +Each cell names the tests that show it. They run in CI. + +- API step / Default (gated): `tests/test_api_step_approval.py::test_the_person_approves_the_exact_request_and_it_is_sent_once`, `tests/test_compatibility_matrix.py::test_api_step_runs_where_the_matrix_says` +- API step / Durable: `tests/test_api_step_durable.py::test_a_durable_run_uses_the_api_step_and_checkpoints_after_it`, `tests/test_api_step_durable.py::test_an_uncertain_send_is_never_resent_by_resume` +- API step / Governed: `tests/test_api_step_governed.py::test_the_default_profile_runs_the_admitted_api_step` +- API step / Pinned: `tests/test_compatibility_matrix.py::test_api_step_runs_where_the_matrix_says` +- API step / Qualified: `tests/test_compatibility_matrix.py::test_api_step_refuses_where_the_matrix_says` +- API step / Managed: `tests/test_compatibility_matrix.py::test_api_step_refuses_where_the_matrix_says` +- Pipelined read-back / Default (gated): `tests/test_readback_pipeline.py::test_confirmed_readback_settles_the_saved_report`, `tests/test_api_step_readback_repair.py::test_an_api_save_reads_back_while_the_next_item_runs`, `tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says` +- Pipelined read-back / Durable: `tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says` +- Pipelined read-back / Governed: `tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says` +- Pipelined read-back / Pinned: `tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says` +- Pipelined read-back / Qualified: `tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says` +- Pipelined read-back / Managed: `tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says` +- Self-repair / Default (gated): `tests/test_self_repair_cli.py::test_run_self_repair_cli_resumes_after_one_phone_tap`, `tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says` +- Self-repair / Durable: `tests/test_self_repair.py::test_renamed_button_is_repaired_approved_on_the_phone_and_verified` +- Self-repair / Governed: `tests/test_self_repair.py::test_renamed_button_is_repaired_approved_on_the_phone_and_verified` +- Self-repair / Pinned: `tests/test_self_repair_cli.py::test_run_self_repair_refuses_a_pinned_bundle`, `tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says` +- Self-repair / Qualified: `tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says` +- Self-repair / Managed: `tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says` +- Standard changes / Default (gated): `tests/test_standard_change.py::test_relabelled_link_is_repaired_without_a_person` +- Standard changes / Durable: `tests/test_standard_change.py::test_relabelled_link_is_repaired_without_a_person` +- Standard changes / Governed: `tests/test_standard_change.py::test_relabelled_link_is_repaired_without_a_person` +- Standard changes / Pinned: `tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says` +- Standard changes / Qualified: `tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says` +- Standard changes / Managed: `tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says` +- Parallel workers / Default (gated): `tests/test_cli_worklist_pool.py::test_two_workers_split_the_items_one_row_each`, `tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says` +- Parallel workers / Durable: `tests/test_compatibility_matrix.py::test_workers_run_governed_durable_api_items` +- Parallel workers / Governed: `tests/test_compatibility_matrix.py::test_workers_run_governed_durable_api_items` +- Parallel workers / Pinned: `tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says` +- Parallel workers / Qualified: `tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says` +- Parallel workers / Managed: `tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says` +- Multi-surface / Default (gated): `tests/test_multi_surface_governed.py::test_governed_durable_multi_surface_run_reaches_verified` +- Multi-surface / Durable: `tests/test_multi_surface_identity.py::test_resume_rebinds_surfaces_by_identity` +- Multi-surface / Governed: `tests/test_multi_surface_governed.py::test_governed_durable_multi_surface_run_reaches_verified` +- Multi-surface / Pinned: `tests/test_multi_surface_governed.py::test_governed_durable_multi_surface_run_reaches_verified` +- Multi-surface / Qualified: `tests/test_multi_surface_governed.py::test_case_evidence_needs_a_matching_observation_per_used_surface` +- Multi-surface / Managed: `tests/test_multi_surface_contract.py::test_managed_dispatch_needs_every_backend_surface_bound` +- API step / Parallel workers: `tests/test_compatibility_matrix.py::test_workers_run_governed_durable_api_items` +- API step / Pipelined read-back: `tests/test_api_step_readback_repair.py::test_an_api_save_reads_back_while_the_next_item_runs` +- API step / Self-repair: `tests/test_api_step_readback_repair.py::test_a_probe_or_resumed_run_never_sends_the_api_step` +- API step / Standard changes: `tests/test_api_step_readback_repair.py::test_a_standard_change_test_record_run_is_gui_only` +- API step / Multi-surface: `tests/test_compatibility_matrix.py::test_api_step_refuses_in_a_multi_surface_run` diff --git a/packages/flow/openadapt_flow/__main__.py b/packages/flow/openadapt_flow/__main__.py index 46023e918..3fdb5bf40 100644 --- a/packages/flow/openadapt_flow/__main__.py +++ b/packages/flow/openadapt_flow/__main__.py @@ -3603,12 +3603,14 @@ def _cmd_run(args: argparse.Namespace) -> int: # Parallel worklist (docs/THROUGHPUT.md): every item gets its own # single-use authorization, bound to that item's one-row worklist and # otherwise identical to the run's (same gate, policy, approvals). - if dispatch_file or qualification_case is not None: - print( - "run REFUSED: --workers and --readback-window apply to local " - "runs, not a managed dispatch or a qualification case. Nothing " - "was executed." - ) + from openadapt_flow.cli_worklist_pool import pool_profile_refusal + + pool_refused = pool_profile_refusal( + managed=bool(dispatch_file), + qualification_case=qualification_case is not None, + ) + if pool_refused is not None: + print(f"run REFUSED: {pool_refused}. Nothing was executed.") return 2 try: relation = loop_relation(workflow) diff --git a/packages/flow/openadapt_flow/cli_worklist_pool.py b/packages/flow/openadapt_flow/cli_worklist_pool.py index 43d6029a3..a09f17bf7 100644 --- a/packages/flow/openadapt_flow/cli_worklist_pool.py +++ b/packages/flow/openadapt_flow/cli_worklist_pool.py @@ -60,6 +60,21 @@ def requested_pool(args: Any, cfg: Any) -> Optional[PoolPolicy]: ) +def pool_profile_refusal(*, managed: bool, qualification_case: bool) -> Optional[str]: + """Why ``--workers`` / ``--readback-window`` can't run here, or None. + + Workers and a read-back window are for local runs. A managed dispatch and + a qualification case each bind one exact run. + """ + + if managed or qualification_case: + return ( + "--workers and --readback-window apply to local runs, not a managed " + "dispatch or a qualification case" + ) + return None + + def loop_relation(workflow: Any) -> str: """The relation of a single-LOOP program (the ``for-each`` shape). diff --git a/packages/flow/openadapt_flow/compatibility.py b/packages/flow/openadapt_flow/compatibility.py new file mode 100644 index 000000000..f5a5cbc2b --- /dev/null +++ b/packages/flow/openadapt_flow/compatibility.py @@ -0,0 +1,655 @@ +"""Which features work under which run profiles: the compatibility matrix. + +Each cell either works or is refused with a named reason. A refused feature +never falls back silently: the run either says why it used the slower or +narrower path (an API step's ``api_fallback``, a pipelined read-back's +``inline_reason``), or the command refuses before anything runs. + +``tests/test_compatibility_matrix.py`` checks every cell against the runtime: +a cell that works must engage the feature, and a refused cell must produce +exactly the reason written here. ``docs/COMPATIBILITY.md`` is generated from +this module by ``scripts/build_compatibility_doc.py``. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Literal, Optional + +Status = Literal["works", "refused"] + + +@dataclass(frozen=True) +class Profile: + key: str + title: str + description: str + + +@dataclass(frozen=True) +class Feature: + key: str + title: str + description: str + doc: str + + +@dataclass(frozen=True) +class Cell: + feature: str + profile: str + status: Status + #: For a refused cell: the reason the run or the command gives, verbatim. + reason: Optional[str] = None + #: A short note for the table, such as a condition the cell needs. + note: Optional[str] = None + #: The tests that show the cell's behavior (``file::test``). + evidence: tuple[str, ...] = () + + +PROFILES: tuple[Profile, ...] = ( + Profile( + "default", + "Default (gated)", + "`openadapt-flow run` with no `--profile`: a run authorization from the " + "run gate, and per-item approval (`each_item`), the default approval mode.", + ), + Profile( + "durable", + "Durable", + "`--durable`: a checkpoint after every verified step, and pause and " + "resume from the last one.", + ), + Profile( + "governed", + "Governed", + "The Standard or Regulated profile contract with a run authorization " + "from the run gate: durable, an identity check on every save, and a " + "record check that reads the system of record. This is how the Python " + "API and the tutorial run it.", + ), + Profile( + "pinned", + "Pinned", + "`--pin-digest`: the run refuses any bundle but the one with that " + "sealed content digest.", + ), + Profile( + "qualified", + "Qualified", + "A signed production qualification admission " + "(`--qualification-authority-file`), which `openadapt-flow run` " + "requires for Standard and Regulated actuation.", + ), + Profile( + "managed", + "Managed", + "A Cloud-dispatched run (`--managed-dispatch-file`), whose run " + "authorization comes from Cloud.", + ), +) + +FEATURES: tuple[Feature, ...] = ( + Feature( + "api_step", + "API step", + "An admitted save compiled from network capture, run as one request in " + "place of its GUI steps.", + "API_TIER.md", + ), + Feature( + "pipelined_readback", + "Pipelined read-back", + "`--readback-window`: an item's record check runs while the next item starts.", + "THROUGHPUT.md", + ), + Feature( + "self_repair", + "Self-repair", + "`--self-repair`: an agent proposes a fix for a halt on a changed " + "screen; probes and a person's approval decide it.", + "SELF_REPAIR.md", + ), + Feature( + "standard_changes", + "Standard changes", + "Self-repair fixes of a class a person approved in advance, in a " + "signed repair policy.", + "SELF_REPAIR.md#standard-changes-repairs-without-a-person", + ), + Feature( + "parallel_workers", + "Parallel workers", + "`--workers N`: worklist items on several signed-in sessions at once.", + "THROUGHPUT.md", + ), + Feature( + "multi_surface", + "Multi-surface", + "One program across several windows, tabs or apps.", + "MULTI_SURFACE.md", + ), +) + +API_QUALIFIED = ( + "a production qualification admission doesn't cover API steps yet, so the " + "qualified GUI path runs" +) +API_MANAGED = "a managed dispatch doesn't bind API steps yet" +DURABLE_INLINE = "durable runs verify inline" +POOL_LOCAL_ONLY = ( + "--workers and --readback-window apply to local runs, not a managed dispatch " + "or a qualification case" +) +REPAIR_PINNED = "--pin-digest pins the original bundle" +REPAIR_QUALIFIED = ( + "a signed qualification admission binds the original bundle; the approved " + "bundle needs its own admission" +) +REPAIR_MANAGED = "a managed dispatch binds the original bundle" + +_REPAIR_EVIDENCE = ( + "tests/test_self_repair.py::test_renamed_button_is_repaired_approved_on_the_phone_and_verified", +) +_STANDARD_EVIDENCE = ( + "tests/test_standard_change.py::test_relabelled_link_is_repaired_without_a_person", +) + +CELLS: tuple[Cell, ...] = ( + # -- API step --------------------------------------------------------------- + Cell( + "api_step", + "default", + "works", + note="the person approves the exact request", + evidence=( + "tests/test_api_step_approval.py::test_the_person_approves_the_exact_request_and_it_is_sent_once", + "tests/test_compatibility_matrix.py::test_api_step_runs_where_the_matrix_says", + ), + ), + Cell( + "api_step", + "durable", + "works", + note="a save that may have landed is never resent", + evidence=( + "tests/test_api_step_durable.py::test_a_durable_run_uses_the_api_step_and_checkpoints_after_it", + "tests/test_api_step_durable.py::test_an_uncertain_send_is_never_resent_by_resume", + ), + ), + Cell( + "api_step", + "governed", + "works", + note="needs `api admit --identity`", + evidence=( + "tests/test_api_step_governed.py::test_the_default_profile_runs_the_admitted_api_step", + ), + ), + Cell( + "api_step", + "pinned", + "works", + note="the pin is checked before the run", + evidence=( + "tests/test_compatibility_matrix.py::test_api_step_runs_where_the_matrix_says", + ), + ), + Cell( + "api_step", + "qualified", + "refused", + reason=API_QUALIFIED, + evidence=( + "tests/test_compatibility_matrix.py::test_api_step_refuses_where_the_matrix_says", + ), + ), + Cell( + "api_step", + "managed", + "refused", + reason=API_MANAGED, + evidence=( + "tests/test_compatibility_matrix.py::test_api_step_refuses_where_the_matrix_says", + ), + ), + # -- pipelined read-back ------------------------------------------------------ + Cell( + "pipelined_readback", + "default", + "works", + note="GUI and API saves", + evidence=( + "tests/test_readback_pipeline.py::test_confirmed_readback_settles_the_saved_report", + "tests/test_api_step_readback_repair.py::test_an_api_save_reads_back_while_the_next_item_runs", + "tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says", + ), + ), + Cell( + "pipelined_readback", + "durable", + "refused", + reason=DURABLE_INLINE, + note="the record check runs before the next item", + evidence=( + "tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says", + ), + ), + Cell( + "pipelined_readback", + "governed", + "refused", + reason=DURABLE_INLINE, + note="Standard and Regulated are durable", + evidence=( + "tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says", + ), + ), + Cell( + "pipelined_readback", + "pinned", + "works", + note="when the run isn't durable", + evidence=( + "tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says", + ), + ), + Cell( + "pipelined_readback", + "qualified", + "refused", + reason=DURABLE_INLINE, + note="Standard and Regulated are durable", + evidence=( + "tests/test_compatibility_matrix.py::test_pipelined_readback_where_the_matrix_says", + ), + ), + Cell( + "pipelined_readback", + "managed", + "refused", + reason=POOL_LOCAL_ONLY, + evidence=( + "tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says", + ), + ), + # -- self-repair --------------------------------------------------------------- + Cell( + "self_repair", + "default", + "works", + evidence=( + "tests/test_self_repair_cli.py::test_run_self_repair_cli_resumes_after_one_phone_tap", + "tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says", + ), + ), + Cell( + "self_repair", + "durable", + "works", + note="self-repair always runs durable", + evidence=_REPAIR_EVIDENCE, + ), + Cell( + "self_repair", + "governed", + "works", + note="Standard contract through the Python API", + evidence=_REPAIR_EVIDENCE, + ), + Cell( + "self_repair", + "pinned", + "refused", + reason=REPAIR_PINNED, + evidence=( + "tests/test_self_repair_cli.py::test_run_self_repair_refuses_a_pinned_bundle", + "tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says", + ), + ), + Cell( + "self_repair", + "qualified", + "refused", + reason=REPAIR_QUALIFIED, + evidence=( + "tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says", + ), + ), + Cell( + "self_repair", + "managed", + "refused", + reason=REPAIR_MANAGED, + evidence=( + "tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says", + ), + ), + # -- standard changes ----------------------------------------------------------- + Cell( + "standard_changes", + "default", + "works", + note="needs a signed repair policy", + evidence=_STANDARD_EVIDENCE, + ), + Cell( + "standard_changes", + "durable", + "works", + note="needs a signed repair policy", + evidence=_STANDARD_EVIDENCE, + ), + Cell( + "standard_changes", + "governed", + "works", + note="needs a signed repair policy", + evidence=_STANDARD_EVIDENCE, + ), + Cell( + "standard_changes", + "pinned", + "refused", + reason=REPAIR_PINNED, + evidence=( + "tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says", + ), + ), + Cell( + "standard_changes", + "qualified", + "refused", + reason=REPAIR_QUALIFIED, + evidence=( + "tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says", + ), + ), + Cell( + "standard_changes", + "managed", + "refused", + reason=REPAIR_MANAGED, + evidence=( + "tests/test_compatibility_matrix.py::test_self_repair_where_the_matrix_says", + ), + ), + # -- parallel workers ----------------------------------------------------------- + Cell( + "parallel_workers", + "default", + "works", + evidence=( + "tests/test_cli_worklist_pool.py::test_two_workers_split_the_items_one_row_each", + "tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says", + ), + ), + Cell( + "parallel_workers", + "durable", + "works", + note="each item has its own durable run", + evidence=( + "tests/test_compatibility_matrix.py::test_workers_run_governed_durable_api_items", + ), + ), + Cell( + "parallel_workers", + "governed", + "works", + note="each item gets its own single-use authorization", + evidence=( + "tests/test_compatibility_matrix.py::test_workers_run_governed_durable_api_items", + ), + ), + Cell( + "parallel_workers", + "pinned", + "works", + evidence=( + "tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says", + ), + ), + Cell( + "parallel_workers", + "qualified", + "works", + note="the admission is checked for every item", + evidence=( + "tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says", + ), + ), + Cell( + "parallel_workers", + "managed", + "refused", + reason=POOL_LOCAL_ONLY, + evidence=( + "tests/test_compatibility_matrix.py::test_the_worklist_pool_where_the_matrix_says", + ), + ), + # -- multi-surface ---------------------------------------------------------------- + Cell( + "multi_surface", + "default", + "works", + note="an API step inside it runs on the GUI (below)", + evidence=( + "tests/test_multi_surface_governed.py::test_governed_durable_multi_surface_run_reaches_verified", + ), + ), + Cell( + "multi_surface", + "durable", + "works", + note="resume re-binds each surface by identity", + evidence=( + "tests/test_multi_surface_identity.py::test_resume_rebinds_surfaces_by_identity", + ), + ), + Cell( + "multi_surface", + "governed", + "works", + evidence=( + "tests/test_multi_surface_governed.py::test_governed_durable_multi_surface_run_reaches_verified", + ), + ), + Cell( + "multi_surface", + "pinned", + "works", + evidence=( + "tests/test_multi_surface_governed.py::test_governed_durable_multi_surface_run_reaches_verified", + ), + ), + Cell( + "multi_surface", + "qualified", + "works", + note="one environment boundary per surface", + evidence=( + "tests/test_multi_surface_governed.py::test_case_evidence_needs_a_matching_observation_per_used_surface", + ), + ), + Cell( + "multi_surface", + "managed", + "works", + note="every backend surface must be bound", + evidence=( + "tests/test_multi_surface_contract.py::test_managed_dispatch_needs_every_backend_surface_bound", + ), + ), +) + +#: The API step combined with the other features, in one run. +API_MULTI_SURFACE = "a run across several windows or tabs uses the GUI steps" +API_REPAIR_PROBE = ( + "self-repair's probes and resumed run exercise the repaired GUI step, and " + "the person approved that step, so its API step doesn't run" +) + +COMBINATIONS: tuple[Cell, ...] = ( + Cell( + "api_step", + "parallel_workers", + "works", + note="each item runs its own API step", + evidence=( + "tests/test_compatibility_matrix.py::test_workers_run_governed_durable_api_items", + ), + ), + Cell( + "api_step", + "pipelined_readback", + "works", + note="when the run isn't durable", + evidence=( + "tests/test_api_step_readback_repair.py::test_an_api_save_reads_back_while_the_next_item_runs", + ), + ), + Cell( + "api_step", + "self_repair", + "refused", + reason=API_REPAIR_PROBE, + note="the first, ordinary run can use the API step", + evidence=( + "tests/test_api_step_readback_repair.py::test_a_probe_or_resumed_run_never_sends_the_api_step", + ), + ), + Cell( + "api_step", + "standard_changes", + "refused", + reason=API_REPAIR_PROBE, + note="the repaired bundle's test-record run is GUI-only too", + evidence=( + "tests/test_api_step_readback_repair.py::test_a_standard_change_test_record_run_is_gui_only", + ), + ), + Cell( + "api_step", + "multi_surface", + "refused", + reason=API_MULTI_SURFACE, + evidence=( + "tests/test_compatibility_matrix.py::test_api_step_refuses_in_a_multi_surface_run", + ), + ), +) + + +def cell(feature: str, profile: str) -> Cell: + """The cell for ``feature`` under ``profile``. Raises ``KeyError``.""" + + for item in CELLS: + if item.feature == feature and item.profile == profile: + return item + raise KeyError((feature, profile)) + + +def _mark(item: Cell) -> str: + if item.status == "works": + text = "Works" + if item.note: + text += f": {item.note}" + return text + text = f"Refused: {item.reason}" + if item.note: + text += f" ({item.note})" + return text + + +def render_markdown() -> str: + """The body of ``docs/COMPATIBILITY.md``, generated from this module.""" + + feature_titles = {f.key: f.title for f in FEATURES} + lines = [ + "# Compatibility: features and run profiles", + "", + "", + "", + "Each feature either works under a run profile or is refused with a", + "named reason. Nothing falls back silently. When a run takes the slower or", + "narrower path, it records why: an API step's `api_fallback`, a", + "pipelined read-back's `inline_reason`, or the command's refusal before", + "anything runs. `tests/test_compatibility_matrix.py` checks every cell", + "against the runtime, and fails when a cell that should work falls back", + "or a refusal gives another reason.", + "", + "## The widest combination today", + "", + "A Regulated, durable, governed run with per-item approval, on parallel", + "workers, using the API step for an admitted save, is the combination", + "where the most work together: each worker's item sends one request a", + "person approved, checks it with the system-of-record read-back, and", + "reads `VERIFIED`. In that run the read-back runs inline (durable), and", + "self-repair runs through the Python API under the Standard contract.", + "Under a signed qualification admission (what `openadapt-flow run`", + "requires for Standard and Regulated actuation) the API step and", + "self-repair are still refused, each with its reason below.", + "", + "## Profiles", + "", + ] + for profile in PROFILES: + lines.append(f"- **{profile.title}.** {profile.description}") + lines += ["", "## Features", ""] + for feature in FEATURES: + lines.append( + f"- **{feature.title}** ([{feature.doc.split('#')[0]}]({feature.doc})). " + f"{feature.description}" + ) + lines += [ + "", + "## At a glance", + "", + "| Feature | " + " | ".join(p.title for p in PROFILES) + " |", + "| --- | " + " | ".join("---" for _ in PROFILES) + " |", + ] + for feature in FEATURES: + marks = [ + "Works" if cell(feature.key, p.key).status == "works" else "Refused" + for p in PROFILES + ] + lines.append(f"| {feature.title} | " + " | ".join(marks) + " |") + lines += [ + "", + "Every refusal names its reason (below), and a run or command that", + "refuses says it in those words.", + ] + lines += ["", "## Features by profile", ""] + for feature in FEATURES: + lines += [f"### {feature.title}", "", "| Profile | Result |", "| --- | --- |"] + for profile in PROFILES: + lines.append( + f"| {profile.title} | {_mark(cell(feature.key, profile.key))} |" + ) + lines.append("") + lines += [ + "## The API step with the other features", + "", + "| With | Result |", + "| --- | --- |", + ] + for item in COMBINATIONS: + lines.append(f"| {feature_titles[item.profile]} | {_mark(item)} |") + lines += [ + "", + "## Evidence", + "", + "Each cell names the tests that show it. They run in CI.", + "", + ] + for item in (*CELLS, *COMBINATIONS): + against = next( + (p.title for p in PROFILES if p.key == item.profile), + feature_titles.get(item.profile, item.profile), + ) + tests = ", ".join(f"`{t}`" for t in item.evidence) + lines.append(f"- {feature_titles[item.feature]} / {against}: {tests}") + lines.append("") + return "\n".join(lines) diff --git a/packages/flow/openadapt_flow/runtime/durable/authority.py b/packages/flow/openadapt_flow/runtime/durable/authority.py index 03e3fe649..c5183fb44 100644 --- a/packages/flow/openadapt_flow/runtime/durable/authority.py +++ b/packages/flow/openadapt_flow/runtime/durable/authority.py @@ -655,6 +655,25 @@ def _connect(self) -> sqlite3.Connection: raise DurableAuthorityBusy( "the external durable authority database permissions are too broad" ) + elif os.name != "nt": + # Create the database owner-only before SQLite opens it. SQLite + # would create it with the process umask and the chmod below comes + # later, so a second durable run starting at the same moment (two + # workers, two processes) saw a world-readable file and refused. + try: + descriptor = os.open( + self.db_path, + os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_NOFOLLOW", 0), + 0o600, + ) + except FileExistsError: + pass + except OSError as exc: + raise DurableAuthorityBusy( + "the external durable authority database can't be created" + ) from exc + else: + os.close(descriptor) connection: Optional[sqlite3.Connection] = None try: connection = sqlite3.connect( diff --git a/packages/flow/scripts/build_compatibility_doc.py b/packages/flow/scripts/build_compatibility_doc.py new file mode 100644 index 000000000..c25e9618d --- /dev/null +++ b/packages/flow/scripts/build_compatibility_doc.py @@ -0,0 +1,36 @@ +"""Write docs/COMPATIBILITY.md from openadapt_flow/compatibility.py. + +Usage (from packages/flow): python scripts/build_compatibility_doc.py +``--check`` exits 1 when the file is out of date instead of writing it. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) + +from openadapt_flow.compatibility import render_markdown # noqa: E402 + +TARGET = ROOT / "docs" / "COMPATIBILITY.md" + + +def main(argv: list[str]) -> int: + text = render_markdown() + if "--check" in argv: + current = TARGET.read_text(encoding="utf-8") if TARGET.exists() else "" + if current != text: + print( + f"{TARGET} is out of date: run python scripts/build_compatibility_doc.py" + ) + return 1 + return 0 + TARGET.write_text(text, encoding="utf-8") + print(f"wrote {TARGET}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/packages/flow/tests/test_compatibility_matrix.py b/packages/flow/tests/test_compatibility_matrix.py new file mode 100644 index 000000000..491ae261b --- /dev/null +++ b/packages/flow/tests/test_compatibility_matrix.py @@ -0,0 +1,386 @@ +"""The compatibility matrix (openadapt_flow/compatibility.py, docs/COMPATIBILITY.md). + +Every feature against every run profile either works or is refused with a +named reason. These tests check each cell against the runtime: + +- a cell that works must engage the feature (the API step sends its request, + the read-back runs on the side thread); a fallback fails the test; +- a refused cell must give exactly the reason the matrix names; +- every cell names tests that exist, and the published page is current. + +The record system is a scripted verifier or the MockMed fault server; the +values are made up. +""" + +from __future__ import annotations + +import argparse +import ast +from pathlib import Path +from types import SimpleNamespace +from typing import Any, Optional + +import pytest + +from openadapt_flow import compatibility as matrix +from openadapt_flow.approval_mode.gate import ApprovalGate +from openadapt_flow.approval_mode.item_approval import ItemApprovalStore +from openadapt_flow.deployment import DeploymentConfig, PolicySection +from openadapt_flow.execution_profiles import ExecutionProfile +from openadapt_flow.run_gate import build_runtime_authorization, evaluate_run_gate +from openadapt_flow.runtime.actuators.web_form import WebFormActuator +from openadapt_flow.runtime.effects.rest import RestRecordVerifier +from openadapt_flow.runtime.readback_pipeline import ReadbackPipeline +from tests.test_api_step_approval import APPROVE, Person +from tests.test_api_step_governed import PARAMS, Transport, _workflow +from tests.test_approval_mode_replayer import RecordSystem +from tests.test_execution_profiles import ( + _KEY, + Replayer, + _gate, + _ReadyVision, + _sealed, +) +from tests.test_replayer import FakeBackend + +FLOW = Path(__file__).resolve().parents[1] +REPO = FLOW.parents[1] + + +class RestRecordSystem(RecordSystem, RestRecordVerifier): + """The scripted record system, read the way a REST read is. + + The read-back pipeline defers only network reads of the system of record + (REST, FHIR, GraphQL, SQL); this is the scripted system as one of them. + """ + + +class Run: + """One item of the API-step program under one run profile.""" + + def __init__( + self, + tmp_path: Path, + profile: str, + *, + pipeline: Optional[ReadbackPipeline] = None, + ) -> None: + self.profile = profile + self.durable = profile in {"durable", "governed", "qualified"} + regulated = profile in {"governed", "qualified"} + self.workflow, self.bundle = _sealed(tmp_path, _workflow(), encrypted=True) + self.verifier = RestRecordSystem() + if regulated: + report = _gate( + self.workflow, + self.bundle, + ExecutionProfile.REGULATED, + verifier=self.verifier, + durable=True, + ) + else: + report = evaluate_run_gate( + self.workflow, + bundle_dir=self.bundle, + deployment=DeploymentConfig(policy=PolicySection(policy="permissive")), + effect_verifier=self.verifier, + effective_durable=self.durable, + ) + assert report.passed, report.render() + self.authorization = build_runtime_authorization( + self.workflow, report, params=PARAMS + ) + if profile == "qualified": + # The fields a signed production admission binds into the run. + self.authorization = self.authorization.model_copy( + update={ + "production_qualification_admission_id": "admission-1", + "production_qualification_admission_sha256": "1" * 64, + "production_qualification_evidence_identity_sha256": "2" * 64, + "production_qualification_runtime_validation_id": "validation-1", + "production_qualification_signer_registry_sha256": "3" * 64, + "production_qualification_signer_registry_revision": 1, + "production_qualification_signer_registry_expires_at": ( + "2099-01-01T00:00:00+00:00" + ), + "production_qualification_authority_sha256": "4" * 64, + } + ) + self.run_dir = tmp_path / "run" + self.person = Person([APPROVE]) + self.store = ItemApprovalStore(self.run_dir) + self.person.store = self.store + self.transport = Transport() + self.pipeline = pipeline + + def replayer(self) -> Any: + gate = ApprovalGate( + "each_item", + approval_store=self.store, + approval_wait_s=30.0, + approval_poll_s=1.0, + sleep=self.person.sleep, + monotonic=self.person.monotonic, + ) + return Replayer( + FakeBackend(), + vision=_ReadyVision(), + effect_verifier=self.verifier, + governed_authorization=self.authorization, + durable=self.durable, + checkpoint_key=_KEY, + require_settled=True, + approval_gate=gate, + web_form_actuator=WebFormActuator(self.transport), + readback_pipeline=self.pipeline, + ) + + def run(self) -> Any: + return self.replayer().run( + self.workflow, + params=PARAMS, + bundle_dir=self.bundle, + run_dir=self.run_dir, + ) + + +def _cells(feature: str, status: str) -> list[str]: + return [ + c.profile for c in matrix.CELLS if c.feature == feature and c.status == status + ] + + +# -- the matrix is complete and backed by tests ------------------------------------- + + +def test_every_feature_has_exactly_one_cell_per_profile(): + pairs = [(c.feature, c.profile) for c in matrix.CELLS] + assert len(pairs) == len(set(pairs)) + assert set(pairs) == { + (f.key, p.key) for f in matrix.FEATURES for p in matrix.PROFILES + } + features = {f.key for f in matrix.FEATURES} + for item in matrix.COMBINATIONS: + assert item.feature in features and item.profile in features + + +@pytest.mark.parametrize( + "item", + [*matrix.CELLS, *matrix.COMBINATIONS], + ids=lambda c: f"{c.feature}-{c.profile}", +) +def test_every_cell_names_a_reason_or_works_and_has_evidence(item): + if item.status == "refused": + assert item.reason, "a refused cell names its reason" + else: + assert item.reason is None + assert item.evidence, "every cell names the tests that show it" + for node in item.evidence: + path, _, name = node.partition("::") + source = FLOW / path + assert source.is_file(), node + names = { + n.name + for n in ast.walk(ast.parse(source.read_text())) + if isinstance(n, ast.FunctionDef) + } + assert name in names, node + + +def test_the_published_page_is_current_and_linked(): + page = FLOW / "docs" / "COMPATIBILITY.md" + assert page.read_text(encoding="utf-8") == matrix.render_markdown(), ( + "run python scripts/build_compatibility_doc.py" + ) + assert ( + "packages/flow/docs/COMPATIBILITY.md" in (REPO / "CAPABILITIES.md").read_text() + ) + + +# -- the API step --------------------------------------------------------------------- + + +@pytest.mark.parametrize("profile", _cells("api_step", "works")) +def test_api_step_runs_where_the_matrix_says(tmp_path, profile): + run = Run(tmp_path, profile) + report = run.run() + + assert [r.actuation for r in report.results] == ["api_folded", "api"], [ + r.api_fallback for r in report.results + ] + assert all(r.api_fallback is None for r in report.results) + assert [path for path, _ in run.transport.posts] == ["/save.php"] + assert report.success is True + (request,) = run.store.requests() + assert request.form_state_kind == "api_request" + if profile == "governed": + assert report.execution_outcome == "VERIFIED" + + +@pytest.mark.parametrize("profile", _cells("api_step", "refused")) +def test_api_step_refuses_where_the_matrix_says(tmp_path, profile): + reason = matrix.cell("api_step", profile).reason + run = Run(tmp_path, "qualified" if profile == "qualified" else "default") + if profile == "managed": + replayer = run.replayer() + replayer.managed_dispatch_binding = object() + _end, _save, why = replayer._web_form_span_at(run.workflow, 0, durable=True) + assert why == reason + return + report = run.run() + assert run.transport.posts == [] + assert report.results[0].api_fallback == reason + + +def test_api_step_refuses_in_a_multi_surface_run(tmp_path): + run = Run(tmp_path, "default") + replayer = run.replayer() + replayer._surface_router = object() + _end, _save, why = replayer._web_form_span_at(run.workflow, 0, durable=False) + assert why == matrix.API_MULTI_SURFACE + + +def test_the_repair_reason_is_the_runtimes(): + from openadapt_flow.repair.self_repair import REPAIR_GUI_ONLY + + assert matrix.API_REPAIR_PROBE == REPAIR_GUI_ONLY + + +# -- pipelined read-back --------------------------------------------------------------- + + +@pytest.mark.parametrize( + "profile", [p.key for p in matrix.PROFILES if p.key != "managed"] +) +def test_pipelined_readback_where_the_matrix_says(tmp_path, profile): + item = matrix.cell("pipelined_readback", profile) + pipeline = ReadbackPipeline() + try: + run = Run(tmp_path, profile, pipeline=pipeline) + report = run.run() + handoff = pipeline.take() + settled = handoff.settled.result(timeout=30) if handoff is not None else None + finally: + pipeline.close() + + assert report.results[-1].effect_verified is True or settled is not None + if item.status == "works": + assert settled is report, pipeline.inline_reason + assert report.success is True + assert report.results[-1].effect_verified is True + else: + assert settled is None + assert pipeline.inline_reason == item.reason + + +# -- the worklist pool ------------------------------------------------------------------- + + +@pytest.mark.parametrize("profile", [p.key for p in matrix.PROFILES]) +def test_the_worklist_pool_where_the_matrix_says(tmp_path, profile): + from openadapt_flow.cli_worklist_pool import pool_profile_refusal + from openadapt_flow.runtime.worklist_pool import ( + FunctionWorker, + PoolPolicy, + run_worklist_pool, + ) + + refusal = pool_profile_refusal( + managed=profile == "managed", qualification_case=False + ) + workers = matrix.cell("parallel_workers", profile) + if workers.status == "refused": + assert refusal == workers.reason + assert matrix.cell("pipelined_readback", profile).reason == refusal + return + assert refusal is None + ran: list[int] = [] + pool = [ + FunctionWorker(f"w{i}", lambda idx, row, d: ran.append(idx) or _Done()) + for i in range(2) + ] + report = run_worklist_pool( + [{"n": str(i)} for i in range(4)], + pool, + run_dir=tmp_path, + policy=PoolPolicy(workers=2, items_independent=True), + ) + assert sorted(ran) == [0, 1, 2, 3] and report.success + + +class _Done: + success = True + execution_outcome = "VERIFIED" + transaction_outcome = "VERIFIED" + halt = None + total_ms = 1.0 + + +def test_workers_run_governed_durable_api_items(tmp_path): + """Two workers, each item a Regulated, durable, gated API-step run.""" + + from openadapt_flow.runtime.worklist_pool import ( + VERIFIED, + FunctionWorker, + PoolPolicy, + run_worklist_pool, + ) + + runs: dict[int, Run] = {} + + def item(idx: int, row: dict[str, Any], item_dir: Path) -> Any: + runs[idx] = Run(item_dir, "governed") + return runs[idx].run() + + report = run_worklist_pool( + [{"n": str(i)} for i in range(4)], + [FunctionWorker(f"w{i}", item) for i in range(2)], + run_dir=tmp_path, + policy=PoolPolicy(workers=2, items_independent=True), + ) + assert {i["state"] for i in report.items} == {VERIFIED}, [ + i.get("detail") or i for i in report.items + ] + for run in runs.values(): + assert [path for path, _ in run.transport.posts] == ["/save.php"] + + +# -- self-repair and standard changes ---------------------------------------------------- + + +def _repair_args(profile: str) -> argparse.Namespace: + return argparse.Namespace( + managed_dispatch_file="dispatch.json" if profile == "managed" else None, + qualification_authority_file=( + "authority.json" if profile == "qualified" else None + ), + _qualification_case_execution=None, + pin_digest="a" * 64 if profile == "pinned" else None, + hold_on_halt=False, + save_healed_to=None, + record_video=False, + repair_agent="mod:factory", + repair_screen_access_by="reviewer", + durable=profile == "durable", + ) + + +@pytest.mark.parametrize("feature", ["self_repair", "standard_changes"]) +@pytest.mark.parametrize("profile", [p.key for p in matrix.PROFILES]) +def test_self_repair_where_the_matrix_says(feature, profile): + from openadapt_flow.cli_self_repair import _refusal + + backend = SimpleNamespace( + kind="web", browser_cdp_endpoint=None, browser_page_url=None + ) + refusal = _refusal(_repair_args(profile), backend, "http://app") + item = matrix.cell(feature, profile) + if item.status == "refused": + assert refusal == item.reason + elif profile == "governed": + # The CLI's Standard profile needs a signed admission (the + # "qualified" column); the Standard contract runs through the + # Python API, which these cells' evidence tests exercise. + assert item.evidence + else: + assert refusal is None diff --git a/packages/flow/tests/test_durable_authority_create.py b/packages/flow/tests/test_durable_authority_create.py new file mode 100644 index 000000000..70588a9ee --- /dev/null +++ b/packages/flow/tests/test_durable_authority_create.py @@ -0,0 +1,69 @@ +"""The external durable authority database is created owner-only. + +Two durable runs that start at the same moment (two workers, or two +processes) both open the authority database. SQLite creates a missing file +with the process umask; if the first run's later chmod hadn't happened yet, +the second run saw a world-readable file and refused. The file now exists, +owner-only, before SQLite opens it. +""" + +from __future__ import annotations + +import os +import sqlite3 +import stat +import threading + +import pytest + +from openadapt_flow.runtime.durable import authority as authority_module +from tests.test_durable_authority_v13 import _fresh_run + + +def _store(path): + """Start one fresh durable run (its own run id and namespace).""" + + import uuid + + return _fresh_run(path, run_id=str(uuid.uuid4()), namespace_id=uuid.uuid4().hex) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_a_fresh_database_is_owner_only_before_sqlite_opens_it(tmp_path, monkeypatch): + db_path = tmp_path / "authority" / "authority.sqlite3" + monkeypatch.setenv(authority_module.AUTHORITY_DB_ENV, str(db_path)) + seen: list[int] = [] + real_connect = sqlite3.connect + + def connect(*args, **kwargs): + if not seen: + seen.append(stat.S_IMODE(os.lstat(db_path).st_mode)) + return real_connect(*args, **kwargs) + + monkeypatch.setattr(authority_module.sqlite3, "connect", connect) + os.umask(0o022) + _store(tmp_path / "a") + assert seen and seen[0] & 0o077 == 0, oct(seen[0]) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX permission bits") +def test_two_durable_runs_starting_together_both_start(tmp_path, monkeypatch): + db_path = tmp_path / "authority" / "authority.sqlite3" + monkeypatch.setenv(authority_module.AUTHORITY_DB_ENV, str(db_path)) + errors: list[BaseException] = [] + barrier = threading.Barrier(4) + + def start(name: str) -> None: + try: + barrier.wait() + _store(tmp_path / name) + except BaseException as exc: # noqa: BLE001 - collected for the assert + errors.append(exc) + + threads = [threading.Thread(target=start, args=(f"run-{i}",)) for i in range(4)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + assert errors == [] + assert stat.S_IMODE(os.lstat(db_path).st_mode) & 0o077 == 0