diff --git a/docs/BYOC_CONNECTOR.md b/docs/BYOC_CONNECTOR.md index c4af427f..5e042e8f 100644 --- a/docs/BYOC_CONNECTOR.md +++ b/docs/BYOC_CONNECTOR.md @@ -67,7 +67,7 @@ Cloud token creation is authenticated and organization-scoped in the dashboard. bundle binding. * **Up (to the control plane):** PHI-free status/metrics + a storage *path* into the customer's own store. **Never** the report body, screenshots, OCR text, or - a patient identifier. + a record identifier. * **The bundle and report bytes** are read from / written to the **customer's own storage** (`--storage-root`, a local encrypted volume). Our control plane holds no URL to them and signs no access. diff --git a/docs/DECISION_DELIVERY.md b/docs/DECISION_DELIVERY.md index a5db0d4d..abd0628e 100644 --- a/docs/DECISION_DELIVERY.md +++ b/docs/DECISION_DELIVERY.md @@ -54,14 +54,48 @@ evidence. a vocabulary the engine already owns — `Rung`, `ActionKind`, the console's halt categories, the ARIA/UIA role names, `RecheckKind` — or is a bounded integer or a boolean. A relay that stores this object is *structurally incapable* of -representing a patient name, an MRN, an observed value, a path, or a workflow -label. +representing a person name, a record identifier, an observed value, a path, or +a workflow label. That is the same kind of guarantee `HumanDecisionTaskV1` already gives, and it is checkable the same way: by reading a schema, not by trusting a detector. The hosted control plane enforces it a third time in Postgres, in the same style as `human_decision_task_contract_valid`. +## V2: qualification-approved entity wording + +V1 uses only domain-neutral wording such as `record` or `item`. A negotiated V2 +task can carry one useful entity class that the exact qualification contract +already approved. The remote emitter accepts only the reviewed remote-safe +vocabulary. +The canonical vocabulary and cross-vertical examples are in +[Attended decisions and the halt-learn loop](https://docs.openadapt.ai/concepts/halt-learn-loop/#a-qualified-entity-label-not-a-guessed-domain). + +The entity class is optional. A person or a qualification agent can set it once +for a workflow version. A reviewed class such as `insurance claim` can cross the +remote boundary. A custom local class remains inside the bundle; the V2 task +carries only its signed neutral `record` or `item` fallback. With no label, the +task uses the signed neutral `record` class. This presentation choice does not +block qualification, certification, or execution. A label change is a contract +change and requires recertification before the new V2 task can be emitted. + +This is not runtime inference. The producer reads the label from the exact +qualified step and binds the task to the qualification project, qualification +revision, qualification contract digest, and bundle digest. The task has no +field for a screenshot, OCR output, parameter, application name, observed +identity value, or model input. A label says what class of entity the workflow +handles; it never says which entity is on screen. + +The runner and client use V2 only after explicit schema negotiation. A peer that +does not negotiate V2 receives the byte-compatible V1 task and renders `record` +or `item`. Before any action continues, the customer-controlled runner reads +the live application and revalidates the required identity and effect +contracts. + +The V2 producer uses the released `openadapt-types` 0.9 contract. A consumer +must explicitly negotiate that schema. Otherwise, Flow emits the byte-compatible +V1 task. + The one thing this tier gives up is `target_label` — the target control's own accessible name, which `halt_detail._safe_target_label` releases locally after six independent proofs. It stays local. The phone therefore says *"OpenAdapt diff --git a/docs/ECOSYSTEM_INTEGRATION.md b/docs/ECOSYSTEM_INTEGRATION.md index 0d03f56f..83ddab80 100644 --- a/docs/ECOSYSTEM_INTEGRATION.md +++ b/docs/ECOSYSTEM_INTEGRATION.md @@ -121,7 +121,7 @@ The string values are byte-identical, so a flow `Step` could emit/ingest an - `Resolution` — which ladder rung resolved the target (`template/…/grounder`), point, confidence, `elapsed_ms`. - `IdentityCheck` — the pre-click same-entity verdict (`verified/mismatch/abstain/ - unreadable` × `structured/pixel/vlm/context/param`). This is the wrong-patient safety + unreadable` × `structured/pixel/vlm/context/param`). This is the wrong-entity safety core. - `HealEvent`, `StepResult`, `RunReport`, `UnarmedStep` — audit/telemetry. - `risk∈{reversible,irreversible}`, `identity_armed` — the halt-on-uncertainty gates. diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index ce5e85e4..7ad0b5df 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -2464,17 +2464,22 @@ def _cmd_qualify(args: argparse.Namespace) -> int: QualificationCaseKind, QualificationCaseResult, QualificationOutcome, + QualifiedEntityLabel, RequalificationCondition, VerificationTier, add_case, add_requalification_condition, certify_project, + entity_label_options, init_project, + list_entity_labels, project_schema, record_case_results, + remove_entity_label, save_qualified_workflow, set_action_classification, set_effect_policy, + set_entity_label, set_identity_policy, set_trusted_runner_key, ) @@ -2488,6 +2493,62 @@ def _cmd_qualify(args: argparse.Namespace) -> int: workflow = _qualification_workflow(args) + if verb == "label": + if args.label_cmd == "list": + print( + json.dumps( + [ + label.model_dump(mode="json") + for label in list_entity_labels(workflow) + ], + indent=2, + ) + ) + return 0 + changed = False + if args.label_cmd == "set": + reviewed_fallback = next( + ( + option["fallback"] + for option in entity_label_options() + if option["label"] == args.label + ), + None, + ) + if ( + reviewed_fallback is not None + and args.fallback is not None + and args.fallback != reviewed_fallback + ): + raise SystemExit( + f"{args.label!r} uses the reviewed fallback {reviewed_fallback!r}" + ) + fallback = reviewed_fallback or args.fallback or "record" + try: + label = QualifiedEntityLabel( + step_id=args.step, label=args.label, fallback=fallback + ) + except ValueError as exc: + raise SystemExit(f"invalid entity label: {exc}") from exc + assert workflow.qualification is not None + changed = workflow.qualification.entity_labels.get(args.step) != label + set_entity_label(workflow, label) + elif args.label_cmd == "remove": + assert workflow.qualification is not None + changed = args.step in workflow.qualification.entity_labels + remove_entity_label(workflow, args.step) + else: # pragma: no cover - argparse requires a known command. + raise SystemExit(f"unknown qualification label command {args.label_cmd!r}") + save_qualified_workflow(workflow, args.bundle) + print(workflow.qualification.model_dump_json(indent=2)) + if changed: + print( + "Certification invalidated. Run `openadapt-flow qualify certify " + " --evidence-root ` before production V2 tasks.", + file=sys.stderr, + ) + return 0 + if verb == "init": environment = EnvironmentBoundary( target_kind=args.target, @@ -2524,6 +2585,7 @@ def _cmd_qualify(args: argparse.Namespace) -> int: else None ), "report": report.model_dump(mode="json"), + "entity_label_options": entity_label_options(), } print(json.dumps(payload, indent=2, sort_keys=True)) else: @@ -4300,6 +4362,43 @@ def build_parser() -> argparse.ArgumentParser: q = qsub.add_parser("schema", help="Print the qualification-project JSON Schema") q.set_defaults(func=_cmd_qualify) + q = qsub.add_parser( + "label", + help="Optionally set, remove, or list presentation-only entity labels", + ) + label_sub = q.add_subparsers(dest="label_cmd", required=True) + label = label_sub.add_parser( + "set", + help="Set an optional presentation label for one qualified step", + ) + label.add_argument("bundle", help="Workflow bundle directory") + label.add_argument("--step", required=True, help="Exact qualified workflow step ID") + label.add_argument( + "--label", + required=True, + help=( + "Local class label, for example: insurance claim. Reviewed labels " + "can cross a remote boundary; other labels use the neutral fallback." + ), + ) + label.add_argument( + "--fallback", + choices=("record", "item"), + default=None, + help=( + "Neutral remote label for a custom class (default: record). " + "Reviewed labels use their canonical fallback." + ), + ) + label.set_defaults(func=_cmd_qualify) + label = label_sub.add_parser("remove", help="Remove a step entity label") + label.add_argument("bundle", help="Workflow bundle directory") + label.add_argument("--step", required=True, help="Exact qualified workflow step ID") + label.set_defaults(func=_cmd_qualify) + label = label_sub.add_parser("list", help="List qualification-owned entity labels") + label.add_argument("bundle", help="Workflow bundle directory") + label.set_defaults(func=_cmd_qualify) + q = qsub.add_parser("init", help="Initialize a bundle's qualification project") q.add_argument("bundle", help="Workflow bundle directory") q.add_argument( diff --git a/openadapt_flow/console/human_decisions.py b/openadapt_flow/console/human_decisions.py index 1e10884e..caaf8739 100644 --- a/openadapt_flow/console/human_decisions.py +++ b/openadapt_flow/console/human_decisions.py @@ -33,8 +33,10 @@ from openadapt_types import ( HUMAN_DECISION_TASK_SCHEMA, + HUMAN_DECISION_TASK_V2_SCHEMA, HumanDecisionReceiptV1, HumanDecisionTaskV1, + HumanDecisionTaskV2, ) from pydantic import BaseModel, ConfigDict, Field @@ -232,7 +234,7 @@ class RemoteDecisionProjection(BaseModel): schema_version: Literal["openadapt.remote-decision-projection/v1"] = ( "openadapt.remote-decision-projection/v1" ) - task: HumanDecisionTaskV1 + task: HumanDecisionTaskV1 | HumanDecisionTaskV2 task_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$") phase: Literal["paused"] = "paused" event_sequence: int = Field(ge=1) @@ -389,6 +391,98 @@ def _remote_delivery_tier( raise AttendedActionRefused(str(exc)) from exc +def _qualified_entity_v2_fields( + run_dir: Path, + *, + step_id: Optional[str], + capability: Any, +) -> Optional[dict[str, Any]]: + """Read V2 bindings only from the current integrity-sealed project. + + This deliberately reads no screenshots, OCR, accessibility observation, + parameters, application name, or model output. A missing optional domain + label uses the reviewed neutral ``record`` class without weakening the V2 + qualification and step bindings. Any missing, unreadable, or mismatched + capability/bundle binding falls back to V1. This gate requires both bundle + integrity and current qualification certification. + """ + + if step_id is None or step_id != capability.step_id: + return None + try: + from openadapt_flow.runtime.durable.checkpoint import CheckpointStore + from openadapt_flow.runtime.durable.program_checkpoint import bundle_version + + manifest = CheckpointStore(run_dir).read_manifest() + if manifest is None: + return None + if bundle_version(manifest.bundle_dir) != capability.bundle_version: + return None + workflow, _ = data.load_workflow_safe(Path(manifest.bundle_dir)) + project = workflow.qualification if workflow is not None else None + certification = project.last_certification if project is not None else None + if ( + workflow is None + or project is None + or certification is None + or not certification.passed + ): + return None + from openadapt_flow.qualification import ( + current_certification_matches, + remote_entity_label, + workflow_contract_sha256, + ) + + # Certification is the existing authority for qualification approval. + # Check its direct bindings before asking it to independently recompute + # the signed-case and policy decision from this exact sealed workflow. + if ( + certification.project_revision != project.revision + or certification.project_contract_sha256 != project.contract_sha256() + or certification.workflow_contract_sha256 + != workflow_contract_sha256(workflow) + or certification.policy_contract_sha256 is None + or not current_certification_matches( + workflow, + policy_contract_digest=certification.policy_contract_sha256, + ) + ): + return None + entity = project.entity_labels.get(step_id) + entity_payload = remote_entity_label(entity) + return { + "qualification_project_id": project.project_id, + "qualification_revision_id": f"qualification_revision_{project.revision}", + "qualification_contract_digest": "sha256:" + project.contract_sha256(), + "qualification_step_id": step_id, + "entity": entity_payload, + } + except (OSError, ValueError, TypeError, AttendedActionRefused): + return None + + +def _peer_negotiated_v2(deployment: Optional[DeploymentConfig]) -> bool: + """Return true only for an explicit authenticated-peer V2 declaration.""" + + return bool( + deployment is not None + and deployment.human_decisions.remote.enabled + and HUMAN_DECISION_TASK_V2_SCHEMA + in deployment.human_decisions.remote.peer_task_schemas + ) + + +def _task_model( + task: dict[str, Any], +) -> type[HumanDecisionTaskV1 | HumanDecisionTaskV2]: + return ( + HumanDecisionTaskV2 + if task.get("schema_version") == HUMAN_DECISION_TASK_V2_SCHEMA + else HumanDecisionTaskV1 + ) + + def _task_and_presentation( run_dir: Path, item: AttentionItem, @@ -540,8 +634,22 @@ def _task_and_presentation( ), "signature_algorithm": "hmac-sha256", } - task = AttendedActionStore(run_dir).seal_human_decision_task(unsigned) - task_digest = HumanDecisionTaskV1.model_validate(task).digest + qualified_v2 = ( + _qualified_entity_v2_fields( + run_dir, + step_id=failed.step_id if failed is not None else None, + capability=capability, + ) + if _peer_negotiated_v2(deployment) + else None + ) + if qualified_v2 is not None: + unsigned.update(qualified_v2) + unsigned["schema_version"] = HUMAN_DECISION_TASK_V2_SCHEMA + task = AttendedActionStore(run_dir).seal_human_decision_task_v2(unsigned) + else: + task = AttendedActionStore(run_dir).seal_human_decision_task(unsigned) + task_digest = _task_model(task).model_validate(task).digest return task, task_digest, presentation @@ -605,7 +713,7 @@ def portable_remote_decision_task( raise AttendedActionRefused( "the run has no current signed human decision task or pause capability" ) - task = HumanDecisionTaskV1.model_validate(task_raw) + task = _task_model(task_raw).model_validate(task_raw) # The rest of `presentation` -- the screenshot artifact ids, the composed # question, the gated control label inside `halt` -- is still discarded. # Only the closed-vocabulary re-projection of `halt` may cross, and only at diff --git a/openadapt_flow/deployment.py b/openadapt_flow/deployment.py index 91c4ee2e..8051ed3b 100644 --- a/openadapt_flow/deployment.py +++ b/openadapt_flow/deployment.py @@ -446,6 +446,14 @@ class RemoteHumanDecisionConfig(BaseModel): context_tier: Literal["remote_closed_context", "remote_identifiers"] = ( "remote_closed_context" ) + #: Exact schemas the authenticated remote peer advertised for this + #: deployment. Empty means no V2 negotiation occurred, so Flow emits V1. + peer_task_schemas: list[ + Literal[ + "openadapt.human-decision-task/v1", + "openadapt.human-decision-task/v2", + ] + ] = Field(default_factory=list) @model_validator(mode="after") def _require_exact_remote_scope(self) -> "RemoteHumanDecisionConfig": @@ -454,6 +462,8 @@ def _require_exact_remote_scope(self) -> "RemoteHumanDecisionConfig": "human_decisions.remote.enabled requires exact tenant_id and " "runner_id bindings" ) + if len(self.peer_task_schemas) != len(set(self.peer_task_schemas)): + raise ValueError("human decision peer task schemas must be unique") return self diff --git a/openadapt_flow/qualification.py b/openadapt_flow/qualification.py index 2a8c4c5e..0eae7faa 100644 --- a/openadapt_flow/qualification.py +++ b/openadapt_flow/qualification.py @@ -57,6 +57,38 @@ _ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") _PARAM_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,127}$") _CONTEXT_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/-]{0,127}$") +_QUALIFIED_ENTITY_LABEL_RE = re.compile(r"^[a-z][a-z0-9]*(?:[ _-][a-z0-9]+){0,3}$") + + +def entity_label_options() -> list[dict[str, str]]: + """Return the reviewed remote-safe class labels and their fallbacks. + + This is the public, JSON-serializable source for Desktop and Flow. The + values are presentation classes only; no identity value or identifier may + enter this API. + """ + + return [ + {"label": "patient record", "fallback": "record"}, + {"label": "member record", "fallback": "record"}, + {"label": "insurance claim", "fallback": "item"}, + {"label": "loan application", "fallback": "item"}, + {"label": "customer account", "fallback": "record"}, + {"label": "service request", "fallback": "item"}, + {"label": "case", "fallback": "item"}, + {"label": "order", "fallback": "item"}, + {"label": "invoice", "fallback": "item"}, + {"label": "document", "fallback": "item"}, + {"label": "record", "fallback": "record"}, + {"label": "item", "fallback": "item"}, + ] + + +#: Derived compatibility view of :func:`entity_label_options` for callers that +#: need only labels. Do not add an independent list of remote-safe classes. +REMOTE_SAFE_ENTITY_LABELS: Final[tuple[str, ...]] = tuple( + option["label"] for option in entity_label_options() +) def _qualification_identifier_sha256(value: str, *, kind: str) -> str: @@ -730,6 +762,69 @@ def _default_fault_cases() -> list[QualificationCase]: ] +class QualifiedEntityLabel(BaseModel): + """An optional local presentation label for one qualified step. + + Qualification and execution do not require this label. Reviewed labels can + cross a remote boundary. A custom label remains local and remote consumers + receive its signed neutral ``record`` or ``item`` fallback. + """ + + model_config = ConfigDict(extra="forbid", frozen=True) + + step_id: str = Field(pattern=_ID_RE.pattern) + label: str = Field( + min_length=1, + max_length=63, + json_schema_extra={ + "examples": ["patient record", "insurance claim", "record"], + "x-openadapt-reviewed-remote-options": list(REMOTE_SAFE_ENTITY_LABELS), + }, + ) + fallback: Literal["record", "item"] + + @field_validator("label") + @classmethod + def _safe_label(cls, value: str) -> str: + if _QUALIFIED_ENTITY_LABEL_RE.fullmatch(value) is None: + raise ValueError( + "entity label must be a lowercase class name of at most four words" + ) + return value + + @model_validator(mode="after") + def _canonical_fallback(self) -> "QualifiedEntityLabel": + expected = next( + ( + option["fallback"] + for option in entity_label_options() + if option["label"] == self.label + ), + None, + ) + if expected is not None and self.fallback != expected: + raise ValueError("entity fallback must match the reviewed class mapping") + return self + + +def remote_entity_label( + entity: Optional[QualifiedEntityLabel], +) -> dict[str, str]: + """Return the reviewed entity label that a remote V2 task can carry. + + Custom qualification labels remain inside the local bundle. Their signed + neutral fallback crosses the remote boundary instead. A missing optional + label uses ``record`` without weakening any other V2 binding. + """ + + if entity is None: + return {"label": "record", "fallback": "record"} + payload = entity.model_dump(mode="json", exclude={"step_id"}) + if payload in entity_label_options(): + return payload + return {"label": entity.fallback, "fallback": entity.fallback} + + class QualificationProject(BaseModel): """Versioned qualification configuration sealed inside a Flow bundle.""" @@ -752,6 +847,10 @@ class QualificationProject(BaseModel): ) identity_policies: dict[str, IdentityPolicy] = Field(default_factory=dict) effect_policies: list[EffectVerificationPolicy] = Field(default_factory=list) + #: Optional presentation-only names selected during qualification. They are + #: not safety evidence, observations, or runtime-derived values. An empty + #: mapping is valid and does not affect qualification admission. + entity_labels: dict[str, "QualifiedEntityLabel"] = Field(default_factory=dict) cases: list[QualificationCase] = Field(default_factory=_default_fault_cases) exclusions: list[str] = Field(default_factory=list) requalification_conditions: list[RequalificationCondition] = Field( @@ -775,6 +874,12 @@ def _consistent_keys(self) -> "QualificationProject": f"action classification key {key!r} does not match step_id " f"{classification.step_id!r}" ) + for key, entity in self.entity_labels.items(): + if key != entity.step_id: + raise ValueError( + f"entity label key {key!r} does not match step_id " + f"{entity.step_id!r}" + ) for key_kind, keys in ( ("runner", self.trusted_runner_keys), ("fault driver", self.trusted_fault_driver_keys), @@ -1290,6 +1395,55 @@ def set_minimum_effect_tier( return project +def set_entity_label( + workflow: "Workflow", label: QualifiedEntityLabel +) -> QualificationProject: + """Set one qualification-owned entity label and invalidate certification.""" + + project = workflow.qualification + if project is None: + raise QualificationError( + "initialize qualification before setting an entity label" + ) + if label.step_id not in _steps_by_id(workflow): + raise QualificationError(f"unknown step id {label.step_id!r}") + if project.entity_labels.get(label.step_id) == label: + return project + previous = project.revision_digest() + project.entity_labels[label.step_id] = label + _touch(project, previous) + _invalidate_certification(workflow) + return project + + +def remove_entity_label(workflow: "Workflow", step_id: str) -> QualificationProject: + """Remove one qualification-owned entity label and invalidate certification.""" + + project = workflow.qualification + if project is None: + raise QualificationError( + "initialize qualification before removing an entity label" + ) + if step_id not in project.entity_labels: + raise QualificationError(f"no entity label is set for step id {step_id!r}") + previous = project.revision_digest() + del project.entity_labels[step_id] + _touch(project, previous) + _invalidate_certification(workflow) + return project + + +def list_entity_labels(workflow: "Workflow") -> list[QualifiedEntityLabel]: + """Return qualification-owned labels in stable step-id order.""" + + project = workflow.qualification + if project is None: + raise QualificationError( + "initialize qualification before listing entity labels" + ) + return [project.entity_labels[key] for key in sorted(project.entity_labels)] + + def set_identity_policy( workflow: "Workflow", policy: IdentityPolicy, diff --git a/openadapt_flow/runtime/durable/attended.py b/openadapt_flow/runtime/durable/attended.py index 7f26b756..75fb208b 100644 --- a/openadapt_flow/runtime/durable/attended.py +++ b/openadapt_flow/runtime/durable/attended.py @@ -1101,12 +1101,32 @@ def seal_human_decision_task(self, unsigned: dict[str, Any]) -> dict[str, Any]: ) from exc return task.model_dump(mode="json") + def seal_human_decision_task_v2(self, unsigned: dict[str, Any]) -> dict[str, Any]: + """Sign the negotiated V2 task under its distinct Types domain.""" + try: + from openadapt_types import sign_human_decision_task_v2_hmac + + task = sign_human_decision_task_v2_hmac( + key=self._key(create=False), + fields=unsigned, + ) + except (ImportError, ValueError) as exc: + raise AttendedActionRefused( + "the shared human decision V2 contract is unavailable or invalid" + ) from exc + return task.model_dump(mode="json") + def verify_human_decision_task(self, task: dict[str, Any]) -> bool: """Verify a projected task without treating it as pause authority.""" try: - from openadapt_types import HumanDecisionTaskV1 + from openadapt_types import HumanDecisionTaskV1, HumanDecisionTaskV2 - validated = HumanDecisionTaskV1.model_validate(task) + model = ( + HumanDecisionTaskV2 + if task.get("schema_version") == "openadapt.human-decision-task/v2" + else HumanDecisionTaskV1 + ) + validated = model.model_validate(task) return validated.verify_hmac(self._key(create=False)) except (ImportError, ValueError, AttendedActionRefused): return False diff --git a/pyproject.toml b/pyproject.toml index c56a9e8f..fc47f5da 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,12 +63,10 @@ dev = [ # uvicorn for the console's boot smoke test. "fastapi>=0.110", "uvicorn>=0.29", - # FLOOR is load-bearing, not cosmetic. The attended `reconcile` action and - # `reconciled_and_resumed` receipt pair are named symbols that only exist - # from 0.7.0; a resolver satisfied by 0.6.x fails at IMPORT, not at resolve, - # and does so inside the operator decision path. Raise this with every new - # symbol taken from the contract. - "openadapt-types>=0.7.0,<0.8.0", + # V2 uses the separately signed entity contract from Types 0.9.0. A peer + # must still explicitly negotiate it; the dependency alone never upgrades + # an existing V1 decision surface. + "openadapt-types>=0.9.0,<0.10.0", # Engineering-hygiene gates (lint+format, type-check, coverage). Pinned to # majors so CI and local dev run the same checkers. "ruff==0.15.22", @@ -91,7 +89,7 @@ grounding = ["openadapt-grounding>=0.1.0"] console = [ "fastapi>=0.110", "uvicorn>=0.29", - "openadapt-types>=0.7.0,<0.8.0", + "openadapt-types>=0.9.0,<0.10.0", ] # WindowsBackend: HTTP client for the WAA (Windows Agent Arena) server. windows = ["requests>=2.31"] @@ -153,10 +151,10 @@ capture = ["openadapt-capture>=1.2.0"] # Bounded pin: openadapt-types is 0-based semver (major_on_zero=false), so # breaking changes land at the MINOR level. The action shim and runtime-overlay # producer and attended-decision portal are field-exact against the released -# 0.7.x schemas. The +# 0.9.x schemas. The # `interop-types` CI job type-checks and tests both boundaries against the real -# package; raise the ceiling only after validating the next minor. -interop = ["openadapt-types>=0.7.0,<0.8.0"] +# package; V2 is consumed only by an explicitly negotiated peer. +interop = ["openadapt-types>=0.9.0,<0.10.0"] [project.scripts] openadapt-flow = "openadapt_flow.__main__:main" diff --git a/tests/test_attended_actions.py b/tests/test_attended_actions.py index b6f5b7ef..20121241 100644 --- a/tests/test_attended_actions.py +++ b/tests/test_attended_actions.py @@ -47,12 +47,16 @@ Transition, Workflow, ) +from openadapt_flow.policy import load_policy, policy_contract_sha256 from openadapt_flow.privacy import reset_scrubbers, set_image_scrubber from openadapt_flow.qualification import ( ActionRiskClassification, EnvironmentBoundary, + QualificationCertification, + QualifiedEntityLabel, init_project, set_action_classification, + set_entity_label, workflow_contract_sha256, ) from openadapt_flow.runtime.authorization import ( @@ -317,7 +321,7 @@ def _request(capability, action="continue", key="request-key-0001"): ) -def _remote_deployment() -> DeploymentConfig: +def _remote_deployment(**remote: object) -> DeploymentConfig: return DeploymentConfig.model_validate( { "human_decisions": { @@ -325,6 +329,7 @@ def _remote_deployment() -> DeploymentConfig: "enabled": True, "tenant_id": "tenant_exact_01", "runner_id": "runner_exact_01", + **remote, } } } @@ -3185,6 +3190,291 @@ def test_remote_projection_is_explicit_aal2_phi_free_and_exactly_bound(tmp_path) assert protected not in serialized +def _v2_candidate_workflow( + *, + with_entity_label: bool = True, + entity_label: str = "patient record", +) -> Workflow: + workflow = Workflow(name="attended-v2", steps=[_step("humanstep", "A")]) + project = init_project( + workflow, + environment=EnvironmentBoundary( + target_kind="web", + application="qualified-app", + application_version="1", + environment_digest="a" * 64, + runtime_version="1.26.0", + ), + ) + if with_entity_label: + set_entity_label( + workflow, + QualifiedEntityLabel( + step_id="humanstep", + label=entity_label, + fallback="record", + ), + ) + policy = load_policy("permissive") + project.last_certification = QualificationCertification( + project_revision=project.revision, + project_contract_sha256=project.contract_sha256(), + workflow_contract_sha256=workflow_contract_sha256(workflow), + environment_contract_sha256=project.environment.contract_sha256(), + policy_name=policy.name, + policy_contract_sha256=policy_contract_sha256(policy), + policy_contract=policy.model_dump(mode="json"), + passed=True, + report_sha256="a" * 64, + case_evidence_contract_sha256="b" * 64, + ) + return workflow + + +def _accept_current_certification(monkeypatch) -> None: + monkeypatch.setattr( + "openadapt_flow.qualification.current_certification_matches", + lambda _workflow, *, policy=None, policy_contract_digest=None: ( + policy is None and policy_contract_digest is not None + ), + ) + + +def test_remote_v2_requires_explicit_peer_negotiation_and_exact_label_binding( + tmp_path, monkeypatch +): + _accept_current_certification(monkeypatch) + workflow = _v2_candidate_workflow() + workflow, _bundle, run, _store, capability = _paused(tmp_path, workflow=workflow) + item = attention_item(run.parent, run) + assert item is not None + + v1 = portable_remote_decision_task(run, item, deployment=_remote_deployment()) + assert v1.task.schema_version == "openadapt.human-decision-task/v1" + + deployment = _remote_deployment( + peer_task_schemas=["openadapt.human-decision-task/v2"] + ) + v2 = portable_remote_decision_task( + run, + item, + deployment=deployment, + ) + assert v2.task.schema_version == "openadapt.human-decision-task/v2" + assert v2.task.entity.label == "patient record" + assert v2.task.entity.fallback.value == "record" + assert v2.task.qualification_step_id == "humanstep" + assert v2.task.qualification_project_id == workflow.qualification.project_id + assert v2.task.qualification_contract_digest == ( + "sha256:" + workflow.qualification.contract_sha256() + ) + decision = execute_remote_attended_action( + run, + item, + _remote_request(v2, capability).model_copy( + update={"action": "escalate", "disposition": "needs_assistance"} + ), + deployment=deployment, + principal=_remote_principal(), + executor=_ResultExecutor(), + ) + assert decision.status == "escalated" + assert decision.decided_by == "human" + + +def test_remote_v2_uses_neutral_entity_when_optional_label_is_absent( + tmp_path, monkeypatch +): + _accept_current_certification(monkeypatch) + workflow = _v2_candidate_workflow(with_entity_label=False) + workflow, _bundle, run, _store, _capability = _paused(tmp_path, workflow=workflow) + item = attention_item(run.parent, run) + assert item is not None + projection = portable_remote_decision_task( + run, + item, + deployment=_remote_deployment( + peer_task_schemas=["openadapt.human-decision-task/v2"] + ), + ) + assert projection.task.schema_version == "openadapt.human-decision-task/v2" + assert projection.task.entity.label == "record" + assert projection.task.entity.fallback.value == "record" + assert projection.task.qualification_step_id == "humanstep" + + +def test_remote_v2_falls_back_without_a_current_certification(tmp_path, monkeypatch): + _accept_current_certification(monkeypatch) + workflow = _v2_candidate_workflow() + assert workflow.qualification is not None + workflow.qualification.last_certification = None + _workflow, _bundle, run, _store, _capability = _paused(tmp_path, workflow=workflow) + item = attention_item(run.parent, run) + assert item is not None + projection = portable_remote_decision_task( + run, + item, + deployment=_remote_deployment( + peer_task_schemas=["openadapt.human-decision-task/v2"] + ), + ) + assert projection.task.schema_version == "openadapt.human-decision-task/v1" + + +@pytest.mark.parametrize("field", ("project_revision", "project_contract_sha256")) +def test_remote_v2_falls_back_for_a_stale_certification(tmp_path, field, monkeypatch): + _accept_current_certification(monkeypatch) + workflow = _v2_candidate_workflow() + assert workflow.qualification is not None + certification = workflow.qualification.last_certification + assert certification is not None + if field == "project_revision": + certification.project_revision += 1 + else: + certification.project_contract_sha256 = "0" * 64 + _workflow, _bundle, run, _store, _capability = _paused(tmp_path, workflow=workflow) + item = attention_item(run.parent, run) + assert item is not None + projection = portable_remote_decision_task( + run, + item, + deployment=_remote_deployment( + peer_task_schemas=["openadapt.human-decision-task/v2"] + ), + ) + assert projection.task.schema_version == "openadapt.human-decision-task/v1" + + +def test_remote_v2_falls_back_when_a_current_certification_has_new_bundle_bytes( + tmp_path, monkeypatch +): + _accept_current_certification(monkeypatch) + workflow = _v2_candidate_workflow() + workflow, bundle, run, _store, _capability = _paused(tmp_path, workflow=workflow) + assert workflow.qualification is not None + certification = workflow.qualification.last_certification + assert ( + certification is not None and certification.policy_contract_sha256 is not None + ) + # `certified_at` is not an input to qualification evaluation. This changes + # sealed bundle bytes while leaving the existing certification current. + certification.certified_at = "2026-07-30T00:00:00+00:00" + workflow.save(bundle) + current = Workflow.load(bundle) + from openadapt_flow import qualification + + assert qualification.current_certification_matches( + current, + policy_contract_digest=certification.policy_contract_sha256, + ) + item = attention_item(run.parent, run) + assert item is not None + projection = portable_remote_decision_task( + run, + item, + deployment=_remote_deployment( + peer_task_schemas=["openadapt.human-decision-task/v2"] + ), + ) + assert projection.task.schema_version == "openadapt.human-decision-task/v1" + + +def test_remote_v2_keeps_a_custom_entity_label_local(tmp_path, monkeypatch): + _accept_current_certification(monkeypatch) + workflow = _v2_candidate_workflow(entity_label="appointment request") + workflow, _bundle, run, _store, _capability = _paused(tmp_path, workflow=workflow) + item = attention_item(run.parent, run) + assert item is not None + projection = portable_remote_decision_task( + run, + item, + deployment=_remote_deployment( + peer_task_schemas=["openadapt.human-decision-task/v2"] + ), + ) + assert projection.task.schema_version == "openadapt.human-decision-task/v2" + assert projection.task.entity.label == "record" + assert projection.task.entity.fallback.value == "record" + assert projection.task.qualification_step_id == "humanstep" + + +def test_remote_v2_falls_back_when_report_step_differs_from_capability(tmp_path): + workflow = Workflow( + name="attended-v2", + steps=[_step("humanstep", "A"), _step("otherstep", "B")], + ) + init_project( + workflow, + environment=EnvironmentBoundary( + target_kind="web", + application="qualified-app", + application_version="1", + environment_digest="a" * 64, + runtime_version="1.26.0", + ), + ) + for step_id in ("humanstep", "otherstep"): + set_entity_label( + workflow, + QualifiedEntityLabel( + step_id=step_id, label="patient record", fallback="record" + ), + ) + _workflow, _bundle, run, _store, _capability = _paused(tmp_path, workflow=workflow) + report_path = run / "report.json" + report_payload = json.loads(report_path.read_text(encoding="utf-8")) + report_payload["results"][0]["step_id"] = "otherstep" + report_path.write_text(json.dumps(report_payload), encoding="utf-8") + item = attention_item(run.parent, run) + assert item is not None + projection = portable_remote_decision_task( + run, + item, + deployment=_remote_deployment( + peer_task_schemas=["openadapt.human-decision-task/v2"] + ), + ) + assert projection.task.schema_version == "openadapt.human-decision-task/v1" + + +def test_remote_v2_falls_back_after_the_paused_bundle_changes(tmp_path): + workflow = Workflow(name="attended-v2", steps=[_step("humanstep", "A")]) + init_project( + workflow, + environment=EnvironmentBoundary( + target_kind="web", + application="qualified-app", + application_version="1", + environment_digest="a" * 64, + runtime_version="1.26.0", + ), + ) + set_entity_label( + workflow, + QualifiedEntityLabel( + step_id="humanstep", label="patient record", fallback="record" + ), + ) + workflow, bundle, run, _store, _capability = _paused(tmp_path, workflow=workflow) + set_entity_label( + workflow, + QualifiedEntityLabel( + step_id="humanstep", label="insurance claim", fallback="item" + ), + ) + workflow.save(bundle) + item = attention_item(run.parent, run) + assert item is not None + projection = portable_remote_decision_task( + run, + item, + deployment=_remote_deployment( + peer_task_schemas=["openadapt.human-decision-task/v2"] + ), + ) + assert projection.task.schema_version == "openadapt.human-decision-task/v1" + + def test_remote_response_refuses_scope_or_binding_drift(tmp_path): _workflow, _bundle, run, _store, capability = _paused(tmp_path) item = attention_item(run.parent, run) diff --git a/tests/test_qualification_project.py b/tests/test_qualification_project.py index fe9e3209..f5477f6d 100644 --- a/tests/test_qualification_project.py +++ b/tests/test_qualification_project.py @@ -58,6 +58,7 @@ policy_contract_sha256, ) from openadapt_flow.qualification import ( + REMOTE_SAFE_ENTITY_LABELS, ActionRiskClass, ActionRiskClassification, EnvironmentBoundary, @@ -73,19 +74,25 @@ QualificationError, QualificationOutcome, QualificationRefusalCode, + QualifiedEntityLabel, RequalificationCondition, VerificationTier, add_case, add_requalification_condition, certify_project, current_certification_matches, + entity_label_options, evaluate_qualification, init_project, + list_entity_labels, + project_schema, qualification_action_requirements, record_case_results, + remove_entity_label, set_action_classification, set_case_scope, set_effect_policy, + set_entity_label, set_identity_policy, set_minimum_effect_tier, set_trusted_fault_driver_key, @@ -184,6 +191,119 @@ def _environment() -> EnvironmentBoundary: ) +def test_entity_labels_are_qualification_contract_and_invalidate_certification(): + workflow = _workflow() + project = init_project(workflow, environment=_environment()) + before = project.contract_sha256() + certification = QualificationCertification( + project_revision=project.revision, + project_contract_sha256=before, + workflow_contract_sha256="a" * 64, + environment_contract_sha256="c" * 64, + policy_name="clinical-write", + policy_contract_sha256="b" * 64, + passed=True, + report_sha256="d" * 64, + certified_at="2026-07-01T00:00:00+00:00", + ) + project.last_certification = certification + + set_entity_label( + workflow, + QualifiedEntityLabel(step_id="save", label="patient record", fallback="record"), + ) + assert project.entity_labels["save"].label == "patient record" + assert project.contract_sha256() != before + assert project.last_certification is None + assert list_entity_labels(workflow) == [project.entity_labels["save"]] + + remove_entity_label(workflow, "save") + assert project.entity_labels == {} + + +def test_entity_label_is_optional_for_certification(tmp_path: Path) -> None: + """Presentation metadata must never become a safety admission gate.""" + + workflow = _workflow() + bundle = tmp_path / "bundle" + (bundle / "templates").mkdir(parents=True) + (bundle / "templates" / "save.png").write_bytes(_qualification_visual_fixture()[1]) + workflow.save(bundle) + workflow = Workflow.load(bundle) + _configure(workflow, tier=VerificationTier.INDEPENDENT_SYSTEM) + assert workflow.qualification is not None + assert workflow.qualification.entity_labels == {} + evidence_root = tmp_path / "evidence" + _record_passing_campaign(workflow, evidence_root) + + report = certify_project( + workflow, + policy=load_policy("clinical-write"), + evidence_root=evidence_root, + ) + + assert report.passed + assert workflow.qualification.entity_labels == {} + + +def test_entity_label_options_are_ordered_and_derive_every_public_label() -> None: + options = entity_label_options() + assert options == [ + {"label": "patient record", "fallback": "record"}, + {"label": "member record", "fallback": "record"}, + {"label": "insurance claim", "fallback": "item"}, + {"label": "loan application", "fallback": "item"}, + {"label": "customer account", "fallback": "record"}, + {"label": "service request", "fallback": "item"}, + {"label": "case", "fallback": "item"}, + {"label": "order", "fallback": "item"}, + {"label": "invoice", "fallback": "item"}, + {"label": "document", "fallback": "item"}, + {"label": "record", "fallback": "record"}, + {"label": "item", "fallback": "item"}, + ] + assert REMOTE_SAFE_ENTITY_LABELS == tuple(option["label"] for option in options) + label_schema = project_schema()["$defs"]["QualifiedEntityLabel"]["properties"][ + "label" + ] + assert "enum" not in label_schema + assert label_schema["x-openadapt-reviewed-remote-options"] == list( + REMOTE_SAFE_ENTITY_LABELS + ) + + +@pytest.mark.parametrize("option", entity_label_options()) +def test_entity_label_accepts_each_reviewed_remote_safe_class( + option: dict[str, str], +) -> None: + entity = QualifiedEntityLabel(step_id="save", **option) + assert entity.model_dump() == {"step_id": "save", **option} + + +def test_entity_label_refuses_a_noncanonical_fallback() -> None: + with pytest.raises(ValueError, match="fallback"): + QualifiedEntityLabel(step_id="save", label="insurance claim", fallback="record") + + +def test_entity_label_accepts_a_custom_local_class() -> None: + entity = QualifiedEntityLabel( + step_id="save", label="appointment request", fallback="item" + ) + assert entity.label == "appointment request" + assert entity.fallback == "item" + + +@pytest.mark.parametrize( + "label", + ("Patient Record", "record/class", "one two three four five"), +) +def test_entity_label_refuses_invalid_local_class_syntax( + label: str, +) -> None: + with pytest.raises(ValueError, match="lowercase class name"): + QualifiedEntityLabel(step_id="save", label=label, fallback="record") + + def test_api_only_qualification_uses_only_executable_targets() -> None: effect = Effect( kind=EffectKind.FIELD_EQUALS, @@ -2872,6 +2992,75 @@ def test_cli_initializes_project_without_raw_manifest_editing( assert main(["qualify", "explain", str(bundle), "--json"]) == 2 payload = capsys.readouterr().out assert '"representative_case_missing"' in payload + assert json.loads(payload)["entity_label_options"] == entity_label_options() + + +def test_cli_entity_label_notice_tracks_real_mutations( + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + workflow = _workflow() + bundle = tmp_path / "bundle" + workflow.save(bundle) + init_project(workflow, environment=_environment()) + workflow.save(bundle) + + set_args = [ + "qualify", + "label", + "set", + str(bundle), + "--step", + "save", + "--label", + "insurance claim", + ] + invalid_args = [*set_args] + invalid_args[invalid_args.index("insurance claim")] = "Patient Record" + with pytest.raises(SystemExit, match="invalid entity label"): + main(invalid_args) + assert Workflow.load(bundle).qualification.entity_labels == {} + + assert main(set_args) == 0 + first = capsys.readouterr() + assert json.loads(first.out)["entity_labels"]["save"]["label"] == "insurance claim" + assert first.err + + assert main(set_args) == 0 + unchanged = capsys.readouterr() + assert json.loads(unchanged.out)["entity_labels"]["save"]["fallback"] == "item" + assert not unchanged.err + + assert ( + main( + [ + "qualify", + "label", + "set", + str(bundle), + "--step", + "save", + "--label", + "appointment request", + "--fallback", + "item", + ] + ) + == 0 + ) + custom = capsys.readouterr() + custom_payload = json.loads(custom.out)["entity_labels"]["save"] + assert custom_payload == { + "step_id": "save", + "label": "appointment request", + "fallback": "item", + } + assert custom.err + + assert main(["qualify", "label", "remove", str(bundle), "--step", "save"]) == 0 + removed = capsys.readouterr() + assert json.loads(removed.out)["entity_labels"] == {} + assert removed.err def test_cli_identity_extract_pattern_round_trips_exactly(tmp_path: Path) -> None: diff --git a/uv.lock b/uv.lock index 73fc1f84..fd9262f2 100644 --- a/uv.lock +++ b/uv.lock @@ -2243,9 +2243,9 @@ requires-dist = [ { name = "openadapt-capture", marker = "extra == 'capture'", specifier = ">=1.2.0" }, { name = "openadapt-grounding", marker = "extra == 'grounding'", specifier = ">=0.1.0" }, { name = "openadapt-privacy", extras = ["presidio"], marker = "extra == 'privacy'", specifier = ">=1.0.0" }, - { name = "openadapt-types", marker = "extra == 'console'", specifier = ">=0.7.0,<0.8.0" }, - { name = "openadapt-types", marker = "extra == 'dev'", specifier = ">=0.7.0,<0.8.0" }, - { name = "openadapt-types", marker = "extra == 'interop'", specifier = ">=0.7.0,<0.8.0" }, + { name = "openadapt-types", marker = "extra == 'console'", specifier = ">=0.9.0,<0.10.0" }, + { name = "openadapt-types", marker = "extra == 'dev'", specifier = ">=0.9.0,<0.10.0" }, + { name = "openadapt-types", marker = "extra == 'interop'", specifier = ">=0.9.0,<0.10.0" }, { name = "opencv-python-headless", specifier = ">=4.9" }, { name = "pillow", specifier = ">=10.0" }, { name = "playwright", marker = "extra == 'browser'", specifier = ">=1.44" }, @@ -2310,14 +2310,14 @@ presidio = [ [[package]] name = "openadapt-types" -version = "0.7.0" +version = "0.9.0" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "pydantic" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/dc/bf/39ec24f545b3fc09bdff8cc61ca448bb968450c43a99f8a6808e538a3c57/openadapt_types-0.7.0.tar.gz", hash = "sha256:80a470aa4e8870d2a2b2645f2dbfac04ab9f158fb6d21b314d9eff0296a0ee2f", size = 126130, upload-time = "2026-07-29T19:32:31.082Z" } +sdist = { url = "https://files.pythonhosted.org/packages/75/a4/bac2263f9996248f0c1a4e1e62b3d0ff1a3cfe0a08cb3b1eaf4ae3a106f7/openadapt_types-0.9.0.tar.gz", hash = "sha256:e377750797c1999c6a74cd8730d9d952e22c68fe8078f8270d0d9c7ca21c84e8", size = 134336 } wheels = [ - { url = "https://files.pythonhosted.org/packages/5c/39/b78fcd51096939884db01ea10bb4594a9e88aaa7d40c3eff537cfba8d971/openadapt_types-0.7.0-py3-none-any.whl", hash = "sha256:4805f2f97dbe46c15874d0fbf88dc9bf94a430b2b9a2f3e7ed24cddef013b7f2", size = 81743, upload-time = "2026-07-29T19:32:29.584Z" }, + { url = "https://files.pythonhosted.org/packages/9f/33/4fb8cd26708808e99c76ea4348fa8ba48bc0c9183ff457a4bec5b6b660c8/openadapt_types-0.9.0-py3-none-any.whl", hash = "sha256:cc7bd12a5d7203b75dd60d860e8393c315d4dd7e9406346859e0ce0aa05dda89", size = 88605 }, ] [[package]]