From a13f7148954fe853b7941db9315ef45b90cbc909 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 18 Aug 2026 12:49:43 -0400 Subject: [PATCH 1/6] feat: add structured push result contract --- docs/PUSH_JSON.md | 114 +++++++++ docs/SANITIZED_ARTIFACTS.md | 5 + openadapt_flow/__main__.py | 393 ++++++++++++++++++++++++++++++- openadapt_flow/hosted.py | 91 ++++++- public-artifacts.json | 4 + schemas/push-result-v1.json | 260 ++++++++++++++++++++ tests/test_hosted.py | 259 ++++++++++++++++++++ tests/test_sanitized_artifact.py | 9 + 8 files changed, 1127 insertions(+), 8 deletions(-) create mode 100644 docs/PUSH_JSON.md create mode 100644 schemas/push-result-v1.json diff --git a/docs/PUSH_JSON.md b/docs/PUSH_JSON.md new file mode 100644 index 00000000..65ed8acf --- /dev/null +++ b/docs/PUSH_JSON.md @@ -0,0 +1,114 @@ +# `push --json` controller contract + +Desktop and other local controllers can add `--json` to `openadapt-flow push`. +Flow then writes one compact JSON object to standard output. The schema is +[`openadapt.push-result/v1`](../schemas/push-result-v1.json). + +The command without `--json` keeps its existing human-readable output. + +## Status and exit code + +| `status` | Exit code | Meaning | `next_action` | +|---|---:|---|---| +| `paused_for_review` | 0 | Flow made a local sanitized derivative. It did not upload it. | `review_local` | +| `accepted_for_ingest` | 0 | The server acknowledged the exact approved archive and returned its stable ingest id. | `parameterize`, `validate_runtime`, or `open_dashboard` | +| `failed` | 1 | Flow did not receive a complete accepted-ingest contract. | `null` or `reconcile` | +| `delivery_uncertain` | 1 | A transport failure occurred after Flow attempted the request. The server can have received it. | `reconcile` | + +Do not retry `delivery_uncertain` automatically. Use `artifact_sha256` to +reconcile the request with the hosted control plane first. + +## Stable fields + +V1 always includes these top-level keys: + +```text +schema, status, workflow_id, artifact_ingest_id, review, attestation, +binding, next_action, dashboard_url, delivery, error +``` + +An unused value is `null`. Flow does not omit the key. + +`artifact_ingest_id` is the server-owned `artifact_ingests.id`. Flow does not +create it. JSON mode does not return `accepted_for_ingest` if a server response +omits this id or does not echo the exact approved archive hash. + +A recording ingest has `workflow_id: null`. It is not a runnable workflow. Its +next action is parameterization or runtime validation. An accepted bundle has +a workflow UUID, the server ingest UUID, a same-origin dashboard URL, and the +exact local runtime-attestation binding. + +The `binding` object lets Desktop detect a stale handoff. It carries: + +- the source tree, derivative tree, approved archive, and acknowledged artifact + SHA-256 values; +- the exact bundle and source-recording SHA-256 values for a bundle; +- the sanitization and certification policies; +- the certification evidence, parameter schema, governed authorization + template, and attested run-report SHA-256 values; and +- the halted run UUID when the bundle resolves a governed halt. + +The local `review.id` is a domain-separated SHA-256 of the canonical sanitized +manifest. It is stable for that exact review candidate. It is local and +non-authoritative. It is not a hosted approval id. The attestation `id` is the +server challenge id that the local runtime-validation attestation signs. + +## Examples + +A raw recording normally pauses locally: + +```bash +openadapt-flow push recording/ --json +``` + +```json +{ + "schema": "openadapt.push-result/v1", + "status": "paused_for_review", + "workflow_id": null, + "artifact_ingest_id": null, + "review": { + "id": "", + "scope": "local_non_authoritative", + "sanitized_path": "", + "command": "openadapt-flow review-sanitized --original " + }, + "attestation": null, + "binding": { + "kind": "recording", + "source_tree_sha256": "", + "derivative_tree_sha256": "", + "approved_archive_sha256": null, + "artifact_sha256": null, + "bundle_sha256": null, + "source_recording_sha256": null, + "sanitization_policy": "outbound-phi-v1", + "certification_policy": null, + "certification_evidence_sha256": null, + "governed_authorization_template_sha256": null, + "parameter_schema_sha256": null, + "attested_run_report_sha256": null, + "resolves_run_id": null + }, + "next_action": "review_local", + "dashboard_url": null, + "delivery": { "attempted": false, "certainty": "not_attempted" }, + "error": null +} +``` + +After local approval, use the same flag for the approved derivative: + +```bash +openadapt-flow push approved-bundle/ --kind bundle \ + --validation-attestation attestation.json --json +``` + +Flow returns `accepted_for_ingest` only after it checks all required server ids, +the echoed artifact hash, and the exact local attestation binding. + +## Error privacy + +Machine-readable errors use bounded messages of at most 500 characters. They +do not copy a token, raw server body, or local source path into the JSON result. +The human-readable command keeps its existing diagnostic output. diff --git a/docs/SANITIZED_ARTIFACTS.md b/docs/SANITIZED_ARTIFACTS.md index a679893a..232619d9 100644 --- a/docs/SANITIZED_ARTIFACTS.md +++ b/docs/SANITIZED_ARTIFACTS.md @@ -37,6 +37,11 @@ openadapt-flow push triage.bundle.sanitized/ --kind bundle \ --validation-attestation triage.validation.json ``` +Desktop and other local controllers can add `--json` to receive the stable +[`openadapt.push-result/v1`](PUSH_JSON.md) result. It distinguishes a local +`paused_for_review` result from a server-acknowledged `accepted_for_ingest` +result. Human-readable output remains the default. + For an existing hosted workflow, add `--workflow-id ` to the bundle push. When the replacement repairs a specific hosted halt, also pass `--resolves-run-id `. Cloud locks that unresolved run in the diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index eb248e05..c4b5534f 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -55,11 +55,13 @@ from __future__ import annotations import argparse +import json import sys from contextlib import contextmanager from pathlib import Path from typing import TYPE_CHECKING, Any, Iterator, Literal, Optional, Sequence, cast from urllib.parse import urlsplit +from uuid import UUID if TYPE_CHECKING: # pragma: no cover from openadapt_flow.backend import Backend @@ -3201,6 +3203,370 @@ def _cmd_connect(args: argparse.Namespace) -> int: return 0 +_PUSH_JSON_SCHEMA = "openadapt.push-result/v1" + + +def _push_json_base(status: str) -> dict[str, Any]: + """Return the complete V1 shape; phase-specific fields stay explicit nulls.""" + return { + "schema": _PUSH_JSON_SCHEMA, + "status": status, + "workflow_id": None, + "artifact_ingest_id": None, + "review": None, + "attestation": None, + "binding": { + "kind": None, + "source_tree_sha256": None, + "derivative_tree_sha256": None, + "approved_archive_sha256": None, + "artifact_sha256": None, + "bundle_sha256": None, + "source_recording_sha256": None, + "sanitization_policy": None, + "certification_policy": None, + "certification_evidence_sha256": None, + "governed_authorization_template_sha256": None, + "parameter_schema_sha256": None, + "attested_run_report_sha256": None, + "resolves_run_id": None, + }, + "next_action": None, + "dashboard_url": None, + "delivery": {"attempted": False, "certainty": "not_attempted"}, + "error": None, + } + + +def _is_sha256(value: Any) -> bool: + return ( + isinstance(value, str) + and len(value) == 64 + and all(char in "0123456789abcdef" for char in value) + ) + + +def _is_uuid(value: Any) -> bool: + if not isinstance(value, str): + return False + try: + return str(UUID(value)) == value + except ValueError: + return False + + +def _is_runtime_attestation_schema(value: Any) -> bool: + prefix = "openadapt.runtime-validation/v" + return ( + isinstance(value, str) + and value.startswith(prefix) + and value.removeprefix(prefix).isdigit() + ) + + +def _required_sha256(value: Any, *, field: str) -> str: + if not _is_sha256(value): + raise ValueError(f"invalid {field}") + return value + + +def _push_json_failure( + *, uncertain: bool = False, context: Optional[dict[str, Any]] = None +) -> dict[str, Any]: + status = "delivery_uncertain" if uncertain else "failed" + document = _push_json_base(status) + if uncertain: + document["next_action"] = "reconcile" + document["delivery"] = {"attempted": True, "certainty": "unknown"} + code = "delivery_uncertain" + message = ( + "The upload did not return a trustworthy delivery result. Reconcile " + "the artifact in the hosted control plane before any retry." + ) + else: + document["delivery"] = { + "attempted": None, + "certainty": "not_accepted", + } + code = "push_failed" + message = "The artifact was not accepted for ingest." + document["error"] = {"code": code, "message": message[:500]} + if uncertain and context: + local = context.get("local_binding") + sanitization = context.get("sanitization") + artifact = ( + sanitization.get("artifact") if isinstance(sanitization, dict) else None + ) + if isinstance(local, dict) and isinstance(artifact, dict): + binding = document["binding"] + values = { + "kind": artifact.get("kind"), + "source_tree_sha256": local.get("source_tree_sha256"), + "derivative_tree_sha256": local.get("derivative_tree_sha256"), + "approved_archive_sha256": local.get("approved_archive_sha256"), + "artifact_sha256": artifact.get("sha256"), + "sanitization_policy": local.get("sanitization_policy"), + "resolves_run_id": context.get("resolves_run_id"), + } + if ( + values["kind"] in ("recording", "bundle") + and all( + _is_sha256(values[key]) + for key in ( + "source_tree_sha256", + "derivative_tree_sha256", + "approved_archive_sha256", + "artifact_sha256", + ) + ) + and values["approved_archive_sha256"] == values["artifact_sha256"] + and isinstance(values["sanitization_policy"], str) + and ( + values["resolves_run_id"] is None + or _is_uuid(values["resolves_run_id"]) + ) + ): + binding.update(values) + review_id = context.get("review_id") + if _is_sha256(review_id): + document["review"] = { + "id": review_id, + "scope": "local_non_authoritative", + "sanitized_path": None, + "command": None, + } + attestation = context.get("attestation_binding") + if isinstance(attestation, dict): + challenge_id = attestation.get("challenge_id") + attestation_schema = attestation.get("schema") + if ( + isinstance(challenge_id, str) + and 1 <= len(challenge_id) <= 200 + and _is_runtime_attestation_schema(attestation_schema) + ): + document["attestation"] = { + "id": challenge_id, + "schema": attestation_schema, + } + optional_binding = { + "bundle_sha256": attestation.get("bundle_sha256"), + "source_recording_sha256": attestation.get( + "source_recording_sha256" + ), + "certification_policy": attestation.get("policy"), + "certification_evidence_sha256": attestation.get( + "policy_evidence_sha256" + ), + "governed_authorization_template_sha256": attestation.get( + "governed_authorization_template_sha256" + ), + "parameter_schema_sha256": attestation.get( + "parameter_schema_sha256" + ), + "attested_run_report_sha256": attestation.get( + "run_report_sha256" + ), + } + for key, value in optional_binding.items(): + if key == "certification_policy": + if isinstance(value, str) and value: + binding[key] = value + elif value is None or _is_sha256(value): + binding[key] = value + return document + + +def _push_json_invalid_response() -> dict[str, Any]: + document = _push_json_base("failed") + document["next_action"] = "reconcile" + document["delivery"] = {"attempted": True, "certainty": "unknown"} + document["error"] = { + "code": "invalid_ingest_response", + "message": ( + "The server response did not prove an exact accepted ingest. " + "Reconcile the artifact in the hosted control plane before any retry." + ), + } + return document + + +def _push_json_result(result: dict[str, Any]) -> dict[str, Any]: + """Build a stable Desktop contract from verified local and server evidence.""" + local_binding = result.get("local_binding") + if not isinstance(local_binding, dict): + raise ValueError("missing local binding") + source_sha = _required_sha256( + local_binding.get("source_tree_sha256"), field="source tree binding" + ) + derivative_sha = _required_sha256( + local_binding.get("derivative_tree_sha256"), + field="derivative tree binding", + ) + sanitization_policy = local_binding.get("sanitization_policy") + if not isinstance(sanitization_policy, str) or not sanitization_policy: + raise ValueError("missing sanitization policy") + review_id = _required_sha256(result.get("review_id"), field="review id") + + if result.get("pending_review") is True and result.get("uploaded") is False: + kind = result.get("kind") or "recording" + if kind not in ("recording", "bundle"): + raise ValueError("invalid review kind") + sanitized_path = result.get("sanitized_path") + review_command = result.get("review_command") + if not isinstance(sanitized_path, str) or not sanitized_path: + raise ValueError("missing sanitized review path") + if not isinstance(review_command, str) or not review_command: + raise ValueError("missing review command") + document = _push_json_base("paused_for_review") + document["review"] = { + "id": review_id, + "scope": "local_non_authoritative", + "sanitized_path": sanitized_path, + "command": review_command, + } + document["binding"].update( + { + "kind": kind, + "source_tree_sha256": source_sha, + "derivative_tree_sha256": derivative_sha, + "sanitization_policy": sanitization_policy, + } + ) + document["next_action"] = "review_local" + return document + + if result.get("uploaded") is not True: + raise ValueError("missing upload acknowledgment") + sanitization = result.get("sanitization") + artifact = sanitization.get("artifact") if isinstance(sanitization, dict) else None + if not isinstance(artifact, dict): + raise ValueError("missing sanitization binding") + kind = artifact.get("kind") + if kind not in ("recording", "bundle") or result.get("kind") != kind: + raise ValueError("ingest kind mismatch") + approved_sha = _required_sha256( + local_binding.get("approved_archive_sha256"), + field="approved archive binding", + ) + artifact_sha = _required_sha256( + artifact.get("sha256"), field="local artifact binding" + ) + if approved_sha != artifact_sha or result.get("artifact_sha256") != artifact_sha: + raise ValueError("server artifact binding mismatch") + artifact_ingest_id = result.get("artifact_ingest_id") + if not _is_uuid(artifact_ingest_id): + raise ValueError("missing server artifact ingest id") + + document = _push_json_base("accepted_for_ingest") + document["artifact_ingest_id"] = artifact_ingest_id + document["review"] = { + "id": review_id, + "scope": "local_non_authoritative", + "sanitized_path": None, + "command": None, + } + document["binding"].update( + { + "kind": kind, + "source_tree_sha256": source_sha, + "derivative_tree_sha256": derivative_sha, + "approved_archive_sha256": approved_sha, + "artifact_sha256": artifact_sha, + "sanitization_policy": sanitization_policy, + } + ) + document["delivery"] = {"attempted": True, "certainty": "accepted"} + + if kind == "recording": + if result.get("workflow_id") is not None: + raise ValueError("recording ingest must not activate a workflow") + server_status = result.get("status") + if server_status == "needs_parameterization": + document["next_action"] = "parameterize" + elif server_status == "needs_runtime_validation": + document["next_action"] = "validate_runtime" + else: + raise ValueError("recording ingest has no governed next action") + return document + + workflow_id = result.get("workflow_id") + if not _is_uuid(workflow_id): + raise ValueError("bundle ingest has no workflow id") + attestation = result.get("attestation_binding") + if not isinstance(attestation, dict): + raise ValueError("bundle ingest has no attestation binding") + challenge_id = attestation.get("challenge_id") + schema = attestation.get("schema") + policy = attestation.get("policy") + if not isinstance(challenge_id, str) or not 1 <= len(challenge_id) <= 200: + raise ValueError("invalid attestation id") + if not _is_runtime_attestation_schema(schema): + raise ValueError("invalid attestation schema") + if not isinstance(policy, str) or not policy: + raise ValueError("invalid certification policy") + bundle_sha = _required_sha256( + attestation.get("bundle_sha256"), field="bundle binding" + ) + if bundle_sha != approved_sha: + raise ValueError("attested bundle does not match approved archive") + policy_evidence_sha = _required_sha256( + attestation.get("policy_evidence_sha256"), + field="policy evidence binding", + ) + parameter_schema_sha = _required_sha256( + attestation.get("parameter_schema_sha256"), + field="parameter schema binding", + ) + run_report_sha = _required_sha256( + attestation.get("run_report_sha256"), field="run report binding" + ) + source_recording_sha = _required_sha256( + attestation.get("source_recording_sha256"), + field="attested source recording binding", + ) + template_sha = attestation.get("governed_authorization_template_sha256") + if template_sha is not None: + template_sha = _required_sha256( + template_sha, field="governed authorization template binding" + ) + resolves_run_id = result.get("resolves_run_id") + if resolves_run_id is not None and not _is_uuid(resolves_run_id): + raise ValueError("invalid resolved run binding") + document["workflow_id"] = workflow_id + document["attestation"] = {"id": challenge_id, "schema": schema} + document["binding"].update( + { + "bundle_sha256": bundle_sha, + "source_recording_sha256": source_recording_sha, + "certification_policy": policy, + "certification_evidence_sha256": policy_evidence_sha, + "governed_authorization_template_sha256": template_sha, + "parameter_schema_sha256": parameter_schema_sha, + "attested_run_report_sha256": run_report_sha, + "resolves_run_id": resolves_run_id, + } + ) + dashboard_url = result.get("dashboard_url") + destination_host = result.get("destination_host") + if not isinstance(dashboard_url, str) or not isinstance(destination_host, str): + raise ValueError("missing trusted dashboard binding") + dashboard = urlsplit(dashboard_url) + destination = urlsplit(destination_host) + if dashboard[:2] != destination[:2] or ( + dashboard.path != f"/dashboard/workflows/{workflow_id}" + or dashboard.query + or dashboard.fragment + ): + raise ValueError("dashboard origin mismatch") + document["dashboard_url"] = dashboard_url + document["next_action"] = "open_dashboard" + return document + + +def _print_push_json(document: dict[str, Any]) -> None: + print(json.dumps(document, sort_keys=True, separators=(",", ":"))) + + def _cmd_push(args: argparse.Namespace) -> int: """Upload the exact approved sanitized archive to ``/api/ingest``. @@ -3208,7 +3574,7 @@ def _cmd_push(args: argparse.Namespace) -> int: a derivative and pauses for review; approved input sends the exact frozen archive and prints the server-assigned workflow id/dashboard URL. """ - from openadapt_flow.hosted import HostedError, push + from openadapt_flow.hosted import HostedDeliveryUncertain, HostedError, push try: result = push( @@ -3227,9 +3593,26 @@ def _cmd_push(args: argparse.Namespace) -> int: auto_approve=args.auto_approve, validation_attestation=args.validation_attestation, ) + except HostedDeliveryUncertain as e: + if args.json: + _print_push_json(_push_json_failure(uncertain=True, context=e.context)) + return 1 + print(f"push failed: {e}") + return 1 except HostedError as e: + if args.json: + _print_push_json(_push_json_failure()) + return 1 print(f"push failed: {e}") return 1 + if args.json: + try: + document = _push_json_result(result) + except ValueError: + _print_push_json(_push_json_invalid_response()) + return 1 + _print_push_json(document) + return 0 if result.get("pending_review"): print(f"Sanitized derivative created at {result['sanitized_path']}.") print( @@ -5669,6 +6052,14 @@ def _repair_store_flag(rp: argparse.ArgumentParser) -> None: "then an existing config migration token)" ), ) + p.add_argument( + "--json", + action="store_true", + help=( + "Emit one stable openadapt.push-result/v1 JSON object for Desktop " + "and other local controllers" + ), + ) p.set_defaults(func=_cmd_push) p = sub.add_parser( diff --git a/openadapt_flow/hosted.py b/openadapt_flow/hosted.py index ed87e070..412e0160 100644 --- a/openadapt_flow/hosted.py +++ b/openadapt_flow/hosted.py @@ -37,6 +37,7 @@ from __future__ import annotations +import hashlib import ipaddress import json import os @@ -59,6 +60,7 @@ __all__ = [ "DEFAULT_HOST", "HostedError", + "HostedDeliveryUncertain", "config_path", "resolve_host", "resolve_token", @@ -114,6 +116,14 @@ class HostedError(RuntimeError): """A hosted-connectivity failure (auth, network, or a non-2xx response).""" +class HostedDeliveryUncertain(HostedError): + """The upload request failed without a trustworthy delivery result.""" + + def __init__(self, message: str, *, context: Optional[dict[str, Any]] = None): + super().__init__(message) + self.context = context or {} + + @dataclass(frozen=True) class DestinationPolicy: """An authenticated upload destination whose trust was explicitly resolved.""" @@ -1098,6 +1108,17 @@ def login( # --------------------------------------------------------------------------- +def _local_review_id(manifest: dict[str, Any]) -> str: + """Return a local, non-authoritative id bound to one review manifest.""" + payload = json.dumps( + manifest, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("utf-8") + return hashlib.sha256(b"openadapt.sanitized-review/v1\0" + payload).hexdigest() + + def push( path: Optional[Any] = None, *, @@ -1221,6 +1242,7 @@ def push( automatic=True, ) else: + review_manifest = load_and_verify_derivative(derivative) return { "uploaded": False, "pending_review": True, @@ -1229,8 +1251,19 @@ def push( f"openadapt-flow review-sanitized {derivative} --original {src}" ), "destination_kind": destination.kind, + "destination_host": resolved_host, "deployment_kind": lane, "phi_mode": _phi_mode(phi_mode), + "kind": actual_kind, + "review_id": _local_review_id(review_manifest), + "local_binding": { + "source_tree_sha256": review_manifest["source_tree_sha256"], + "derivative_tree_sha256": review_manifest[ + "derivative_tree_sha256" + ], + "approved_archive_sha256": None, + "sanitization_policy": review_manifest["policy_version"], + }, } except SanitizationError as exc: raise HostedError(f"Artifact sanitization failed: {exc}") from exc @@ -1257,6 +1290,7 @@ def push( data["workflow_id"] = normalized_workflow_id if normalized_resolves_run_id is not None: data["resolves_run_id"] = normalized_resolves_run_id + attestation: Optional[dict[str, Any]] = None if kind == "bundle": if validation_attestation is None: raise HostedError( @@ -1326,6 +1360,36 @@ def push( data["sanitization_manifest"] = json.dumps( ingest_manifest, sort_keys=True, separators=(",", ":") ) + local_result: dict[str, Any] = { + "sanitization": ingest_manifest, + "approval": approval, + "destination_kind": destination.kind, + "destination_host": resolved_host, + "review_id": _local_review_id(local_manifest), + "local_binding": { + "source_tree_sha256": local_manifest["source_tree_sha256"], + "derivative_tree_sha256": local_manifest["derivative_tree_sha256"], + "approved_archive_sha256": approval["approved_derivative_sha256"], + "sanitization_policy": local_manifest["policy_version"], + }, + "resolves_run_id": normalized_resolves_run_id, + } + if attestation is not None: + certification = attestation.get("certification") or {} + replay = attestation.get("replay") or {} + local_result["attestation_binding"] = { + "schema": attestation.get("schema"), + "challenge_id": attestation.get("challenge_id"), + "source_recording_sha256": attestation.get("source_recording_sha256"), + "bundle_sha256": attestation.get("bundle_sha256"), + "parameter_schema_sha256": attestation.get("parameter_schema_sha256"), + "policy": certification.get("policy"), + "policy_evidence_sha256": certification.get("evidence_sha256"), + "run_report_sha256": replay.get("report_sha256"), + "governed_authorization_template_sha256": attestation.get( + "governed_authorization_template_sha256" + ), + } try: with archive_path.open("rb") as fh: resp = httpx.post( @@ -1343,8 +1407,9 @@ def push( follow_redirects=False, ) except httpx.HTTPError as exc: - raise HostedError( - f"Upload to {resolved_host}/api/ingest failed: {exc}" + raise HostedDeliveryUncertain( + f"Upload to {resolved_host}/api/ingest failed: {exc}", + context=local_result, ) from exc if resp.status_code == 401: raise HostedError("Ingest token was rejected (401).") @@ -1352,14 +1417,26 @@ def push( raise HostedError( f"Ingest returned {resp.status_code} (expected 201): {_body_snippet(resp)}" ) - payload = resp.json() - ingest = payload.get("ingest", payload) if isinstance(payload, dict) else {} + try: + payload = resp.json() + except ValueError as exc: + raise HostedDeliveryUncertain( + "Ingest returned 201 without a valid JSON acknowledgment", + context=local_result, + ) from exc + ingest = payload.get("ingest", payload) if isinstance(payload, dict) else None + if not isinstance(ingest, dict): + raise HostedDeliveryUncertain( + "Ingest returned 201 without a valid result object", + context=local_result, + ) workflow_id = ingest.get("workflow_id") result = dict(ingest) + # Never trust a dashboard URL supplied by the remote response. Construct it + # only from the already validated destination origin. + result.pop("dashboard_url", None) result["uploaded"] = True - result["sanitization"] = ingest_manifest - result["approval"] = approval - result["destination_kind"] = destination.kind + result.update(local_result) if workflow_id: result["dashboard_url"] = f"{resolved_host}/dashboard/workflows/{workflow_id}" return result diff --git a/public-artifacts.json b/public-artifacts.json index 031d5883..519144a2 100644 --- a/public-artifacts.json +++ b/public-artifacts.json @@ -5727,6 +5727,10 @@ "path": "schemas/public-demo-evidence-v1.json", "sha256": "fc7bca6e1ea0cfc9c70c11368d7b59d542f75f98f2dbaf4d869247225e53c1fd" }, + { + "path": "schemas/push-result-v1.json", + "sha256": "67e198e2bb6c7a503b60d75bc2adf862c771e650719872d6472b70be9a07693c" + }, { "path": "schemas/runtime-validation-attestation-v1.json", "sha256": "5168eb014c8559d5e722fdd3af4fc2cf394d0c1ae065e1176075c029881cce7c" diff --git a/schemas/push-result-v1.json b/schemas/push-result-v1.json new file mode 100644 index 00000000..f7d958b4 --- /dev/null +++ b/schemas/push-result-v1.json @@ -0,0 +1,260 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://openadapt.ai/schemas/push-result-v1.json", + "title": "OpenAdapt push result V1", + "type": "object", + "additionalProperties": false, + "required": [ + "schema", + "status", + "workflow_id", + "artifact_ingest_id", + "review", + "attestation", + "binding", + "next_action", + "dashboard_url", + "delivery", + "error" + ], + "properties": { + "schema": { "const": "openadapt.push-result/v1" }, + "status": { + "enum": [ + "paused_for_review", + "accepted_for_ingest", + "failed", + "delivery_uncertain" + ] + }, + "workflow_id": { "$ref": "#/$defs/nullableUuid" }, + "artifact_ingest_id": { "$ref": "#/$defs/nullableUuid" }, + "review": { + "oneOf": [ + { "type": "null" }, + { + "type": "object", + "additionalProperties": false, + "required": ["id", "scope", "sanitized_path", "command"], + "properties": { + "id": { "$ref": "#/$defs/sha256" }, + "scope": { "const": "local_non_authoritative" }, + "sanitized_path": { "type": ["string", "null"], "minLength": 1 }, + "command": { "type": ["string", "null"], "minLength": 1 } + } + } + ] + }, + "attestation": { + "oneOf": [ + { "type": "null" }, + { + "type": "object", + "additionalProperties": false, + "required": ["id", "schema"], + "properties": { + "id": { "type": "string", "minLength": 1, "maxLength": 200 }, + "schema": { + "type": "string", + "pattern": "^openadapt\\.runtime-validation/v[0-9]+$" + } + } + } + ] + }, + "binding": { + "type": "object", + "additionalProperties": false, + "required": [ + "kind", + "source_tree_sha256", + "derivative_tree_sha256", + "approved_archive_sha256", + "artifact_sha256", + "bundle_sha256", + "source_recording_sha256", + "sanitization_policy", + "certification_policy", + "certification_evidence_sha256", + "governed_authorization_template_sha256", + "parameter_schema_sha256", + "attested_run_report_sha256", + "resolves_run_id" + ], + "properties": { + "kind": { "enum": ["recording", "bundle", null] }, + "source_tree_sha256": { "$ref": "#/$defs/nullableSha256" }, + "derivative_tree_sha256": { "$ref": "#/$defs/nullableSha256" }, + "approved_archive_sha256": { "$ref": "#/$defs/nullableSha256" }, + "artifact_sha256": { "$ref": "#/$defs/nullableSha256" }, + "bundle_sha256": { "$ref": "#/$defs/nullableSha256" }, + "source_recording_sha256": { "$ref": "#/$defs/nullableSha256" }, + "sanitization_policy": { "type": ["string", "null"], "minLength": 1 }, + "certification_policy": { "type": ["string", "null"], "minLength": 1 }, + "certification_evidence_sha256": { "$ref": "#/$defs/nullableSha256" }, + "governed_authorization_template_sha256": { + "$ref": "#/$defs/nullableSha256" + }, + "parameter_schema_sha256": { "$ref": "#/$defs/nullableSha256" }, + "attested_run_report_sha256": { "$ref": "#/$defs/nullableSha256" }, + "resolves_run_id": { "$ref": "#/$defs/nullableUuid" } + } + }, + "next_action": { + "enum": [ + "review_local", + "parameterize", + "validate_runtime", + "open_dashboard", + "reconcile", + null + ] + }, + "dashboard_url": { + "oneOf": [{ "type": "null" }, { "type": "string", "format": "uri" }] + }, + "delivery": { + "type": "object", + "additionalProperties": false, + "required": ["attempted", "certainty"], + "properties": { + "attempted": { "type": ["boolean", "null"] }, + "certainty": { + "enum": ["not_attempted", "not_accepted", "accepted", "unknown"] + } + } + }, + "error": { + "oneOf": [ + { "type": "null" }, + { + "type": "object", + "additionalProperties": false, + "required": ["code", "message"], + "properties": { + "code": { + "enum": [ + "push_failed", + "delivery_uncertain", + "invalid_ingest_response" + ] + }, + "message": { "type": "string", "minLength": 1, "maxLength": 500 } + } + } + ] + } + }, + "$defs": { + "sha256": { "type": "string", "pattern": "^[a-f0-9]{64}$" }, + "nullableSha256": { + "oneOf": [{ "type": "null" }, { "$ref": "#/$defs/sha256" }] + }, + "nullableUuid": { + "oneOf": [ + { "type": "null" }, + { "$ref": "#/$defs/uuid" } + ] + }, + "uuid": { + "type": "string", + "pattern": "^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$" + } + }, + "allOf": [ + { + "if": { "properties": { "status": { "const": "paused_for_review" } } }, + "then": { + "properties": { + "workflow_id": { "type": "null" }, + "artifact_ingest_id": { "type": "null" }, + "review": { "type": "object" }, + "attestation": { "type": "null" }, + "next_action": { "const": "review_local" }, + "error": { "type": "null" }, + "delivery": { + "properties": { + "attempted": { "const": false }, + "certainty": { "const": "not_attempted" } + } + } + } + } + }, + { + "if": { "properties": { "status": { "const": "accepted_for_ingest" } } }, + "then": { + "properties": { + "artifact_ingest_id": { "$ref": "#/$defs/uuid" }, + "review": { "type": "object" }, + "error": { "type": "null" }, + "delivery": { + "properties": { + "attempted": { "const": true }, + "certainty": { "const": "accepted" } + } + } + } + } + }, + { + "if": { + "properties": { + "status": { "const": "accepted_for_ingest" }, + "binding": { "properties": { "kind": { "const": "recording" } } } + } + }, + "then": { + "properties": { + "workflow_id": { "type": "null" }, + "attestation": { "type": "null" }, + "next_action": { "enum": ["parameterize", "validate_runtime"] }, + "dashboard_url": { "type": "null" } + } + } + }, + { + "if": { + "properties": { + "status": { "const": "accepted_for_ingest" }, + "binding": { "properties": { "kind": { "const": "bundle" } } } + } + }, + "then": { + "properties": { + "workflow_id": { "$ref": "#/$defs/uuid" }, + "attestation": { "type": "object" }, + "next_action": { "const": "open_dashboard" }, + "dashboard_url": { "type": "string", "format": "uri" }, + "binding": { + "properties": { + "bundle_sha256": { "$ref": "#/$defs/sha256" }, + "source_recording_sha256": { "$ref": "#/$defs/sha256" }, + "certification_policy": { "type": "string", "minLength": 1 }, + "certification_evidence_sha256": { "$ref": "#/$defs/sha256" }, + "parameter_schema_sha256": { "$ref": "#/$defs/sha256" }, + "attested_run_report_sha256": { "$ref": "#/$defs/sha256" } + } + } + } + } + }, + { + "if": { "properties": { "status": { "const": "delivery_uncertain" } } }, + "then": { + "properties": { + "next_action": { "const": "reconcile" }, + "delivery": { + "properties": { + "attempted": { "const": true }, + "certainty": { "const": "unknown" } + } + }, + "error": { + "properties": { "code": { "const": "delivery_uncertain" } } + } + } + } + } + ] +} diff --git a/tests/test_hosted.py b/tests/test_hosted.py index 67af94ff..df43078d 100644 --- a/tests/test_hosted.py +++ b/tests/test_hosted.py @@ -18,6 +18,7 @@ from uuid import UUID import httpx +import jsonschema import pytest from openadapt_flow import hosted, privacy @@ -944,6 +945,38 @@ def test_push_401(tmp_path, monkeypatch): hosted.push(rec, token="tok", host="https://h.test") +def test_push_transport_error_has_uncertain_delivery_type(tmp_path, monkeypatch): + rec = _make_recording(tmp_path, "rec") + privacy.set_text_scrubber(_FakeScrubber()) + + def fail(*args, **kwargs): + raise httpx.ReadTimeout("response timed out") + + monkeypatch.setattr(httpx, "post", fail) + with pytest.raises(hosted.HostedDeliveryUncertain) as raised: + hosted.push(rec, token="tok", host="https://h.test") + assert ( + raised.value.context["sanitization"]["artifact"]["sha256"] + == (raised.value.context["local_binding"]["approved_archive_sha256"]) + ) + + +def test_push_201_without_json_acknowledgment_is_delivery_uncertain( + tmp_path, monkeypatch +): + rec = _make_recording(tmp_path, "rec") + privacy.set_text_scrubber(_FakeScrubber()) + monkeypatch.setattr( + httpx, + "post", + lambda *args, **kwargs: httpx.Response(201, text="not-json"), + ) + + with pytest.raises(hosted.HostedDeliveryUncertain) as raised: + hosted.push(rec, token="tok", host="https://h.test") + assert raised.value.context["local_binding"]["approved_archive_sha256"] + + def test_push_bundle_requires_verified_sanitization_not_attestation( tmp_path, monkeypatch ): @@ -2833,6 +2866,232 @@ def fake_push( assert "Dashboard" in out +_PUSH_ARTIFACT_SHA = "a" * 64 +_PUSH_SOURCE_SHA = "b" * 64 +_PUSH_DERIVATIVE_SHA = "c" * 64 +_PUSH_REVIEW_ID = "d" * 64 +_PUSH_POLICY_SHA = "e" * 64 +_PUSH_PARAMETER_SHA = "f" * 64 +_PUSH_REPORT_SHA = "1" * 64 +_PUSH_RECORDING_SHA = "2" * 64 +_PUSH_TEMPLATE_SHA = "3" * 64 +_PUSH_INGEST_ID = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" +_PUSH_WORKFLOW_ID = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + + +def _assert_push_json_schema(value): + schema = json.loads( + ( + Path(__file__).resolve().parents[1] / "schemas" / "push-result-v1.json" + ).read_text(encoding="utf-8") + ) + jsonschema.Draft202012Validator(schema).validate(value) + + +def _json_push_base(*, kind="recording"): + return { + "uploaded": True, + "kind": kind, + "workflow_id": None, + "artifact_ingest_id": _PUSH_INGEST_ID, + "artifact_sha256": _PUSH_ARTIFACT_SHA, + "sanitization": {"artifact": {"kind": kind, "sha256": _PUSH_ARTIFACT_SHA}}, + "local_binding": { + "source_tree_sha256": _PUSH_SOURCE_SHA, + "derivative_tree_sha256": _PUSH_DERIVATIVE_SHA, + "approved_archive_sha256": _PUSH_ARTIFACT_SHA, + "sanitization_policy": "outbound-phi-v1", + }, + "review_id": _PUSH_REVIEW_ID, + "destination_host": "https://h.test", + } + + +def test_push_parser_accepts_json_without_changing_other_arguments(): + args = build_parser().parse_args(["push", "approved", "--kind", "bundle", "--json"]) + assert args.path == "approved" + assert args.kind == "bundle" + assert args.json is True + assert args.func.__name__ == "_cmd_push" + + +def test_cli_push_json_paused_for_review(monkeypatch, capsys): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: { + "uploaded": False, + "pending_review": True, + "kind": "recording", + "sanitized_path": "/safe/derivative", + "review_command": "openadapt-flow review-sanitized /safe/derivative", + "review_id": _PUSH_REVIEW_ID, + "local_binding": { + "source_tree_sha256": _PUSH_SOURCE_SHA, + "derivative_tree_sha256": _PUSH_DERIVATIVE_SHA, + "approved_archive_sha256": None, + "sanitization_policy": "outbound-phi-v1", + }, + }, + ) + + assert main(["push", "raw", "--json"]) == 0 + value = json.loads(capsys.readouterr().out) + _assert_push_json_schema(value) + assert value["schema"] == "openadapt.push-result/v1" + assert value["status"] == "paused_for_review" + assert value["workflow_id"] is None + assert value["artifact_ingest_id"] is None + assert value["review"] == { + "id": _PUSH_REVIEW_ID, + "scope": "local_non_authoritative", + "sanitized_path": "/safe/derivative", + "command": "openadapt-flow review-sanitized /safe/derivative", + } + assert value["binding"]["source_tree_sha256"] == _PUSH_SOURCE_SHA + assert value["binding"]["approved_archive_sha256"] is None + assert value["next_action"] == "review_local" + assert value["delivery"] == { + "attempted": False, + "certainty": "not_attempted", + } + assert value["error"] is None + + +@pytest.mark.parametrize( + ("server_status", "next_action"), + [ + ("needs_parameterization", "parameterize"), + ("needs_runtime_validation", "validate_runtime"), + ], +) +def test_cli_push_json_recording_accepted( + monkeypatch, capsys, server_status, next_action +): + result = _json_push_base() + result["status"] = server_status + monkeypatch.setattr(hosted, "push", lambda *args, **kwargs: result) + + assert main(["push", "approved", "--json"]) == 0 + value = json.loads(capsys.readouterr().out) + _assert_push_json_schema(value) + assert value["status"] == "accepted_for_ingest" + assert value["workflow_id"] is None + assert value["artifact_ingest_id"] == _PUSH_INGEST_ID + assert value["next_action"] == next_action + assert value["binding"]["artifact_sha256"] == _PUSH_ARTIFACT_SHA + assert value["delivery"] == {"attempted": True, "certainty": "accepted"} + + +def test_cli_push_json_bundle_accepted_with_exact_attestation_binding( + monkeypatch, capsys +): + result = _json_push_base(kind="bundle") + result.update( + { + "workflow_id": _PUSH_WORKFLOW_ID, + "dashboard_url": ( + f"https://h.test/dashboard/workflows/{_PUSH_WORKFLOW_ID}" + ), + "resolves_run_id": "cccccccc-cccc-4ccc-8ccc-cccccccccccc", + "attestation_binding": { + "schema": "openadapt.runtime-validation/v3", + "challenge_id": "challenge-7", + "source_recording_sha256": _PUSH_RECORDING_SHA, + "bundle_sha256": _PUSH_ARTIFACT_SHA, + "parameter_schema_sha256": _PUSH_PARAMETER_SHA, + "policy": "clinical-write", + "policy_evidence_sha256": _PUSH_POLICY_SHA, + "run_report_sha256": _PUSH_REPORT_SHA, + "governed_authorization_template_sha256": _PUSH_TEMPLATE_SHA, + }, + } + ) + monkeypatch.setattr(hosted, "push", lambda *args, **kwargs: result) + + assert main(["push", "approved", "--kind", "bundle", "--json"]) == 0 + value = json.loads(capsys.readouterr().out) + _assert_push_json_schema(value) + assert value["status"] == "accepted_for_ingest" + assert value["workflow_id"] == _PUSH_WORKFLOW_ID + assert value["artifact_ingest_id"] == _PUSH_INGEST_ID + assert value["attestation"] == { + "id": "challenge-7", + "schema": "openadapt.runtime-validation/v3", + } + assert value["binding"] == { + "kind": "bundle", + "source_tree_sha256": _PUSH_SOURCE_SHA, + "derivative_tree_sha256": _PUSH_DERIVATIVE_SHA, + "approved_archive_sha256": _PUSH_ARTIFACT_SHA, + "artifact_sha256": _PUSH_ARTIFACT_SHA, + "bundle_sha256": _PUSH_ARTIFACT_SHA, + "source_recording_sha256": _PUSH_RECORDING_SHA, + "sanitization_policy": "outbound-phi-v1", + "certification_policy": "clinical-write", + "certification_evidence_sha256": _PUSH_POLICY_SHA, + "governed_authorization_template_sha256": _PUSH_TEMPLATE_SHA, + "parameter_schema_sha256": _PUSH_PARAMETER_SHA, + "attested_run_report_sha256": _PUSH_REPORT_SHA, + "resolves_run_id": "cccccccc-cccc-4ccc-8ccc-cccccccccccc", + } + assert value["next_action"] == "open_dashboard" + + +def test_cli_push_json_refuses_false_success_without_server_ingest_id( + monkeypatch, capsys +): + result = _json_push_base() + result["status"] = "needs_parameterization" + result["artifact_ingest_id"] = None + monkeypatch.setattr(hosted, "push", lambda *args, **kwargs: result) + + assert main(["push", "approved", "--json"]) == 1 + value = json.loads(capsys.readouterr().out) + _assert_push_json_schema(value) + assert value["status"] == "failed" + assert value["error"]["code"] == "invalid_ingest_response" + assert value["delivery"] == {"attempted": True, "certainty": "unknown"} + assert value["next_action"] == "reconcile" + + +def test_cli_push_json_transport_error_is_delivery_uncertain(monkeypatch, capsys): + def fail(*args, **kwargs): + raise hosted.HostedDeliveryUncertain( + "secret token and /private/source must not reach JSON", + context=_json_push_base(), + ) + + monkeypatch.setattr(hosted, "push", fail) + assert main(["push", "raw", "--json"]) == 1 + value = json.loads(capsys.readouterr().out) + _assert_push_json_schema(value) + assert value["status"] == "delivery_uncertain" + assert value["next_action"] == "reconcile" + assert value["delivery"] == {"attempted": True, "certainty": "unknown"} + assert value["artifact_ingest_id"] is None + assert value["binding"]["artifact_sha256"] == _PUSH_ARTIFACT_SHA + assert "secret" not in value["error"]["message"] + assert "/private/source" not in value["error"]["message"] + assert len(value["error"]["message"]) <= 500 + + +def test_cli_push_json_preflight_error_is_bounded_and_nonzero(monkeypatch, capsys): + def fail(*args, **kwargs): + raise hosted.HostedError("/private/source " + "x" * 1000) + + monkeypatch.setattr(hosted, "push", fail) + assert main(["push", "raw", "--json"]) == 1 + value = json.loads(capsys.readouterr().out) + _assert_push_json_schema(value) + assert value["status"] == "failed" + assert value["error"] == { + "code": "push_failed", + "message": "The artifact was not accepted for ingest.", + } + assert value["delivery"] == {"attempted": None, "certainty": "not_accepted"} + + def test_cli_report_break_dispatch(monkeypatch, capsys): captured: dict = {} diff --git a/tests/test_sanitized_artifact.py b/tests/test_sanitized_artifact.py index 82fec3ac..899727e9 100644 --- a/tests/test_sanitized_artifact.py +++ b/tests/test_sanitized_artifact.py @@ -822,6 +822,14 @@ def test_raw_push_creates_derivative_and_pauses_for_review(tmp_path, monkeypatch ) assert result["pending_review"] is True + assert re.fullmatch(r"[a-f0-9]{64}", result["review_id"]) + assert result["review_id"] != result["local_binding"]["derivative_tree_sha256"] + assert result["local_binding"] == { + "source_tree_sha256": result["local_binding"]["source_tree_sha256"], + "derivative_tree_sha256": result["local_binding"]["derivative_tree_sha256"], + "approved_archive_sha256": None, + "sanitization_policy": "outbound-phi-v1", + } assert Path(result["sanitized_path"], MANIFEST_NAME).is_file() assert source.name not in Path(result["sanitized_path"]).name again = hosted.push( @@ -834,6 +842,7 @@ def test_raw_push_creates_derivative_and_pauses_for_review(tmp_path, monkeypatch token="token", ) assert again["sanitized_path"] == result["sanitized_path"] + assert again["review_id"] == result["review_id"] def test_changed_bundle_is_not_uploaded_as_executable(tmp_path, monkeypatch): From e2faca36f591b966e5f153ae799c8eeec025c0ac Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 18 Aug 2026 13:01:43 -0400 Subject: [PATCH 2/6] fix: bind structured push to retained server version --- docs/PUSH_JSON.md | 16 ++++-- openadapt_flow/__main__.py | 33 +++++++++++ openadapt_flow/hosted.py | 12 +++- public-artifacts.json | 2 +- schemas/push-result-v1.json | 80 +++++++++++++++++++++++++-- tests/test_hosted.py | 95 +++++++++++++++++++++++++++++++- tests/test_sanitized_artifact.py | 1 + 7 files changed, 226 insertions(+), 13 deletions(-) diff --git a/docs/PUSH_JSON.md b/docs/PUSH_JSON.md index 65ed8acf..502ba0a3 100644 --- a/docs/PUSH_JSON.md +++ b/docs/PUSH_JSON.md @@ -13,7 +13,7 @@ The command without `--json` keeps its existing human-readable output. | `paused_for_review` | 0 | Flow made a local sanitized derivative. It did not upload it. | `review_local` | | `accepted_for_ingest` | 0 | The server acknowledged the exact approved archive and returned its stable ingest id. | `parameterize`, `validate_runtime`, or `open_dashboard` | | `failed` | 1 | Flow did not receive a complete accepted-ingest contract. | `null` or `reconcile` | -| `delivery_uncertain` | 1 | A transport failure occurred after Flow attempted the request. The server can have received it. | `reconcile` | +| `delivery_uncertain` | 1 | A transport failure or an ambiguous server response occurred after Flow attempted the request. The server can have received it. | `reconcile` | Do not retry `delivery_uncertain` automatically. Use `artifact_sha256` to reconcile the request with the hosted control plane first. @@ -46,7 +46,9 @@ The `binding` object lets Desktop detect a stale handoff. It carries: - the sanitization and certification policies; - the certification evidence, parameter schema, governed authorization template, and attested run-report SHA-256 values; and -- the halted run UUID when the bundle resolves a governed halt. +- the halted run UUID when the bundle resolves a governed halt; and +- the server-retained organization, bundle-version, and runtime-validation + identifiers for an accepted bundle. The local `review.id` is a domain-separated SHA-256 of the canonical sanitized manifest. It is stable for that exact review candidate. It is local and @@ -88,7 +90,11 @@ openadapt-flow push recording/ --json "governed_authorization_template_sha256": null, "parameter_schema_sha256": null, "attested_run_report_sha256": null, - "resolves_run_id": null + "resolves_run_id": null, + "organization_id": null, + "bundle_version_id": null, + "bundle_version": null, + "runtime_validation_id": null }, "next_action": "review_local", "dashboard_url": null, @@ -105,7 +111,9 @@ openadapt-flow push approved-bundle/ --kind bundle \ ``` Flow returns `accepted_for_ingest` only after it checks all required server ids, -the echoed artifact hash, and the exact local attestation binding. +the echoed artifact hash, the exact local attestation binding, and the retained +server bundle-version record. The server record must bind its organization, +workflow, artifact hash, version number, and runtime-validation identifier. ## Error privacy diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index c4b5534f..e4a20968 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -3230,6 +3230,10 @@ def _push_json_base(status: str) -> dict[str, Any]: "parameter_schema_sha256": None, "attested_run_report_sha256": None, "resolves_run_id": None, + "organization_id": None, + "bundle_version_id": None, + "bundle_version": None, + "runtime_validation_id": None, }, "next_action": None, "dashboard_url": None, @@ -3492,6 +3496,31 @@ def _push_json_result(result: dict[str, Any]) -> dict[str, Any]: workflow_id = result.get("workflow_id") if not _is_uuid(workflow_id): raise ValueError("bundle ingest has no workflow id") + if result.get("status") != "accepted": + raise ValueError("bundle ingest is not accepted") + version = result.get("version") + if not isinstance(version, dict): + raise ValueError("bundle ingest has no retained version binding") + bundle_version_id = version.get("id") + organization_id = version.get("org_id") + version_workflow_id = version.get("workflow_id") + version_artifact_sha = version.get("artifact_sha256") + runtime_validation_id = version.get("runtime_validation_id") + version_number = version.get("version") + if not _is_uuid(bundle_version_id): + raise ValueError("invalid retained bundle version id") + if not _is_uuid(organization_id): + raise ValueError("invalid retained organization binding") + if version_workflow_id != workflow_id: + raise ValueError("retained bundle version workflow mismatch") + if version_artifact_sha != artifact_sha: + raise ValueError("retained bundle version artifact mismatch") + if not _is_uuid(runtime_validation_id): + raise ValueError("invalid retained runtime validation binding") + if not isinstance(version_number, int) or isinstance(version_number, bool): + raise ValueError("invalid retained bundle version") + if version_number < 1: + raise ValueError("invalid retained bundle version") attestation = result.get("attestation_binding") if not isinstance(attestation, dict): raise ValueError("bundle ingest has no attestation binding") @@ -3544,6 +3573,10 @@ def _push_json_result(result: dict[str, Any]) -> dict[str, Any]: "parameter_schema_sha256": parameter_schema_sha, "attested_run_report_sha256": run_report_sha, "resolves_run_id": resolves_run_id, + "organization_id": organization_id, + "bundle_version_id": bundle_version_id, + "bundle_version": version_number, + "runtime_validation_id": runtime_validation_id, } ) dashboard_url = result.get("dashboard_url") diff --git a/openadapt_flow/hosted.py b/openadapt_flow/hosted.py index 412e0160..2491b65d 100644 --- a/openadapt_flow/hosted.py +++ b/openadapt_flow/hosted.py @@ -1413,10 +1413,20 @@ def push( ) from exc if resp.status_code == 401: raise HostedError("Ingest token was rejected (401).") - if resp.status_code != 201: + if 400 <= resp.status_code < 500 and resp.status_code not in {408, 409}: raise HostedError( f"Ingest returned {resp.status_code} (expected 201): {_body_snippet(resp)}" ) + if resp.status_code != 201: + # A timeout/conflict, redirect, or server failure is not proof that the + # hosted side rejected the artifact. The request body can have reached + # ingest before the response failed. Keep the exact local binding so a + # controller can reconcile it, and never invite a blind retry. + raise HostedDeliveryUncertain( + f"Ingest returned {resp.status_code} without a trustworthy " + f"acceptance result: {_body_snippet(resp)}", + context=local_result, + ) try: payload = resp.json() except ValueError as exc: diff --git a/public-artifacts.json b/public-artifacts.json index 519144a2..98b4317a 100644 --- a/public-artifacts.json +++ b/public-artifacts.json @@ -5729,7 +5729,7 @@ }, { "path": "schemas/push-result-v1.json", - "sha256": "67e198e2bb6c7a503b60d75bc2adf862c771e650719872d6472b70be9a07693c" + "sha256": "e0e06bb922ecb9cbc53800bb376d69e040b82fcf4bfa00b899a285377e7efbd7" }, { "path": "schemas/runtime-validation-attestation-v1.json", diff --git a/schemas/push-result-v1.json b/schemas/push-result-v1.json index f7d958b4..f6691ef0 100644 --- a/schemas/push-result-v1.json +++ b/schemas/push-result-v1.json @@ -79,7 +79,11 @@ "governed_authorization_template_sha256", "parameter_schema_sha256", "attested_run_report_sha256", - "resolves_run_id" + "resolves_run_id", + "organization_id", + "bundle_version_id", + "bundle_version", + "runtime_validation_id" ], "properties": { "kind": { "enum": ["recording", "bundle", null] }, @@ -97,7 +101,16 @@ }, "parameter_schema_sha256": { "$ref": "#/$defs/nullableSha256" }, "attested_run_report_sha256": { "$ref": "#/$defs/nullableSha256" }, - "resolves_run_id": { "$ref": "#/$defs/nullableUuid" } + "resolves_run_id": { "$ref": "#/$defs/nullableUuid" }, + "organization_id": { "$ref": "#/$defs/nullableUuid" }, + "bundle_version_id": { "$ref": "#/$defs/nullableUuid" }, + "bundle_version": { + "oneOf": [ + { "type": "null" }, + { "type": "integer", "minimum": 1 } + ] + }, + "runtime_validation_id": { "$ref": "#/$defs/nullableUuid" } } }, "next_action": { @@ -168,10 +181,29 @@ "properties": { "workflow_id": { "type": "null" }, "artifact_ingest_id": { "type": "null" }, - "review": { "type": "object" }, + "review": { + "type": "object", + "properties": { + "sanitized_path": { "type": "string", "minLength": 1 }, + "command": { "type": "string", "minLength": 1 } + } + }, "attestation": { "type": "null" }, "next_action": { "const": "review_local" }, "error": { "type": "null" }, + "binding": { + "properties": { + "kind": { "enum": ["recording", "bundle"] }, + "source_tree_sha256": { "$ref": "#/$defs/sha256" }, + "derivative_tree_sha256": { "$ref": "#/$defs/sha256" }, + "approved_archive_sha256": { "type": "null" }, + "artifact_sha256": { "type": "null" }, + "organization_id": { "type": "null" }, + "bundle_version_id": { "type": "null" }, + "bundle_version": { "type": "null" }, + "runtime_validation_id": { "type": "null" } + } + }, "delivery": { "properties": { "attempted": { "const": false }, @@ -186,8 +218,24 @@ "then": { "properties": { "artifact_ingest_id": { "$ref": "#/$defs/uuid" }, - "review": { "type": "object" }, + "review": { + "type": "object", + "properties": { + "sanitized_path": { "type": "null" }, + "command": { "type": "null" } + } + }, "error": { "type": "null" }, + "binding": { + "properties": { + "kind": { "enum": ["recording", "bundle"] }, + "source_tree_sha256": { "$ref": "#/$defs/sha256" }, + "derivative_tree_sha256": { "$ref": "#/$defs/sha256" }, + "approved_archive_sha256": { "$ref": "#/$defs/sha256" }, + "artifact_sha256": { "$ref": "#/$defs/sha256" }, + "sanitization_policy": { "type": "string", "minLength": 1 } + } + }, "delivery": { "properties": { "attempted": { "const": true }, @@ -209,7 +257,23 @@ "workflow_id": { "type": "null" }, "attestation": { "type": "null" }, "next_action": { "enum": ["parameterize", "validate_runtime"] }, - "dashboard_url": { "type": "null" } + "dashboard_url": { "type": "null" }, + "binding": { + "properties": { + "bundle_sha256": { "type": "null" }, + "source_recording_sha256": { "type": "null" }, + "certification_policy": { "type": "null" }, + "certification_evidence_sha256": { "type": "null" }, + "governed_authorization_template_sha256": { "type": "null" }, + "parameter_schema_sha256": { "type": "null" }, + "attested_run_report_sha256": { "type": "null" }, + "resolves_run_id": { "type": "null" }, + "organization_id": { "type": "null" }, + "bundle_version_id": { "type": "null" }, + "bundle_version": { "type": "null" }, + "runtime_validation_id": { "type": "null" } + } + } } } }, @@ -233,7 +297,11 @@ "certification_policy": { "type": "string", "minLength": 1 }, "certification_evidence_sha256": { "$ref": "#/$defs/sha256" }, "parameter_schema_sha256": { "$ref": "#/$defs/sha256" }, - "attested_run_report_sha256": { "$ref": "#/$defs/sha256" } + "attested_run_report_sha256": { "$ref": "#/$defs/sha256" }, + "organization_id": { "$ref": "#/$defs/uuid" }, + "bundle_version_id": { "$ref": "#/$defs/uuid" }, + "bundle_version": { "type": "integer", "minimum": 1 }, + "runtime_validation_id": { "$ref": "#/$defs/uuid" } } } } diff --git a/tests/test_hosted.py b/tests/test_hosted.py index df43078d..124ef870 100644 --- a/tests/test_hosted.py +++ b/tests/test_hosted.py @@ -933,10 +933,36 @@ def test_push_non_201(tmp_path, monkeypatch): monkeypatch.setattr( httpx, "post", lambda url, **kw: httpx.Response(502, text="store down") ) - with pytest.raises(hosted.HostedError, match="502"): + with pytest.raises(hosted.HostedDeliveryUncertain, match="502"): hosted.push(rec, token="tok", host="https://h.test") +@pytest.mark.parametrize("status", [400, 403, 404, 413, 422, 429]) +def test_push_definite_client_rejection_is_not_delivery_uncertain( + tmp_path, monkeypatch, status +): + rec = _make_recording(tmp_path, "rec") + privacy.set_text_scrubber(_FakeScrubber()) + monkeypatch.setattr( + httpx, "post", lambda url, **kw: httpx.Response(status, text="rejected") + ) + with pytest.raises(hosted.HostedError, match=str(status)) as raised: + hosted.push(rec, token="tok", host="https://h.test") + assert not isinstance(raised.value, hosted.HostedDeliveryUncertain) + + +@pytest.mark.parametrize("status", [200, 302, 408, 409, 500, 502, 503]) +def test_push_ambiguous_response_is_delivery_uncertain(tmp_path, monkeypatch, status): + rec = _make_recording(tmp_path, "rec") + privacy.set_text_scrubber(_FakeScrubber()) + monkeypatch.setattr( + httpx, "post", lambda url, **kw: httpx.Response(status, text="ambiguous") + ) + with pytest.raises(hosted.HostedDeliveryUncertain, match=str(status)) as raised: + hosted.push(rec, token="tok", host="https://h.test") + assert raised.value.context["local_binding"]["approved_archive_sha256"] + + def test_push_401(tmp_path, monkeypatch): rec = _make_recording(tmp_path, "rec") privacy.set_text_scrubber(_FakeScrubber()) @@ -2877,6 +2903,9 @@ def fake_push( _PUSH_TEMPLATE_SHA = "3" * 64 _PUSH_INGEST_ID = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" _PUSH_WORKFLOW_ID = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" +_PUSH_ORG_ID = "dddddddd-dddd-4ddd-8ddd-dddddddddddd" +_PUSH_VERSION_ID = "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee" +_PUSH_RUNTIME_VALIDATION_ID = "ffffffff-ffff-4fff-8fff-ffffffffffff" def _assert_push_json_schema(value): @@ -3005,6 +3034,15 @@ def test_cli_push_json_bundle_accepted_with_exact_attestation_binding( "run_report_sha256": _PUSH_REPORT_SHA, "governed_authorization_template_sha256": _PUSH_TEMPLATE_SHA, }, + "status": "accepted", + "version": { + "id": _PUSH_VERSION_ID, + "org_id": _PUSH_ORG_ID, + "workflow_id": _PUSH_WORKFLOW_ID, + "version": 3, + "artifact_sha256": _PUSH_ARTIFACT_SHA, + "runtime_validation_id": _PUSH_RUNTIME_VALIDATION_ID, + }, } ) monkeypatch.setattr(hosted, "push", lambda *args, **kwargs: result) @@ -3034,10 +3072,65 @@ def test_cli_push_json_bundle_accepted_with_exact_attestation_binding( "parameter_schema_sha256": _PUSH_PARAMETER_SHA, "attested_run_report_sha256": _PUSH_REPORT_SHA, "resolves_run_id": "cccccccc-cccc-4ccc-8ccc-cccccccccccc", + "organization_id": _PUSH_ORG_ID, + "bundle_version_id": _PUSH_VERSION_ID, + "bundle_version": 3, + "runtime_validation_id": _PUSH_RUNTIME_VALIDATION_ID, } assert value["next_action"] == "open_dashboard" +@pytest.mark.parametrize( + ("field", "value"), + [ + ("org_id", "not-a-uuid"), + ("workflow_id", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"), + ("artifact_sha256", "9" * 64), + ("runtime_validation_id", None), + ], +) +def test_cli_push_json_refuses_mismatched_server_version_binding( + monkeypatch, capsys, field, value +): + result = _json_push_base(kind="bundle") + result.update( + { + "workflow_id": _PUSH_WORKFLOW_ID, + "dashboard_url": ( + f"https://h.test/dashboard/workflows/{_PUSH_WORKFLOW_ID}" + ), + "status": "accepted", + "version": { + "id": _PUSH_VERSION_ID, + "org_id": _PUSH_ORG_ID, + "workflow_id": _PUSH_WORKFLOW_ID, + "version": 3, + "artifact_sha256": _PUSH_ARTIFACT_SHA, + "runtime_validation_id": _PUSH_RUNTIME_VALIDATION_ID, + }, + "attestation_binding": { + "schema": "openadapt.runtime-validation/v3", + "challenge_id": "challenge-7", + "source_recording_sha256": _PUSH_RECORDING_SHA, + "bundle_sha256": _PUSH_ARTIFACT_SHA, + "parameter_schema_sha256": _PUSH_PARAMETER_SHA, + "policy": "clinical-write", + "policy_evidence_sha256": _PUSH_POLICY_SHA, + "run_report_sha256": _PUSH_REPORT_SHA, + "governed_authorization_template_sha256": _PUSH_TEMPLATE_SHA, + }, + } + ) + result["version"][field] = value + monkeypatch.setattr(hosted, "push", lambda *args, **kwargs: result) + + assert main(["push", "approved", "--kind", "bundle", "--json"]) == 1 + document = json.loads(capsys.readouterr().out) + _assert_push_json_schema(document) + assert document["status"] == "failed" + assert document["error"]["code"] == "invalid_ingest_response" + + def test_cli_push_json_refuses_false_success_without_server_ingest_id( monkeypatch, capsys ): diff --git a/tests/test_sanitized_artifact.py b/tests/test_sanitized_artifact.py index 899727e9..e97fe3e9 100644 --- a/tests/test_sanitized_artifact.py +++ b/tests/test_sanitized_artifact.py @@ -877,6 +877,7 @@ def test_schema_files_are_valid_json(): "runtime-validation-attestation-v1.json", "runtime-validation-attestation-v2.json", "runtime-validation-attestation-v3.json", + "push-result-v1.json", ): schema = json.loads((root / name).read_text()) assert schema["$schema"].endswith("2020-12/schema") From 7311f66d4b178bf00d41ee5917e5d8f7f6d8c9e9 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 18 Aug 2026 13:05:48 -0400 Subject: [PATCH 3/6] fix: reject contradictory push result states --- public-artifacts.json | 2 +- schemas/push-result-v1.json | 75 +++++++++++++++++++++++++++++++++++++ tests/test_hosted.py | 71 +++++++++++++++++++++++++++++++++-- 3 files changed, 144 insertions(+), 4 deletions(-) diff --git a/public-artifacts.json b/public-artifacts.json index 98b4317a..89382f63 100644 --- a/public-artifacts.json +++ b/public-artifacts.json @@ -5729,7 +5729,7 @@ }, { "path": "schemas/push-result-v1.json", - "sha256": "e0e06bb922ecb9cbc53800bb376d69e040b82fcf4bfa00b899a285377e7efbd7" + "sha256": "5a989065581c6e5cc09ebe386c88e192450ff8d5f898ad46cbeeffcb0b21e29f" }, { "path": "schemas/runtime-validation-attestation-v1.json", diff --git a/schemas/push-result-v1.json b/schemas/push-result-v1.json index f6691ef0..40e48ffe 100644 --- a/schemas/push-result-v1.json +++ b/schemas/push-result-v1.json @@ -307,11 +307,85 @@ } } }, + { + "if": { "properties": { "status": { "const": "failed" } } }, + "then": { + "properties": { + "workflow_id": { "type": "null" }, + "artifact_ingest_id": { "type": "null" }, + "review": { "type": "null" }, + "attestation": { "type": "null" }, + "next_action": { "enum": ["reconcile", null] }, + "dashboard_url": { "type": "null" }, + "delivery": { + "properties": { + "attempted": { "enum": [true, null] }, + "certainty": { "enum": ["not_accepted", "unknown"] } + } + }, + "error": { + "type": "object", + "properties": { + "code": { "enum": ["push_failed", "invalid_ingest_response"] } + } + } + } + } + }, + { + "if": { + "properties": { + "status": { "const": "failed" }, + "error": { + "type": "object", + "properties": { "code": { "const": "push_failed" } }, + "required": ["code"] + } + } + }, + "then": { + "properties": { + "next_action": { "type": "null" }, + "delivery": { + "properties": { + "attempted": { "type": "null" }, + "certainty": { "const": "not_accepted" } + } + } + } + } + }, + { + "if": { + "properties": { + "status": { "const": "failed" }, + "error": { + "type": "object", + "properties": { "code": { "const": "invalid_ingest_response" } }, + "required": ["code"] + } + } + }, + "then": { + "properties": { + "next_action": { "const": "reconcile" }, + "delivery": { + "properties": { + "attempted": { "const": true }, + "certainty": { "const": "unknown" } + } + } + } + } + }, { "if": { "properties": { "status": { "const": "delivery_uncertain" } } }, "then": { "properties": { + "workflow_id": { "type": "null" }, + "artifact_ingest_id": { "type": "null" }, "next_action": { "const": "reconcile" }, + "dashboard_url": { "type": "null" }, "delivery": { "properties": { "attempted": { "const": true }, @@ -319,6 +393,7 @@ } }, "error": { + "type": "object", "properties": { "code": { "const": "delivery_uncertain" } } } } diff --git a/tests/test_hosted.py b/tests/test_hosted.py index 124ef870..a35f2e01 100644 --- a/tests/test_hosted.py +++ b/tests/test_hosted.py @@ -2908,13 +2908,16 @@ def fake_push( _PUSH_RUNTIME_VALIDATION_ID = "ffffffff-ffff-4fff-8fff-ffffffffffff" -def _assert_push_json_schema(value): - schema = json.loads( +def _push_json_schema(): + return json.loads( ( Path(__file__).resolve().parents[1] / "schemas" / "push-result-v1.json" ).read_text(encoding="utf-8") ) - jsonschema.Draft202012Validator(schema).validate(value) + + +def _assert_push_json_schema(value): + jsonschema.Draft202012Validator(_push_json_schema()).validate(value) def _json_push_base(*, kind="recording"): @@ -3185,6 +3188,68 @@ def fail(*args, **kwargs): assert value["delivery"] == {"attempted": None, "certainty": "not_accepted"} +@pytest.mark.parametrize( + ("field", "value"), + [ + ("error", None), + ("next_action", "open_dashboard"), + ], +) +def test_push_json_schema_rejects_conflicting_failed_state( + monkeypatch, capsys, field, value +): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: (_ for _ in ()).throw(hosted.HostedError("private")), + ) + assert main(["push", "raw", "--json"]) == 1 + document = json.loads(capsys.readouterr().out) + document[field] = value + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + +def test_push_json_schema_rejects_accepted_certainty_on_failure(monkeypatch, capsys): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: (_ for _ in ()).throw(hosted.HostedError("private")), + ) + assert main(["push", "raw", "--json"]) == 1 + document = json.loads(capsys.readouterr().out) + document["delivery"] = {"attempted": True, "certainty": "accepted"} + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + +@pytest.mark.parametrize( + ("field", "value"), + [ + ("error", None), + ( + "error", + {"code": "push_failed", "message": "The upload is uncertain."}, + ), + ], +) +def test_push_json_schema_rejects_conflicting_uncertain_state( + monkeypatch, capsys, field, value +): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: (_ for _ in ()).throw( + hosted.HostedDeliveryUncertain("private") + ), + ) + assert main(["push", "raw", "--json"]) == 1 + document = json.loads(capsys.readouterr().out) + document[field] = value + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + def test_cli_report_break_dispatch(monkeypatch, capsys): captured: dict = {} From 8099a2273997d1ea35782cff0e042dd0af37890b Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 18 Aug 2026 13:10:49 -0400 Subject: [PATCH 4/6] fix: forbid server bindings on unsuccessful push --- public-artifacts.json | 2 +- schemas/push-result-v1.json | 16 ++++++++++++++++ tests/test_hosted.py | 30 ++++++++++++++++++++++++++++++ 3 files changed, 47 insertions(+), 1 deletion(-) diff --git a/public-artifacts.json b/public-artifacts.json index 89382f63..2d7b2531 100644 --- a/public-artifacts.json +++ b/public-artifacts.json @@ -5729,7 +5729,7 @@ }, { "path": "schemas/push-result-v1.json", - "sha256": "5a989065581c6e5cc09ebe386c88e192450ff8d5f898ad46cbeeffcb0b21e29f" + "sha256": "16b5db599da35a96825bb8ad8fd767cb9b933e679d7f95cfe8e3af6f73f4f514" }, { "path": "schemas/runtime-validation-attestation-v1.json", diff --git a/schemas/push-result-v1.json b/schemas/push-result-v1.json index 40e48ffe..c7e2b056 100644 --- a/schemas/push-result-v1.json +++ b/schemas/push-result-v1.json @@ -317,6 +317,14 @@ "attestation": { "type": "null" }, "next_action": { "enum": ["reconcile", null] }, "dashboard_url": { "type": "null" }, + "binding": { + "properties": { + "organization_id": { "type": "null" }, + "bundle_version_id": { "type": "null" }, + "bundle_version": { "type": "null" }, + "runtime_validation_id": { "type": "null" } + } + }, "delivery": { "properties": { "attempted": { "enum": [true, null] }, @@ -386,6 +394,14 @@ "artifact_ingest_id": { "type": "null" }, "next_action": { "const": "reconcile" }, "dashboard_url": { "type": "null" }, + "binding": { + "properties": { + "organization_id": { "type": "null" }, + "bundle_version_id": { "type": "null" }, + "bundle_version": { "type": "null" }, + "runtime_validation_id": { "type": "null" } + } + }, "delivery": { "properties": { "attempted": { "const": true }, diff --git a/tests/test_hosted.py b/tests/test_hosted.py index a35f2e01..51d09216 100644 --- a/tests/test_hosted.py +++ b/tests/test_hosted.py @@ -3223,6 +3223,19 @@ def test_push_json_schema_rejects_accepted_certainty_on_failure(monkeypatch, cap _assert_push_json_schema(document) +def test_push_json_schema_rejects_server_binding_on_failure(monkeypatch, capsys): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: (_ for _ in ()).throw(hosted.HostedError("private")), + ) + assert main(["push", "raw", "--json"]) == 1 + document = json.loads(capsys.readouterr().out) + document["binding"]["runtime_validation_id"] = _PUSH_RUNTIME_VALIDATION_ID + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + @pytest.mark.parametrize( ("field", "value"), [ @@ -3250,6 +3263,23 @@ def test_push_json_schema_rejects_conflicting_uncertain_state( _assert_push_json_schema(document) +def test_push_json_schema_rejects_server_binding_on_uncertain_state( + monkeypatch, capsys +): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: (_ for _ in ()).throw( + hosted.HostedDeliveryUncertain("private") + ), + ) + assert main(["push", "raw", "--json"]) == 1 + document = json.loads(capsys.readouterr().out) + document["binding"]["organization_id"] = _PUSH_ORG_ID + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + def test_cli_report_break_dispatch(monkeypatch, capsys): captured: dict = {} From 28b19afaf260bf7a99224b24ed3bad85f4cf6f5a Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 18 Aug 2026 13:11:46 -0400 Subject: [PATCH 5/6] fix: close paused push schema state --- public-artifacts.json | 2 +- schemas/push-result-v1.json | 1 + tests/test_hosted.py | 26 ++++++++++++++++++++++++++ 3 files changed, 28 insertions(+), 1 deletion(-) diff --git a/public-artifacts.json b/public-artifacts.json index 2d7b2531..74e076f3 100644 --- a/public-artifacts.json +++ b/public-artifacts.json @@ -5729,7 +5729,7 @@ }, { "path": "schemas/push-result-v1.json", - "sha256": "16b5db599da35a96825bb8ad8fd767cb9b933e679d7f95cfe8e3af6f73f4f514" + "sha256": "cdf50f23992c3b6a3509c409bc0992d1461ec1bb6043e3cdcf97c3314b60ade7" }, { "path": "schemas/runtime-validation-attestation-v1.json", diff --git a/schemas/push-result-v1.json b/schemas/push-result-v1.json index c7e2b056..0bee0ae5 100644 --- a/schemas/push-result-v1.json +++ b/schemas/push-result-v1.json @@ -190,6 +190,7 @@ }, "attestation": { "type": "null" }, "next_action": { "const": "review_local" }, + "dashboard_url": { "type": "null" }, "error": { "type": "null" }, "binding": { "properties": { diff --git a/tests/test_hosted.py b/tests/test_hosted.py index 51d09216..062e7001 100644 --- a/tests/test_hosted.py +++ b/tests/test_hosted.py @@ -2990,6 +2990,32 @@ def test_cli_push_json_paused_for_review(monkeypatch, capsys): assert value["error"] is None +def test_push_json_schema_rejects_dashboard_on_paused_review(monkeypatch, capsys): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: { + "uploaded": False, + "pending_review": True, + "kind": "recording", + "sanitized_path": "/safe/derivative", + "review_command": "openadapt-flow review-sanitized /safe/derivative", + "review_id": _PUSH_REVIEW_ID, + "local_binding": { + "source_tree_sha256": _PUSH_SOURCE_SHA, + "derivative_tree_sha256": _PUSH_DERIVATIVE_SHA, + "approved_archive_sha256": None, + "sanitization_policy": "outbound-phi-v1", + }, + }, + ) + assert main(["push", "raw", "--json"]) == 0 + document = json.loads(capsys.readouterr().out) + document["dashboard_url"] = "https://evil.example/runs/x" + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + @pytest.mark.parametrize( ("server_status", "next_action"), [ From 763079fb5ab4e2d3e9e00005f4561e4699c1bba9 Mon Sep 17 00:00:00 2001 From: Richard Abrich Date: Tue, 18 Aug 2026 17:06:16 -0400 Subject: [PATCH 6/6] fix: prove an exact accepted ingest before push reports success A 201 status code alone did not prove the hosted control plane retained the intended artifact. Require the complete server-owned identity chain in hosted.push() itself, so the human-readable command gets the same protection as --json: the server ingest id, the artifact kind and hash, a governed next action for a recording, and, for a bundle, the accepted status plus the exact retained version binding (id, org, workflow, artifact hash, resolved run, runtime-validation id, version number). Anything short of that is an uncertain delivery, never a success. Also: - Upload a verified private snapshot of the approved archive so a path swap between approval and egress cannot ship unapproved bytes. - Bind the accepted workflow to the requested --workflow-id, and the retained version's promoted_from_run_id to --resolves-run-id. - Replace the review "command" string with a typed review_sanitized action plus a separate original_path, so a controller never builds a shell command from a local path. The human-readable output still prints the review command. - Pin the accepted runtime-validation attestation schemas to v1, v2 and v3, and require the governed authorization template exactly on v3. - Accept canonical RFC 9562 UUID versions 1-8 for server-owned identifiers, matching the hosted control plane's own check and Flow's report-run regex. - Collapse the obsolete invalid_ingest_response failure code into delivery_uncertain, and carry the local reconciliation binding into it. - Report an uncertain delivery in human-readable mode with the artifact hash and an explicit do-not-retry instruction, without leaking server text. Verified against openadapt-cloud origin/main: /api/ingest returns ingest.artifact_ingest_id for accepted, deduplicated, and recording responses, and the accepted bundle response carries the retained version record. --- docs/PUSH_JSON.md | 23 ++- openadapt_flow/__main__.py | 171 +++++++++++----- openadapt_flow/hosted.py | 147 +++++++++++++- public-artifacts.json | 2 +- schemas/push-result-v1.json | 260 +++++++++++++++++++------ tests/test_hosted.py | 325 ++++++++++++++++++++++++++++--- tests/test_runtime_validation.py | 29 ++- tests/test_sanitized_artifact.py | 57 +++++- 8 files changed, 870 insertions(+), 144 deletions(-) diff --git a/docs/PUSH_JSON.md b/docs/PUSH_JSON.md index 502ba0a3..0c801fad 100644 --- a/docs/PUSH_JSON.md +++ b/docs/PUSH_JSON.md @@ -12,12 +12,17 @@ The command without `--json` keeps its existing human-readable output. |---|---:|---|---| | `paused_for_review` | 0 | Flow made a local sanitized derivative. It did not upload it. | `review_local` | | `accepted_for_ingest` | 0 | The server acknowledged the exact approved archive and returned its stable ingest id. | `parameterize`, `validate_runtime`, or `open_dashboard` | -| `failed` | 1 | Flow did not receive a complete accepted-ingest contract. | `null` or `reconcile` | -| `delivery_uncertain` | 1 | A transport failure or an ambiguous server response occurred after Flow attempted the request. The server can have received it. | `reconcile` | +| `failed` | 1 | Flow rejected the request before it had proof of an upload attempt or received a definite rejection. | `null` | +| `delivery_uncertain` | 1 | A transport failure or an incomplete or ambiguous server response occurred after Flow attempted the request. The server can have received it. | `reconcile` | Do not retry `delivery_uncertain` automatically. Use `artifact_sha256` to reconcile the request with the hosted control plane first. +A `201` status is not sufficient proof. Flow also requires the exact server +ingest id, artifact kind and hash, governed next action, and, for a bundle, the +complete retained version identity chain. An incomplete or contradictory `201` +response is `delivery_uncertain` in both JSON and human-readable modes. + ## Stable fields V1 always includes these top-level keys: @@ -38,6 +43,10 @@ next action is parameterization or runtime validation. An accepted bundle has a workflow UUID, the server ingest UUID, a same-origin dashboard URL, and the exact local runtime-attestation binding. +Push-result V1 accepts runtime-validation attestation schemas V1, V2, and V3. +An unrecognized attestation schema is an uncertain delivery, not an accepted +handoff. + The `binding` object lets Desktop detect a stale handoff. It carries: - the source tree, derivative tree, approved archive, and acknowledged artifact @@ -55,6 +64,11 @@ manifest. It is stable for that exact review candidate. It is local and non-authoritative. It is not a hosted approval id. The attestation `id` is the server challenge id that the local runtime-validation attestation signs. +For `paused_for_review`, `review.action` is the typed local action +`review_sanitized`. `sanitized_path` and `original_path` are separate values. +A controller must pass them as process arguments. It must not construct or run +a shell command from either path. + ## Examples A raw recording normally pauses locally: @@ -73,7 +87,8 @@ openadapt-flow push recording/ --json "id": "", "scope": "local_non_authoritative", "sanitized_path": "", - "command": "openadapt-flow review-sanitized --original " + "action": "review_sanitized", + "original_path": "" }, "attestation": null, "binding": { @@ -114,6 +129,8 @@ Flow returns `accepted_for_ingest` only after it checks all required server ids, the echoed artifact hash, the exact local attestation binding, and the retained server bundle-version record. The server record must bind its organization, workflow, artifact hash, version number, and runtime-validation identifier. +For a replacement, the retained version must also bind the exact halted run. +When the request does not resolve a halt, that retained field must be `null`. ## Error privacy diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index e4a20968..68e1b17b 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -3204,6 +3204,13 @@ def _cmd_connect(args: argparse.Namespace) -> int: _PUSH_JSON_SCHEMA = "openadapt.push-result/v1" +_PUSH_RUNTIME_ATTESTATION_SCHEMAS = frozenset( + { + "openadapt.runtime-validation/v1", + "openadapt.runtime-validation/v2", + "openadapt.runtime-validation/v3", + } +) def _push_json_base(status: str) -> dict[str, Any]: @@ -3254,18 +3261,14 @@ def _is_uuid(value: Any) -> bool: if not isinstance(value, str): return False try: - return str(UUID(value)) == value + parsed = UUID(value) + return str(parsed) == value and parsed.version in {1, 2, 3, 4, 5, 6, 7, 8} except ValueError: return False def _is_runtime_attestation_schema(value: Any) -> bool: - prefix = "openadapt.runtime-validation/v" - return ( - isinstance(value, str) - and value.startswith(prefix) - and value.removeprefix(prefix).isdigit() - ) + return isinstance(value, str) and value in _PUSH_RUNTIME_ATTESTATION_SCHEMAS def _required_sha256(value: Any, *, field: str) -> str: @@ -3337,22 +3340,14 @@ def _push_json_failure( "id": review_id, "scope": "local_non_authoritative", "sanitized_path": None, - "command": None, + "action": None, + "original_path": None, } attestation = context.get("attestation_binding") if isinstance(attestation, dict): challenge_id = attestation.get("challenge_id") attestation_schema = attestation.get("schema") - if ( - isinstance(challenge_id, str) - and 1 <= len(challenge_id) <= 200 - and _is_runtime_attestation_schema(attestation_schema) - ): - document["attestation"] = { - "id": challenge_id, - "schema": attestation_schema, - } - optional_binding = { + bundle_binding = { "bundle_sha256": attestation.get("bundle_sha256"), "source_recording_sha256": attestation.get( "source_recording_sha256" @@ -3371,26 +3366,56 @@ def _push_json_failure( "run_report_sha256" ), } - for key, value in optional_binding.items(): - if key == "certification_policy": - if isinstance(value, str) and value: - binding[key] = value - elif value is None or _is_sha256(value): - binding[key] = value + template_sha = bundle_binding[ + "governed_authorization_template_sha256" + ] + has_complete_binding = ( + isinstance(challenge_id, str) + and 1 <= len(challenge_id) <= 200 + and _is_runtime_attestation_schema(attestation_schema) + and bundle_binding["bundle_sha256"] == values["artifact_sha256"] + and all( + _is_sha256(bundle_binding[key]) + for key in ( + "bundle_sha256", + "source_recording_sha256", + "certification_evidence_sha256", + "parameter_schema_sha256", + "attested_run_report_sha256", + ) + ) + and isinstance(bundle_binding["certification_policy"], str) + and bool(bundle_binding["certification_policy"]) + and ( + ( + attestation_schema == "openadapt.runtime-validation/v3" + and _is_sha256(template_sha) + ) + or ( + attestation_schema + in { + "openadapt.runtime-validation/v1", + "openadapt.runtime-validation/v2", + } + and template_sha is None + ) + ) + ) + if has_complete_binding: + document["attestation"] = { + "id": challenge_id, + "schema": attestation_schema, + } + binding.update(bundle_binding) return document -def _push_json_invalid_response() -> dict[str, Any]: - document = _push_json_base("failed") - document["next_action"] = "reconcile" - document["delivery"] = {"attempted": True, "certainty": "unknown"} - document["error"] = { - "code": "invalid_ingest_response", - "message": ( - "The server response did not prove an exact accepted ingest. " - "Reconcile the artifact in the hosted control plane before any retry." - ), - } +def _push_json_invalid_response(context: dict[str, Any]) -> dict[str, Any]: + document = _push_json_failure(uncertain=True, context=context) + document["error"]["message"] = ( + "The server response did not prove an exact accepted ingest. " + "Reconcile the artifact in the hosted control plane before any retry." + ) return document @@ -3416,17 +3441,21 @@ def _push_json_result(result: dict[str, Any]) -> dict[str, Any]: if kind not in ("recording", "bundle"): raise ValueError("invalid review kind") sanitized_path = result.get("sanitized_path") - review_command = result.get("review_command") + review_action = result.get("review_action") + original_path = result.get("original_path") if not isinstance(sanitized_path, str) or not sanitized_path: raise ValueError("missing sanitized review path") - if not isinstance(review_command, str) or not review_command: - raise ValueError("missing review command") + if review_action != "review_sanitized": + raise ValueError("missing review action") + if not isinstance(original_path, str) or not original_path: + raise ValueError("missing original review path") document = _push_json_base("paused_for_review") document["review"] = { "id": review_id, "scope": "local_non_authoritative", "sanitized_path": sanitized_path, - "command": review_command, + "action": review_action, + "original_path": original_path, } document["binding"].update( { @@ -3467,7 +3496,8 @@ def _push_json_result(result: dict[str, Any]) -> dict[str, Any]: "id": review_id, "scope": "local_non_authoritative", "sanitized_path": None, - "command": None, + "action": None, + "original_path": None, } document["binding"].update( { @@ -3496,6 +3526,11 @@ def _push_json_result(result: dict[str, Any]) -> dict[str, Any]: workflow_id = result.get("workflow_id") if not _is_uuid(workflow_id): raise ValueError("bundle ingest has no workflow id") + requested_workflow_id = result.get("requested_workflow_id") + if requested_workflow_id is not None and not _is_uuid(requested_workflow_id): + raise ValueError("invalid requested workflow binding") + if requested_workflow_id is not None and workflow_id != requested_workflow_id: + raise ValueError("accepted workflow does not match the requested workflow") if result.get("status") != "accepted": raise ValueError("bundle ingest is not accepted") version = result.get("version") @@ -3506,6 +3541,7 @@ def _push_json_result(result: dict[str, Any]) -> dict[str, Any]: version_workflow_id = version.get("workflow_id") version_artifact_sha = version.get("artifact_sha256") runtime_validation_id = version.get("runtime_validation_id") + version_resolves_run_id = version.get("promoted_from_run_id") version_number = version.get("version") if not _is_uuid(bundle_version_id): raise ValueError("invalid retained bundle version id") @@ -3558,9 +3594,22 @@ def _push_json_result(result: dict[str, Any]) -> dict[str, Any]: template_sha = _required_sha256( template_sha, field="governed authorization template binding" ) + if schema == "openadapt.runtime-validation/v3" and template_sha is None: + raise ValueError("v3 attestation has no governed authorization template") + if ( + schema + in { + "openadapt.runtime-validation/v1", + "openadapt.runtime-validation/v2", + } + and template_sha is not None + ): + raise ValueError("legacy attestation has an unexpected template binding") resolves_run_id = result.get("resolves_run_id") if resolves_run_id is not None and not _is_uuid(resolves_run_id): raise ValueError("invalid resolved run binding") + if version_resolves_run_id != resolves_run_id: + raise ValueError("retained resolved run binding mismatch") document["workflow_id"] = workflow_id document["attestation"] = {"id": challenge_id, "schema": schema} document["binding"].update( @@ -3630,7 +3679,15 @@ def _cmd_push(args: argparse.Namespace) -> int: if args.json: _print_push_json(_push_json_failure(uncertain=True, context=e.context)) return 1 - print(f"push failed: {e}") + reconciliation = _push_json_failure(uncertain=True, context=e.context) + artifact_sha256 = reconciliation["binding"]["artifact_sha256"] + print("Push delivery is uncertain. The server can have received the artifact.") + if artifact_sha256 is not None: + print(f"Artifact SHA-256: {artifact_sha256}") + print( + "Do not retry this upload. Reconcile the artifact in the hosted " + "control plane first." + ) return 1 except HostedError as e: if args.json: @@ -3642,7 +3699,7 @@ def _cmd_push(args: argparse.Namespace) -> int: try: document = _push_json_result(result) except ValueError: - _print_push_json(_push_json_invalid_response()) + _print_push_json(_push_json_invalid_response(result)) return 1 _print_push_json(document) return 0 @@ -3651,15 +3708,29 @@ def _cmd_push(args: argparse.Namespace) -> int: print( "Upload paused for local review; the original was not modified or uploaded." ) - print(result["review_command"]) + print(f"Review original: {result['original_path']}") + print(f"Sanitized derivative: {result['sanitized_path']}") + print( + "Review locally: openadapt-flow review-sanitized " + f"{result['sanitized_path']} --original {result['original_path']}" + ) return 0 - workflow_id = result.get("workflow_id", "") - compile_status = (result.get("compile") or {}).get("status", "?") - print( - f"Pushed. workflow_id={workflow_id} " - f"(name={result.get('workflow_name')!r}, kind={result.get('kind')}, " - f"compile={compile_status})." - ) + kind = result.get("kind") + server_status = result.get("status", "?") + if result.get("workflow_id"): + version = result.get("version") + version_number = version.get("version") if isinstance(version, dict) else None + detail = ( + f"name={result.get('workflow_name')!r}, kind={kind}, status={server_status}" + ) + if version_number is not None: + detail += f", version={version_number}" + print(f"Pushed. workflow_id={result['workflow_id']} ({detail}).") + else: + print( + f"Pushed. artifact_ingest_id={result.get('artifact_ingest_id')} " + f"(kind={kind}, status={server_status})." + ) if result.get("dashboard_url"): print(f"Dashboard: {result['dashboard_url']}") return 0 diff --git a/openadapt_flow/hosted.py b/openadapt_flow/hosted.py index 2491b65d..4a0f111f 100644 --- a/openadapt_flow/hosted.py +++ b/openadapt_flow/hosted.py @@ -119,9 +119,9 @@ class HostedError(RuntimeError): class HostedDeliveryUncertain(HostedError): """The upload request failed without a trustworthy delivery result.""" - def __init__(self, message: str, *, context: Optional[dict[str, Any]] = None): + def __init__(self, message: str, *, context: dict[str, Any]): super().__init__(message) - self.context = context or {} + self.context = context @dataclass(frozen=True) @@ -1119,6 +1119,108 @@ def _local_review_id(manifest: dict[str, Any]) -> str: return hashlib.sha256(b"openadapt.sanitized-review/v1\0" + payload).hexdigest() +def _is_push_contract_uuid(value: Any) -> bool: + """Return whether *value* is a canonical RFC 9562 UUID (versions 1-8). + + This matches the hosted control plane's own identifier check and the + ``_UUID_RE`` used for run reports, so a retained server id is never + refused only because of its version nibble. + """ + if not isinstance(value, str): + return False + try: + parsed = UUID(value) + except ValueError: + return False + return str(parsed) == value and parsed.version in {1, 2, 3, 4, 5, 6, 7, 8} + + +def _validate_ingest_acknowledgment( + ingest: dict[str, Any], + *, + expected_kind: str, + expected_artifact_sha256: str, + expected_workflow_id: Optional[str], + expected_resolves_run_id: Optional[str], +) -> None: + """Require the complete server-owned identity chain for a 201 response. + + A status code alone is not proof that the intended artifact was retained. + This check protects both the human CLI and the structured controller mode. + """ + if not _is_push_contract_uuid(ingest.get("artifact_ingest_id")): + raise ValueError("artifact_ingest_id is missing or invalid") + if ingest.get("kind") != expected_kind: + raise ValueError("artifact kind does not match the approved archive") + if ingest.get("artifact_sha256") != expected_artifact_sha256: + raise ValueError("artifact hash does not match the approved archive") + + workflow_id = ingest.get("workflow_id") + if expected_kind == "recording": + if workflow_id is not None: + raise ValueError("recording ingest cannot activate a workflow") + if ingest.get("status") not in { + "needs_parameterization", + "needs_runtime_validation", + }: + raise ValueError("recording ingest has no governed next action") + return + + if ingest.get("status") != "accepted": + raise ValueError("bundle ingest is not accepted") + if not _is_push_contract_uuid(workflow_id): + raise ValueError("bundle workflow_id is missing or invalid") + if expected_workflow_id is not None and workflow_id != expected_workflow_id: + raise ValueError("bundle workflow_id does not match the requested workflow") + version = ingest.get("version") + if not isinstance(version, dict): + raise ValueError("retained bundle version is missing") + if not _is_push_contract_uuid(version.get("id")): + raise ValueError("retained bundle version id is invalid") + if not _is_push_contract_uuid(version.get("org_id")): + raise ValueError("retained organization id is invalid") + if version.get("workflow_id") != workflow_id: + raise ValueError("retained bundle workflow does not match") + if version.get("artifact_sha256") != expected_artifact_sha256: + raise ValueError("retained bundle artifact does not match") + if version.get("promoted_from_run_id") != expected_resolves_run_id: + raise ValueError("retained resolved-run binding does not match") + if not _is_push_contract_uuid(version.get("runtime_validation_id")): + raise ValueError("retained runtime validation id is invalid") + version_number = version.get("version") + if ( + not isinstance(version_number, int) + or isinstance(version_number, bool) + or version_number < 1 + ): + raise ValueError("retained bundle version number is invalid") + + +def _verified_archive_snapshot(archive_path: Path, *, expected_sha256: str) -> Any: + """Copy the approved archive to a private file and verify those exact bytes. + + The source path can be replaced after approval. The upload must therefore + use a verified file descriptor that is independent of that mutable path. + """ + snapshot = tempfile.TemporaryFile(mode="w+b") + digest = hashlib.sha256() + try: + with archive_path.open("rb") as source: + while chunk := source.read(1024 * 1024): + digest.update(chunk) + snapshot.write(chunk) + if digest.hexdigest() != expected_sha256: + raise HostedError( + "Approved immutable archive changed before upload; refusing egress" + ) + snapshot.flush() + snapshot.seek(0) + return snapshot + except Exception: + snapshot.close() + raise + + def push( path: Optional[Any] = None, *, @@ -1163,6 +1265,8 @@ def push( normalized_workflow_id = str(UUID(str(workflow_id).strip())) except (ValueError, AttributeError) as exc: raise HostedError("--workflow-id must be a valid UUID") from exc + if not _is_push_contract_uuid(normalized_workflow_id): + raise HostedError("--workflow-id must be a canonical RFC 9562 UUID") normalized_resolves_run_id: Optional[str] = None if resolves_run_id is not None: if requested_kind != "bundle" or normalized_workflow_id is None: @@ -1173,6 +1277,8 @@ def push( normalized_resolves_run_id = str(UUID(str(resolves_run_id).strip())) except (ValueError, AttributeError) as exc: raise HostedError("--resolves-run-id must be a valid UUID") from exc + if not _is_push_contract_uuid(normalized_resolves_run_id): + raise HostedError("--resolves-run-id must be a canonical RFC 9562 UUID") resolved_host = resolve_host(host) resolved_token = resolve_token(token, host=resolved_host) lane = resolve_deployment_kind(deployment_kind) @@ -1247,9 +1353,8 @@ def push( "uploaded": False, "pending_review": True, "sanitized_path": str(derivative), - "review_command": ( - f"openadapt-flow review-sanitized {derivative} --original {src}" - ), + "review_action": "review_sanitized", + "original_path": str(src), "destination_kind": destination.kind, "destination_host": resolved_host, "deployment_kind": lane, @@ -1373,6 +1478,7 @@ def push( "sanitization_policy": local_manifest["policy_version"], }, "resolves_run_id": normalized_resolves_run_id, + "requested_workflow_id": normalized_workflow_id, } if attestation is not None: certification = attestation.get("certification") or {} @@ -1391,7 +1497,18 @@ def push( ), } try: - with archive_path.open("rb") as fh: + archive_snapshot = _verified_archive_snapshot( + archive_path, + expected_sha256=approval["approved_derivative_sha256"], + ) + except HostedError: + raise + except OSError as exc: + raise HostedError( + "Approved immutable archive could not be prepared for upload" + ) from exc + try: + with archive_snapshot as fh: resp = httpx.post( f"{resolved_host}/api/ingest", headers=_auth_headers(resolved_token), @@ -1411,6 +1528,11 @@ def push( f"Upload to {resolved_host}/api/ingest failed: {exc}", context=local_result, ) from exc + except OSError as exc: + raise HostedDeliveryUncertain( + f"Upload to {resolved_host}/api/ingest failed during dispatch", + context=local_result, + ) from exc if resp.status_code == 401: raise HostedError("Ingest token was rejected (401).") if 400 <= resp.status_code < 500 and resp.status_code not in {408, 409}: @@ -1440,6 +1562,19 @@ def push( "Ingest returned 201 without a valid result object", context=local_result, ) + try: + _validate_ingest_acknowledgment( + ingest, + expected_kind=kind, + expected_artifact_sha256=ingest_manifest["artifact"]["sha256"], + expected_workflow_id=normalized_workflow_id, + expected_resolves_run_id=normalized_resolves_run_id, + ) + except ValueError as exc: + raise HostedDeliveryUncertain( + "Ingest returned 201 without a complete exact artifact acknowledgment", + context=local_result, + ) from exc workflow_id = ingest.get("workflow_id") result = dict(ingest) # Never trust a dashboard URL supplied by the remote response. Construct it diff --git a/public-artifacts.json b/public-artifacts.json index 74e076f3..c8ec46c7 100644 --- a/public-artifacts.json +++ b/public-artifacts.json @@ -5729,7 +5729,7 @@ }, { "path": "schemas/push-result-v1.json", - "sha256": "cdf50f23992c3b6a3509c409bc0992d1461ec1bb6043e3cdcf97c3314b60ade7" + "sha256": "d0b719735fd81aa9291629d6aa5dad5d67565bf7e752a4b90170da74dc9098cc" }, { "path": "schemas/runtime-validation-attestation-v1.json", diff --git a/schemas/push-result-v1.json b/schemas/push-result-v1.json index 0bee0ae5..775f7670 100644 --- a/schemas/push-result-v1.json +++ b/schemas/push-result-v1.json @@ -35,12 +35,24 @@ { "type": "object", "additionalProperties": false, - "required": ["id", "scope", "sanitized_path", "command"], + "required": [ + "id", + "scope", + "sanitized_path", + "action", + "original_path" + ], "properties": { "id": { "$ref": "#/$defs/sha256" }, "scope": { "const": "local_non_authoritative" }, "sanitized_path": { "type": ["string", "null"], "minLength": 1 }, - "command": { "type": ["string", "null"], "minLength": 1 } + "action": { + "oneOf": [ + { "type": "null" }, + { "const": "review_sanitized" } + ] + }, + "original_path": { "type": ["string", "null"], "minLength": 1 } } } ] @@ -55,8 +67,11 @@ "properties": { "id": { "type": "string", "minLength": 1, "maxLength": 200 }, "schema": { - "type": "string", - "pattern": "^openadapt\\.runtime-validation/v[0-9]+$" + "enum": [ + "openadapt.runtime-validation/v1", + "openadapt.runtime-validation/v2", + "openadapt.runtime-validation/v3" + ] } } } @@ -146,11 +161,7 @@ "required": ["code", "message"], "properties": { "code": { - "enum": [ - "push_failed", - "delivery_uncertain", - "invalid_ingest_response" - ] + "enum": ["push_failed", "delivery_uncertain"] }, "message": { "type": "string", "minLength": 1, "maxLength": 500 } } @@ -171,7 +182,7 @@ }, "uuid": { "type": "string", - "pattern": "^[a-f0-9]{8}-[a-f0-9]{4}-[1-5][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$" + "pattern": "^[a-f0-9]{8}-[a-f0-9]{4}-[1-8][a-f0-9]{3}-[89ab][a-f0-9]{3}-[a-f0-9]{12}$" } }, "allOf": [ @@ -185,7 +196,8 @@ "type": "object", "properties": { "sanitized_path": { "type": "string", "minLength": 1 }, - "command": { "type": "string", "minLength": 1 } + "action": { "const": "review_sanitized" }, + "original_path": { "type": "string", "minLength": 1 } } }, "attestation": { "type": "null" }, @@ -197,8 +209,17 @@ "kind": { "enum": ["recording", "bundle"] }, "source_tree_sha256": { "$ref": "#/$defs/sha256" }, "derivative_tree_sha256": { "$ref": "#/$defs/sha256" }, + "sanitization_policy": { "type": "string", "minLength": 1 }, "approved_archive_sha256": { "type": "null" }, "artifact_sha256": { "type": "null" }, + "bundle_sha256": { "type": "null" }, + "source_recording_sha256": { "type": "null" }, + "certification_policy": { "type": "null" }, + "certification_evidence_sha256": { "type": "null" }, + "governed_authorization_template_sha256": { "type": "null" }, + "parameter_schema_sha256": { "type": "null" }, + "attested_run_report_sha256": { "type": "null" }, + "resolves_run_id": { "type": "null" }, "organization_id": { "type": "null" }, "bundle_version_id": { "type": "null" }, "bundle_version": { "type": "null" }, @@ -223,7 +244,8 @@ "type": "object", "properties": { "sanitized_path": { "type": "null" }, - "command": { "type": "null" } + "action": { "type": "null" }, + "original_path": { "type": "null" } } }, "error": { "type": "null" }, @@ -309,33 +331,31 @@ } }, { - "if": { "properties": { "status": { "const": "failed" } } }, - "then": { + "if": { "properties": { - "workflow_id": { "type": "null" }, - "artifact_ingest_id": { "type": "null" }, - "review": { "type": "null" }, - "attestation": { "type": "null" }, - "next_action": { "enum": ["reconcile", null] }, - "dashboard_url": { "type": "null" }, - "binding": { - "properties": { - "organization_id": { "type": "null" }, - "bundle_version_id": { "type": "null" }, - "bundle_version": { "type": "null" }, - "runtime_validation_id": { "type": "null" } - } + "status": { + "enum": ["accepted_for_ingest", "delivery_uncertain"] }, - "delivery": { + "attestation": { + "type": "object", "properties": { - "attempted": { "enum": [true, null] }, - "certainty": { "enum": ["not_accepted", "unknown"] } - } + "schema": { "const": "openadapt.runtime-validation/v3" } + }, + "required": ["schema"] }, - "error": { - "type": "object", + "binding": { + "properties": { "kind": { "const": "bundle" } }, + "required": ["kind"] + } + } + }, + "then": { + "properties": { + "binding": { "properties": { - "code": { "enum": ["push_failed", "invalid_ingest_response"] } + "governed_authorization_template_sha256": { + "$ref": "#/$defs/sha256" + } } } } @@ -344,44 +364,79 @@ { "if": { "properties": { - "status": { "const": "failed" }, - "error": { + "status": { + "enum": ["accepted_for_ingest", "delivery_uncertain"] + }, + "attestation": { "type": "object", - "properties": { "code": { "const": "push_failed" } }, - "required": ["code"] + "properties": { + "schema": { + "enum": [ + "openadapt.runtime-validation/v1", + "openadapt.runtime-validation/v2" + ] + } + }, + "required": ["schema"] + }, + "binding": { + "properties": { "kind": { "const": "bundle" } }, + "required": ["kind"] } } }, "then": { "properties": { - "next_action": { "type": "null" }, - "delivery": { + "binding": { "properties": { - "attempted": { "type": "null" }, - "certainty": { "const": "not_accepted" } + "governed_authorization_template_sha256": { "type": "null" } } } } } }, { - "if": { - "properties": { - "status": { "const": "failed" }, - "error": { - "type": "object", - "properties": { "code": { "const": "invalid_ingest_response" } }, - "required": ["code"] - } - } - }, + "if": { "properties": { "status": { "const": "failed" } } }, "then": { "properties": { - "next_action": { "const": "reconcile" }, + "workflow_id": { "type": "null" }, + "artifact_ingest_id": { "type": "null" }, + "review": { "type": "null" }, + "attestation": { "type": "null" }, + "next_action": { "type": "null" }, + "dashboard_url": { "type": "null" }, + "binding": { + "properties": { + "kind": { "type": "null" }, + "source_tree_sha256": { "type": "null" }, + "derivative_tree_sha256": { "type": "null" }, + "approved_archive_sha256": { "type": "null" }, + "artifact_sha256": { "type": "null" }, + "bundle_sha256": { "type": "null" }, + "source_recording_sha256": { "type": "null" }, + "sanitization_policy": { "type": "null" }, + "certification_policy": { "type": "null" }, + "certification_evidence_sha256": { "type": "null" }, + "governed_authorization_template_sha256": { "type": "null" }, + "parameter_schema_sha256": { "type": "null" }, + "attested_run_report_sha256": { "type": "null" }, + "resolves_run_id": { "type": "null" }, + "organization_id": { "type": "null" }, + "bundle_version_id": { "type": "null" }, + "bundle_version": { "type": "null" }, + "runtime_validation_id": { "type": "null" } + } + }, "delivery": { "properties": { - "attempted": { "const": true }, - "certainty": { "const": "unknown" } + "attempted": { "type": "null" }, + "certainty": { "const": "not_accepted" } + } + }, + "error": { + "type": "object", + "properties": { + "code": { "const": "push_failed" } } } } @@ -393,10 +448,24 @@ "properties": { "workflow_id": { "type": "null" }, "artifact_ingest_id": { "type": "null" }, + "review": { + "type": "object", + "properties": { + "sanitized_path": { "type": "null" }, + "action": { "type": "null" }, + "original_path": { "type": "null" } + } + }, "next_action": { "const": "reconcile" }, "dashboard_url": { "type": "null" }, "binding": { "properties": { + "kind": { "enum": ["recording", "bundle"] }, + "source_tree_sha256": { "$ref": "#/$defs/sha256" }, + "derivative_tree_sha256": { "$ref": "#/$defs/sha256" }, + "approved_archive_sha256": { "$ref": "#/$defs/sha256" }, + "artifact_sha256": { "$ref": "#/$defs/sha256" }, + "sanitization_policy": { "type": "string", "minLength": 1 }, "organization_id": { "type": "null" }, "bundle_version_id": { "type": "null" }, "bundle_version": { "type": "null" }, @@ -415,6 +484,87 @@ } } } + }, + { + "if": { + "properties": { + "status": { "const": "delivery_uncertain" }, + "binding": { + "properties": { "kind": { "const": "recording" } }, + "required": ["kind"] + } + } + }, + "then": { + "properties": { + "attestation": { "type": "null" }, + "binding": { + "properties": { + "bundle_sha256": { "type": "null" }, + "source_recording_sha256": { "type": "null" }, + "certification_policy": { "type": "null" }, + "certification_evidence_sha256": { "type": "null" }, + "governed_authorization_template_sha256": { "type": "null" }, + "parameter_schema_sha256": { "type": "null" }, + "attested_run_report_sha256": { "type": "null" }, + "resolves_run_id": { "type": "null" } + } + } + } + } + }, + { + "if": { + "properties": { + "status": { "const": "delivery_uncertain" }, + "binding": { + "properties": { "kind": { "const": "bundle" } }, + "required": ["kind"] + } + } + }, + "then": { + "oneOf": [ + { + "properties": { + "attestation": { "type": "object" }, + "binding": { + "properties": { + "bundle_sha256": { "$ref": "#/$defs/sha256" }, + "source_recording_sha256": { "$ref": "#/$defs/sha256" }, + "certification_policy": { + "type": "string", + "minLength": 1 + }, + "certification_evidence_sha256": { + "$ref": "#/$defs/sha256" + }, + "parameter_schema_sha256": { "$ref": "#/$defs/sha256" }, + "attested_run_report_sha256": { + "$ref": "#/$defs/sha256" + } + } + } + } + }, + { + "properties": { + "attestation": { "type": "null" }, + "binding": { + "properties": { + "bundle_sha256": { "type": "null" }, + "source_recording_sha256": { "type": "null" }, + "certification_policy": { "type": "null" }, + "certification_evidence_sha256": { "type": "null" }, + "governed_authorization_template_sha256": { "type": "null" }, + "parameter_schema_sha256": { "type": "null" }, + "attested_run_report_sha256": { "type": "null" } + } + } + } + } + ] + } } ] } diff --git a/tests/test_hosted.py b/tests/test_hosted.py index 062e7001..62158f91 100644 --- a/tests/test_hosted.py +++ b/tests/test_hosted.py @@ -807,6 +807,31 @@ def fake(url, **kw): return fake +def _accepted_recording_post(recorder=None, *, status="needs_parameterization"): + """Return a server double with the complete recording-ingest contract.""" + + def fake(url, **kw): + if recorder is not None: + recorder["url"] = url + recorder["kw"] = kw + manifest = json.loads(kw["data"]["sanitization_manifest"]) + artifact_sha256 = manifest["artifact"]["sha256"] + return httpx.Response( + 201, + json={ + "ingest": { + "workflow_id": None, + "artifact_ingest_id": _PUSH_INGEST_ID, + "kind": "recording", + "artifact_sha256": artifact_sha256, + "status": status, + } + }, + ) + + return fake + + class _FakeScrubber: """A fast text+image scrubber double (satisfies privacy.Scrubber). @@ -831,18 +856,11 @@ def test_push_success(tmp_path, monkeypatch): rec = _make_recording(tmp_path, "rec") privacy.set_text_scrubber(_FakeScrubber()) # cloud recording => scrub before upload recorder: dict = {} - body = { - "ingest": { - "workflow_id": "wf_123", - "workflow_name": "Pushed recording", - "kind": "recording", - "compile": {"status": "compiled", "steps": 4}, - } - } - monkeypatch.setattr(httpx, "post", _capture_post(recorder, 201, body)) + monkeypatch.setattr(httpx, "post", _accepted_recording_post(recorder)) result = hosted.push(rec, name="My flow", host="https://h.test", token="tok") - assert result["workflow_id"] == "wf_123" - assert result["dashboard_url"] == "https://h.test/dashboard/workflows/wf_123" + assert result["workflow_id"] is None + assert result["artifact_ingest_id"] == _PUSH_INGEST_ID + assert "dashboard_url" not in result assert recorder["url"] == "https://h.test/api/ingest" assert recorder["kw"]["data"]["kind"] == "recording" assert recorder["kw"]["data"]["name"] == "My flow" @@ -863,10 +881,11 @@ def test_push_default_path_uses_latest(tmp_path, monkeypatch): monkeypatch.setattr( httpx, "post", - _capture_post(recorder, 201, {"ingest": {"workflow_id": "wf_9"}}), + _accepted_recording_post(recorder), ) result = hosted.push(host="https://h.test", token="tok") - assert result["workflow_id"] == "wf_9" + assert result["workflow_id"] is None + assert result["artifact_ingest_id"] == _PUSH_INGEST_ID def test_push_bad_kind(tmp_path): @@ -1003,6 +1022,25 @@ def test_push_201_without_json_acknowledgment_is_delivery_uncertain( assert raised.value.context["local_binding"]["approved_archive_sha256"] +def test_push_201_without_complete_exact_acknowledgment_is_delivery_uncertain( + tmp_path, monkeypatch +): + rec = _make_recording(tmp_path, "rec") + privacy.set_text_scrubber(_FakeScrubber()) + monkeypatch.setattr( + httpx, + "post", + lambda *args, **kwargs: httpx.Response( + 201, + json={"ingest": {"workflow_id": None, "kind": "recording"}}, + ), + ) + + with pytest.raises(hosted.HostedDeliveryUncertain) as raised: + hosted.push(rec, token="tok", host="https://h.test") + assert raised.value.context["local_binding"]["approved_archive_sha256"] + + def test_push_bundle_requires_verified_sanitization_not_attestation( tmp_path, monkeypatch ): @@ -1154,7 +1192,7 @@ def test_push_recording_on_byoc_is_sanitized_before_upload(tmp_path, monkeypatch monkeypatch.setattr( httpx, "post", - lambda *a, **k: httpx.Response(201, json={"ingest": {"workflow_id": "wf"}}), + _accepted_recording_post(), ) result = hosted.push( rec, deployment_kind="byoc", host="https://h.test", token="tok" @@ -1172,7 +1210,7 @@ def test_push_recording_under_phi_mode_is_sanitized_before_upload( monkeypatch.setattr( httpx, "post", - lambda *a, **k: httpx.Response(201, json={"ingest": {"workflow_id": "wf"}}), + _accepted_recording_post(), ) result = hosted.push( rec, deployment_kind="cloud", host="https://h.test", token="tok" @@ -1214,13 +1252,26 @@ def test_push_recording_scrubs_before_upload(tmp_path, monkeypatch): def fake_post(url, **kw): # Read the uploaded zip bytes so we can assert what actually shipped. captured["bytes"] = kw["files"]["file"][1].read() - return httpx.Response(201, json={"ingest": {"workflow_id": "wf_s"}}) + manifest = json.loads(kw["data"]["sanitization_manifest"]) + return httpx.Response( + 201, + json={ + "ingest": { + "workflow_id": None, + "artifact_ingest_id": _PUSH_INGEST_ID, + "kind": "recording", + "artifact_sha256": manifest["artifact"]["sha256"], + "status": "needs_parameterization", + } + }, + ) monkeypatch.setattr(httpx, "post", fake_post) result = hosted.push( rec, deployment_kind="cloud", host="https://h.test", token="tok" ) - assert result["workflow_id"] == "wf_s" + assert result["workflow_id"] is None + assert result["artifact_ingest_id"] == _PUSH_INGEST_ID # scrub path ran: text + image scrubbers were both invoked. assert fake.text_calls >= 1 assert fake.image_calls == 3 # transform, stable second pass, approval rescan @@ -2956,7 +3007,8 @@ def test_cli_push_json_paused_for_review(monkeypatch, capsys): "pending_review": True, "kind": "recording", "sanitized_path": "/safe/derivative", - "review_command": "openadapt-flow review-sanitized /safe/derivative", + "review_action": "review_sanitized", + "original_path": "/safe/source", "review_id": _PUSH_REVIEW_ID, "local_binding": { "source_tree_sha256": _PUSH_SOURCE_SHA, @@ -2978,7 +3030,8 @@ def test_cli_push_json_paused_for_review(monkeypatch, capsys): "id": _PUSH_REVIEW_ID, "scope": "local_non_authoritative", "sanitized_path": "/safe/derivative", - "command": "openadapt-flow review-sanitized /safe/derivative", + "action": "review_sanitized", + "original_path": "/safe/source", } assert value["binding"]["source_tree_sha256"] == _PUSH_SOURCE_SHA assert value["binding"]["approved_archive_sha256"] is None @@ -2999,7 +3052,8 @@ def test_push_json_schema_rejects_dashboard_on_paused_review(monkeypatch, capsys "pending_review": True, "kind": "recording", "sanitized_path": "/safe/derivative", - "review_command": "openadapt-flow review-sanitized /safe/derivative", + "review_action": "review_sanitized", + "original_path": "/safe/source", "review_id": _PUSH_REVIEW_ID, "local_binding": { "source_tree_sha256": _PUSH_SOURCE_SHA, @@ -3016,6 +3070,62 @@ def test_push_json_schema_rejects_dashboard_on_paused_review(monkeypatch, capsys _assert_push_json_schema(document) +def test_push_json_schema_rejects_bundle_binding_on_paused_review(monkeypatch, capsys): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: { + "uploaded": False, + "pending_review": True, + "kind": "bundle", + "sanitized_path": "/safe/derivative", + "review_action": "review_sanitized", + "original_path": "/safe/source", + "review_id": _PUSH_REVIEW_ID, + "local_binding": { + "source_tree_sha256": _PUSH_SOURCE_SHA, + "derivative_tree_sha256": _PUSH_DERIVATIVE_SHA, + "approved_archive_sha256": None, + "sanitization_policy": "outbound-phi-v1", + }, + }, + ) + assert main(["push", "raw", "--kind", "bundle", "--json"]) == 0 + document = json.loads(capsys.readouterr().out) + document["binding"]["bundle_sha256"] = _PUSH_ARTIFACT_SHA + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + +def test_push_json_schema_requires_sanitization_policy_on_paused_review( + monkeypatch, capsys +): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: { + "uploaded": False, + "pending_review": True, + "kind": "recording", + "sanitized_path": "/safe/derivative", + "review_action": "review_sanitized", + "original_path": "/safe/source", + "review_id": _PUSH_REVIEW_ID, + "local_binding": { + "source_tree_sha256": _PUSH_SOURCE_SHA, + "derivative_tree_sha256": _PUSH_DERIVATIVE_SHA, + "approved_archive_sha256": None, + "sanitization_policy": "outbound-phi-v1", + }, + }, + ) + assert main(["push", "raw", "--json"]) == 0 + document = json.loads(capsys.readouterr().out) + document["binding"]["sanitization_policy"] = None + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + @pytest.mark.parametrize( ("server_status", "next_action"), [ @@ -3071,6 +3181,7 @@ def test_cli_push_json_bundle_accepted_with_exact_attestation_binding( "version": 3, "artifact_sha256": _PUSH_ARTIFACT_SHA, "runtime_validation_id": _PUSH_RUNTIME_VALIDATION_ID, + "promoted_from_run_id": ("cccccccc-cccc-4ccc-8ccc-cccccccccccc"), }, } ) @@ -3108,6 +3219,31 @@ def test_cli_push_json_bundle_accepted_with_exact_attestation_binding( } assert value["next_action"] == "open_dashboard" + missing_template = json.loads(json.dumps(value)) + missing_template["binding"]["governed_authorization_template_sha256"] = None + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(missing_template) + + legacy_with_template = json.loads(json.dumps(value)) + legacy_with_template["attestation"]["schema"] = "openadapt.runtime-validation/v2" + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(legacy_with_template) + + future_schema = json.loads(json.dumps(value)) + future_schema["attestation"]["schema"] = "openadapt.runtime-validation/v999" + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(future_schema) + + result["attestation_binding"]["schema"] = "openadapt.runtime-validation/v999" + assert main(["push", "approved", "--kind", "bundle", "--json"]) == 1 + rejected = json.loads(capsys.readouterr().out) + _assert_push_json_schema(rejected) + assert rejected["status"] == "delivery_uncertain" + assert rejected["binding"]["kind"] == "bundle" + assert rejected["binding"]["artifact_sha256"] == _PUSH_ARTIFACT_SHA + assert rejected["attestation"] is None + assert rejected["binding"]["bundle_sha256"] is None + @pytest.mark.parametrize( ("field", "value"), @@ -3116,6 +3252,7 @@ def test_cli_push_json_bundle_accepted_with_exact_attestation_binding( ("workflow_id", "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa"), ("artifact_sha256", "9" * 64), ("runtime_validation_id", None), + ("promoted_from_run_id", "99999999-9999-4999-8999-999999999999"), ], ) def test_cli_push_json_refuses_mismatched_server_version_binding( @@ -3128,6 +3265,7 @@ def test_cli_push_json_refuses_mismatched_server_version_binding( "dashboard_url": ( f"https://h.test/dashboard/workflows/{_PUSH_WORKFLOW_ID}" ), + "resolves_run_id": "cccccccc-cccc-4ccc-8ccc-cccccccccccc", "status": "accepted", "version": { "id": _PUSH_VERSION_ID, @@ -3136,6 +3274,7 @@ def test_cli_push_json_refuses_mismatched_server_version_binding( "version": 3, "artifact_sha256": _PUSH_ARTIFACT_SHA, "runtime_validation_id": _PUSH_RUNTIME_VALIDATION_ID, + "promoted_from_run_id": ("cccccccc-cccc-4ccc-8ccc-cccccccccccc"), }, "attestation_binding": { "schema": "openadapt.runtime-validation/v3", @@ -3156,8 +3295,9 @@ def test_cli_push_json_refuses_mismatched_server_version_binding( assert main(["push", "approved", "--kind", "bundle", "--json"]) == 1 document = json.loads(capsys.readouterr().out) _assert_push_json_schema(document) - assert document["status"] == "failed" - assert document["error"]["code"] == "invalid_ingest_response" + assert document["status"] == "delivery_uncertain" + assert document["error"]["code"] == "delivery_uncertain" + assert document["binding"]["artifact_sha256"] == _PUSH_ARTIFACT_SHA def test_cli_push_json_refuses_false_success_without_server_ingest_id( @@ -3171,10 +3311,11 @@ def test_cli_push_json_refuses_false_success_without_server_ingest_id( assert main(["push", "approved", "--json"]) == 1 value = json.loads(capsys.readouterr().out) _assert_push_json_schema(value) - assert value["status"] == "failed" - assert value["error"]["code"] == "invalid_ingest_response" + assert value["status"] == "delivery_uncertain" + assert value["error"]["code"] == "delivery_uncertain" assert value["delivery"] == {"attempted": True, "certainty": "unknown"} assert value["next_action"] == "reconcile" + assert value["binding"]["artifact_sha256"] == _PUSH_ARTIFACT_SHA def test_cli_push_json_transport_error_is_delivery_uncertain(monkeypatch, capsys): @@ -3262,6 +3403,19 @@ def test_push_json_schema_rejects_server_binding_on_failure(monkeypatch, capsys) _assert_push_json_schema(document) +def test_push_json_schema_rejects_local_binding_on_failure(monkeypatch, capsys): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: (_ for _ in ()).throw(hosted.HostedError("private")), + ) + assert main(["push", "raw", "--json"]) == 1 + document = json.loads(capsys.readouterr().out) + document["binding"]["source_tree_sha256"] = _PUSH_SOURCE_SHA + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + @pytest.mark.parametrize( ("field", "value"), [ @@ -3279,7 +3433,7 @@ def test_push_json_schema_rejects_conflicting_uncertain_state( hosted, "push", lambda *args, **kwargs: (_ for _ in ()).throw( - hosted.HostedDeliveryUncertain("private") + hosted.HostedDeliveryUncertain("private", context=_json_push_base()) ), ) assert main(["push", "raw", "--json"]) == 1 @@ -3296,7 +3450,7 @@ def test_push_json_schema_rejects_server_binding_on_uncertain_state( hosted, "push", lambda *args, **kwargs: (_ for _ in ()).throw( - hosted.HostedDeliveryUncertain("private") + hosted.HostedDeliveryUncertain("private", context=_json_push_base()) ), ) assert main(["push", "raw", "--json"]) == 1 @@ -3306,6 +3460,40 @@ def test_push_json_schema_rejects_server_binding_on_uncertain_state( _assert_push_json_schema(document) +def test_push_json_schema_requires_reconciliation_hash_on_uncertain_state( + monkeypatch, capsys +): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: (_ for _ in ()).throw( + hosted.HostedDeliveryUncertain("private", context=_json_push_base()) + ), + ) + assert main(["push", "raw", "--json"]) == 1 + document = json.loads(capsys.readouterr().out) + document["binding"]["artifact_sha256"] = None + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + +def test_push_json_schema_rejects_obsolete_invalid_response_failure( + monkeypatch, capsys +): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: (_ for _ in ()).throw(hosted.HostedError("private")), + ) + assert main(["push", "raw", "--json"]) == 1 + document = json.loads(capsys.readouterr().out) + document["next_action"] = "reconcile" + document["delivery"] = {"attempted": True, "certainty": "unknown"} + document["error"]["code"] = "invalid_ingest_response" + with pytest.raises(jsonschema.ValidationError): + _assert_push_json_schema(document) + + def test_cli_report_break_dispatch(monkeypatch, capsys): captured: dict = {} @@ -3491,3 +3679,88 @@ def fake_push(*a, **k): rc = main(["push"]) assert rc == 1 assert "push failed" in capsys.readouterr().out + + +def test_cli_push_uncertain_delivery_requires_hash_reconciliation(monkeypatch, capsys): + def fake_push(*args, **kwargs): + raise hosted.HostedDeliveryUncertain( + "secret token and /private/source must not reach output", + context=_json_push_base(), + ) + + monkeypatch.setattr(hosted, "push", fake_push) + + assert main(["push", "raw"]) == 1 + output = capsys.readouterr().out + assert "delivery is uncertain" in output + assert f"Artifact SHA-256: {_PUSH_ARTIFACT_SHA}" in output + assert "Do not retry" in output + assert "Reconcile" in output + assert "secret token" not in output + assert "/private/source" not in output + + +def test_cli_push_human_recording_reports_ingest_id_not_workflow(monkeypatch, capsys): + result = _json_push_base() + result["status"] = "needs_parameterization" + monkeypatch.setattr(hosted, "push", lambda *args, **kwargs: result) + + assert main(["push", "approved"]) == 0 + output = capsys.readouterr().out + assert f"artifact_ingest_id={_PUSH_INGEST_ID}" in output + assert "status=needs_parameterization" in output + assert "workflow_id=None" not in output + assert "compile=" not in output + + +def test_cli_push_human_paused_review_prints_local_review_command(monkeypatch, capsys): + monkeypatch.setattr( + hosted, + "push", + lambda *args, **kwargs: { + "uploaded": False, + "pending_review": True, + "kind": "recording", + "sanitized_path": "/safe/derivative", + "review_action": "review_sanitized", + "original_path": "/safe/source", + "review_id": _PUSH_REVIEW_ID, + "local_binding": { + "source_tree_sha256": _PUSH_SOURCE_SHA, + "derivative_tree_sha256": _PUSH_DERIVATIVE_SHA, + "approved_archive_sha256": None, + "sanitization_policy": "outbound-phi-v1", + }, + }, + ) + assert main(["push", "raw"]) == 0 + output = capsys.readouterr().out + assert "paused for local review" in output + assert ( + "openadapt-flow review-sanitized /safe/derivative --original /safe/source" + in output + ) + + +def test_cli_push_json_accepts_rfc9562_v7_server_ids(monkeypatch, capsys): + """Server-owned ids match the hosted UUID contract (versions 1-8), not 1-5.""" + result = _json_push_base() + result["status"] = "needs_parameterization" + result["artifact_ingest_id"] = "019194b0-1234-7abc-8def-0123456789ab" + monkeypatch.setattr(hosted, "push", lambda *args, **kwargs: result) + + assert main(["push", "approved", "--json"]) == 0 + value = json.loads(capsys.readouterr().out) + _assert_push_json_schema(value) + assert value["status"] == "accepted_for_ingest" + assert value["artifact_ingest_id"] == "019194b0-1234-7abc-8def-0123456789ab" + + +def test_push_contract_uuid_rejects_nil_and_noncanonical_forms(): + assert hosted._is_push_contract_uuid("019194b0-1234-7abc-8def-0123456789ab") + assert hosted._is_push_contract_uuid(_PUSH_INGEST_ID) + assert not hosted._is_push_contract_uuid("00000000-0000-0000-0000-000000000000") + assert not hosted._is_push_contract_uuid(_PUSH_INGEST_ID.upper()) + assert not hosted._is_push_contract_uuid(_PUSH_INGEST_ID.replace("-", "")) + assert not hosted._is_push_contract_uuid("aaaaaaaa-aaaa-9aaa-8aaa-aaaaaaaaaaaa") + assert not hosted._is_push_contract_uuid(None) diff --git a/tests/test_runtime_validation.py b/tests/test_runtime_validation.py index dc832290..fec62568 100644 --- a/tests/test_runtime_validation.py +++ b/tests/test_runtime_validation.py @@ -682,7 +682,34 @@ def test_bundle_push_requires_and_sends_validation_attestation(tmp_path, monkeyp def post(url, **kwargs): captured.update(kwargs) - return httpx.Response(201, json={"ingest": {"workflow_id": "wf-1"}}) + artifact_sha256 = json.loads(kwargs["data"]["sanitization_manifest"])[ + "artifact" + ]["sha256"] + return httpx.Response( + 201, + json={ + "ingest": { + "workflow_id": "ec726a3e-dcaf-40cf-870a-867d104002dd", + "artifact_ingest_id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "kind": "bundle", + "artifact_sha256": artifact_sha256, + "status": "accepted", + "version": { + "id": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb", + "org_id": "cccccccc-cccc-4ccc-8ccc-cccccccccccc", + "workflow_id": "ec726a3e-dcaf-40cf-870a-867d104002dd", + "version": 2, + "artifact_sha256": artifact_sha256, + "runtime_validation_id": ( + "eeeeeeee-eeee-4eee-8eee-eeeeeeeeeeee" + ), + "promoted_from_run_id": ( + "d3ecf64d-0d25-4df7-9264-77bf7d266d77" + ), + }, + } + }, + ) monkeypatch.setattr(httpx, "post", post) result = hosted.push( diff --git a/tests/test_sanitized_artifact.py b/tests/test_sanitized_artifact.py index e97fe3e9..a7162de8 100644 --- a/tests/test_sanitized_artifact.py +++ b/tests/test_sanitized_artifact.py @@ -764,10 +764,24 @@ def test_byoc_phi_mode_can_upload_exact_approved_sanitized_bytes(tmp_path, monke captured: dict = {} def post(url, **kwargs): + # The upload file is a private verified snapshot, not this mutable path. + approved_archive_path(dest).write_bytes(b"unapproved replacement") captured["url"] = url captured["archive"] = kwargs["files"]["file"][1].read() captured["data"] = kwargs["data"] - return httpx.Response(201, json={"ingest": {"workflow_id": "wf_1"}}) + envelope = json.loads(kwargs["data"]["sanitization_manifest"]) + return httpx.Response( + 201, + json={ + "ingest": { + "workflow_id": None, + "artifact_ingest_id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "kind": "recording", + "artifact_sha256": envelope["artifact"]["sha256"], + "status": "needs_parameterization", + } + }, + ) monkeypatch.setattr(httpx, "post", post) result = hosted.push( @@ -788,6 +802,30 @@ def post(url, **kwargs): assert envelope["artifact"]["sha256"] == hashlib.sha256(expected).hexdigest() +def test_push_refuses_archive_replaced_after_approval_before_snapshot( + tmp_path, monkeypatch +): + source = _recording(tmp_path) + dest = tmp_path / "sanitized" + sanitize_artifact(source, dest, kind="recording") + approve_derivative(dest, source=source, reviewer="privacy-officer") + real_snapshot = hosted._verified_archive_snapshot + + def replace_then_snapshot(archive_path, *, expected_sha256): + archive_path.write_bytes(b"Jane Doe unapproved replacement") + return real_snapshot(archive_path, expected_sha256=expected_sha256) + + monkeypatch.setattr(hosted, "_verified_archive_snapshot", replace_then_snapshot) + monkeypatch.setattr( + httpx, + "post", + lambda *args, **kwargs: pytest.fail("changed bytes must not egress"), + ) + + with pytest.raises(hosted.HostedError, match="changed before upload"): + hosted.push(dest, host=hosted.DEFAULT_HOST, token="token") + + def test_upload_filename_and_workflow_name_do_not_leak_phi(tmp_path, monkeypatch): source = _recording(tmp_path) dest = tmp_path / "Jane Doe sanitized" @@ -797,7 +835,19 @@ def test_upload_filename_and_workflow_name_do_not_leak_phi(tmp_path, monkeypatch def post(url, **kwargs): captured.update(kwargs) - return httpx.Response(201, json={"ingest": {"workflow_id": "wf_1"}}) + envelope = json.loads(kwargs["data"]["sanitization_manifest"]) + return httpx.Response( + 201, + json={ + "ingest": { + "workflow_id": None, + "artifact_ingest_id": "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa", + "kind": "recording", + "artifact_sha256": envelope["artifact"]["sha256"], + "status": "needs_parameterization", + } + }, + ) monkeypatch.setattr(httpx, "post", post) hosted.push(dest, name="Jane Doe intake", host=hosted.DEFAULT_HOST, token="token") @@ -822,6 +872,9 @@ def test_raw_push_creates_derivative_and_pauses_for_review(tmp_path, monkeypatch ) assert result["pending_review"] is True + assert result["review_action"] == "review_sanitized" + assert result["original_path"] == str(source) + assert "review_command" not in result assert re.fullmatch(r"[a-f0-9]{64}", result["review_id"]) assert result["review_id"] != result["local_binding"]["derivative_tree_sha256"] assert result["local_binding"] == {