diff --git a/README.md b/README.md index 037957f2..0c19a458 100644 --- a/README.md +++ b/README.md @@ -88,6 +88,27 @@ delivery is uncertain, so the transaction ends in `RECONCILIATION_REQUIRED` for a person to settle. Longer walkthrough, including the `--guided` presentation mode and the hand-driven stages: [docs/TUTORIAL.md](docs/TUTORIAL.md). +## Reference Execute server + +You can host the public Execute HTTP contract on this machine, in one process. + +```bash +pip install 'openadapt-flow[execute]' +openadapt-flow serve-execute --port 8787 --seed-mockmed +``` + +That command binds loopback, generates an Ed25519 key on first start, and +keeps it under `~/.openadapt/execute-ref/`. Health is `GET /health`. Submit +`openadapt.execute-request/v1` to `POST /v1/executions`, poll +`GET /v1/executions/{id}` until `terminal`, then read +`GET /v1/executions/{id}/receipt`. The same process also speaks MCP at +`POST /mcp`. + +Receipts are self-signed with that local key. They aren't OpenAdapt +production Seals. `GET /seals/{id}` is the local verify page; it shows the +key fingerprint and a $0 meter. Counterparties that require an OpenAdapt Seal +still use Cloud. + ## Record and rehearse your workflow Install the extras for the surface that will record and replay the workflow: diff --git a/openadapt_flow/__main__.py b/openadapt_flow/__main__.py index 49daee28..dbc32c2d 100644 --- a/openadapt_flow/__main__.py +++ b/openadapt_flow/__main__.py @@ -58,6 +58,8 @@ command. - ``console`` — serve the localhost-only operator console (a read-first web UI over bundles / runs / skill libraries; requires the ``console`` extra). +- ``serve-execute`` — host the public Execute HTTP+MCP contract on this + machine (self-signed local receipts; not an OpenAdapt production Seal). - ``emit-skill`` — emit an Agent Skills folder for a bundle. - ``emit-mcp`` — emit a standalone MCP ``server.py`` for a bundle. - ``connector`` — the BYOC (bring-your-own-cloud) outbound-pull daemon: @@ -7616,6 +7618,42 @@ def _repair_store_flag(rp: argparse.ArgumentParser) -> None: _add_deployment_flags(p) p.set_defaults(func=_cmd_console) + p = sub.add_parser( + "serve-execute", + help=( + "Host the public Execute v1 HTTP+MCP contract locally. Receipts " + "are self-signed with a local key. They are not OpenAdapt " + "production Seals. Needs `pip install 'openadapt-flow[execute]'`" + ), + ) + p.add_argument( + "--port", + type=int, + default=8787, + help="Port (default: 8787)", + ) + p.add_argument( + "--host", + default="127.0.0.1", + help="Bind address (default: 127.0.0.1)", + ) + p.add_argument( + "--data-dir", + default=None, + help="Local data directory (default: ~/.openadapt/execute-ref)", + ) + p.add_argument( + "--token", + default=None, + help="Bearer token (default: generated on first start in --data-dir)", + ) + p.add_argument( + "--seed-mockmed", + action="store_true", + help="Write the synthetic MockMed ok and banner-lie admissions", + ) + p.set_defaults(func=_cmd_serve_execute) + p = sub.add_parser( "business-decisions", help="Customer-runner typed-decision relay; it never resumes or acts.", @@ -7873,6 +7911,48 @@ def _cmd_console(args: argparse.Namespace) -> int: return 0 +def _cmd_serve_execute(args: argparse.Namespace) -> int: + from importlib.util import find_spec + + missing = [ + name + for name in ("fastapi", "uvicorn", "openadapt_types") + if find_spec(name) is None + ] + if missing: + raise SystemExit( + f"serve-execute needs {', '.join(missing)} — install the " + "execute extra: pip install 'openadapt-flow[execute]'" + ) + from openadapt_flow.execute import SELF_SIGNED_NOTICE + from openadapt_flow.execute.app import serve + from openadapt_flow.execute.service import ExecuteService, default_data_dir + + data_dir = Path(args.data_dir) if args.data_dir else default_data_dir() + store = ExecuteService( + data_dir, + token=args.token, + seed_mockmed=bool(args.seed_mockmed), + ) + print("openadapt-flow reference Execute") + print(f" http://{args.host}:{args.port}") + print(f" data dir {data_dir}") + print(f" token {store.token}") + print(" issuer self_signed") + print(f" fingerprint {store.fingerprint}") + print(f" {SELF_SIGNED_NOTICE}") + if args.seed_mockmed: + print(" seeded MockMed admissions (ok + banner-lie)") + serve( + data_dir, + host=args.host, + port=args.port, + token=store.token, + service=store, + ) + return 0 + + def _connector_flags(args: argparse.Namespace) -> dict[str, object]: """Collect the connector CLI flags into the settings-resolution dict.""" keys = ( diff --git a/openadapt_flow/execute/__init__.py b/openadapt_flow/execute/__init__.py new file mode 100644 index 00000000..d103c0ff --- /dev/null +++ b/openadapt_flow/execute/__init__.py @@ -0,0 +1,24 @@ +"""MIT reference Execute server: one process, local self-signed receipts. + +This is the engine-side host for the public Execute v1 request schema. It is +complete for one operator on one machine. It is not OpenAdapt Cloud: receipts +are self-signed with a local Ed25519 key, never an OpenAdapt production Seal. +""" + +from __future__ import annotations + +from typing import Literal + +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 8787 +DEFAULT_DATA_DIRNAME = "execute-ref" +SELF_SIGNED_NOTICE: Literal[ + "Self-signed. Counterparties that require an OpenAdapt Seal still use Cloud." +] = "Self-signed. Counterparties that require an OpenAdapt Seal still use Cloud." + +__all__ = [ + "DEFAULT_HOST", + "DEFAULT_PORT", + "DEFAULT_DATA_DIRNAME", + "SELF_SIGNED_NOTICE", +] diff --git a/openadapt_flow/execute/app.py b/openadapt_flow/execute/app.py new file mode 100644 index 00000000..f51e65ce --- /dev/null +++ b/openadapt_flow/execute/app.py @@ -0,0 +1,330 @@ +"""HTTP + MCP surface for the MIT reference Execute server.""" + +from __future__ import annotations + +import hmac +import html +import json +from pathlib import Path +from typing import Any, Optional + +from fastapi import FastAPI, Header, HTTPException, Request +from fastapi.responses import HTMLResponse, JSONResponse, Response + +from openadapt_flow import __version__ +from openadapt_flow.execute.dispatch import Runner +from openadapt_flow.execute.models import SelfSignedSealV1 +from openadapt_flow.execute.service import ExecuteService, ExecuteServiceError + +_JSON = "application/json" +_MCP_PROTOCOL = "2024-11-05" + + +def create_app( + data_dir: Path | str, + *, + token: str | None = None, + runner: Runner | None = None, + process_inline: bool = True, + seed_mockmed: bool = False, + service: ExecuteService | None = None, +) -> FastAPI: + """Build the one-process HTTP+MCP app over a local data directory.""" + + store = service or ExecuteService( + data_dir, + token=token, + runner=runner, + process_inline=process_inline, + seed_mockmed=seed_mockmed, + ) + app = FastAPI( + title="OpenAdapt reference Execute", + version=__version__, + docs_url=None, + redoc_url=None, + openapi_url=None, + ) + app.state.execute = store + + @app.exception_handler(ExecuteServiceError) + async def _service_error( + _request: Request, exc: ExecuteServiceError + ) -> JSONResponse: + return JSONResponse(status_code=exc.status_code, content=exc.body()) + + @app.get("/health") + def health() -> dict[str, Any]: + return { + "status": "ok", + "service": "openadapt-execute-ref", + "issuer": "self_signed", + "issuer_key_fingerprint": store.fingerprint, + "production_seal": False, + } + + @app.post("/v1/executions", status_code=202, response_model=None) + async def create_execution( + request: Request, + authorization: Optional[str] = Header(default=None), + ) -> JSONResponse: + _require_bearer(store, authorization) + payload = await _json_object(request) + accepted = store.create_execution(payload) + return JSONResponse( + status_code=202, + content=accepted.model_dump(mode="json"), + ) + + @app.get("/v1/executions/{execution_id}") + def get_execution( + execution_id: str, + authorization: Optional[str] = Header(default=None), + ) -> dict[str, Any]: + _require_bearer(store, authorization) + return store.get_status(execution_id).model_dump(mode="json") + + @app.get("/v1/executions/{execution_id}/receipt", response_model=None) + def get_receipt( + execution_id: str, + authorization: Optional[str] = Header(default=None), + ) -> JSONResponse: + _require_bearer(store, authorization) + receipt = store.get_receipt(execution_id) + body = receipt.model_dump(mode="json") + return JSONResponse( + content=body, + headers=_issuer_headers(store), + ) + + @app.get("/seals/{seal_id}", response_model=None) + def get_seal( + seal_id: str, + request: Request, + authorization: Optional[str] = Header(default=None), + ) -> Response: + # Local analog of a public verify page: the seal has no PHI. + del authorization + seal = store.get_seal(seal_id) + accept = request.headers.get("accept", "") + want_json = request.query_params.get("format") == "json" or ( + _JSON in accept and "text/html" not in accept + ) + if want_json: + return JSONResponse( + content=seal.model_dump(mode="json"), + headers=_issuer_headers(store), + ) + return HTMLResponse(_render_seal_html(seal), headers=_issuer_headers(store)) + + @app.post("/mcp", response_model=None) + async def mcp_endpoint( + request: Request, + authorization: Optional[str] = Header(default=None), + ) -> JSONResponse | Response: + _require_bearer(store, authorization) + payload = await _json_object(request) + result = _handle_mcp(store, payload) + if result is None: + return Response(status_code=204) + return JSONResponse(content=result) + + return app + + +def serve( + data_dir: Path | str, + *, + host: str = "127.0.0.1", + port: int = 8787, + token: str | None = None, + seed_mockmed: bool = False, + service: ExecuteService | None = None, +) -> None: + """Block on uvicorn. Caller prints the banner before this.""" + + import uvicorn + + app = create_app( + data_dir, + token=token, + seed_mockmed=seed_mockmed, + service=service, + ) + uvicorn.run(app, host=host, port=port, log_level="info") + + +def _require_bearer(store: ExecuteService, authorization: Optional[str]) -> None: + scheme, separator, token = (authorization or "").partition(" ") + if separator != " " or scheme.lower() != "bearer": + raise HTTPException(status_code=401, detail="bearer token required") + if not hmac.compare_digest(token.strip(), store.token): + raise HTTPException(status_code=401, detail="invalid bearer token") + + +def _issuer_headers(store: ExecuteService) -> dict[str, str]: + return { + "X-OpenAdapt-Issuer": "self_signed", + "X-OpenAdapt-Issuer-Fingerprint": store.fingerprint, + "X-OpenAdapt-Production-Seal": "false", + } + + +async def _json_object(request: Request) -> dict[str, Any]: + try: + payload = await request.json() + except Exception as exc: + raise HTTPException(status_code=400, detail="body must be JSON") from exc + if not isinstance(payload, dict): + raise HTTPException(status_code=400, detail="body must be a JSON object") + return payload + + +def _handle_mcp( + store: ExecuteService, payload: dict[str, Any] +) -> dict[str, Any] | None: + if payload.get("jsonrpc") != "2.0": + return { + "jsonrpc": "2.0", + "id": payload.get("id"), + "error": {"code": -32600, "message": "jsonrpc must be 2.0"}, + } + method = payload.get("method") + rpc_id = payload.get("id") + params = payload.get("params") or {} + if method == "notifications/initialized": + return None + if method == "initialize": + return { + "jsonrpc": "2.0", + "id": rpc_id, + "result": { + "protocolVersion": _MCP_PROTOCOL, + "capabilities": {"tools": {}}, + "serverInfo": { + "name": "openadapt-execute-ref", + "version": __version__, + }, + }, + } + if method == "ping": + return {"jsonrpc": "2.0", "id": rpc_id, "result": {}} + if method == "tools/list": + return { + "jsonrpc": "2.0", + "id": rpc_id, + "result": {"tools": _mcp_tools()}, + } + if method == "tools/call": + name = params.get("name") + arguments = params.get("arguments") or {} + try: + text = _call_mcp_tool(store, str(name), arguments) + except ExecuteServiceError as exc: + return { + "jsonrpc": "2.0", + "id": rpc_id, + "result": { + "content": [{"type": "text", "text": json.dumps(exc.body())}], + "isError": True, + }, + } + return { + "jsonrpc": "2.0", + "id": rpc_id, + "result": {"content": [{"type": "text", "text": text}]}, + } + return { + "jsonrpc": "2.0", + "id": rpc_id, + "error": {"code": -32601, "message": f"unknown method {method!r}"}, + } + + +def _mcp_tools() -> list[dict[str, Any]]: + from openadapt_types.execute import ExecuteRequestV1 + + request_schema = ExecuteRequestV1.model_json_schema() + return [ + { + "name": "create_execution", + "description": ( + "Submit one qualified execution. Same body as POST /v1/executions." + ), + "inputSchema": request_schema, + }, + { + "name": "get_execution", + "description": "Read Execute lifecycle state.", + "inputSchema": { + "type": "object", + "properties": {"execution_id": {"type": "string"}}, + "required": ["execution_id"], + }, + }, + { + "name": "get_execution_receipt", + "description": "Read the terminal Execute receipt.", + "inputSchema": { + "type": "object", + "properties": {"execution_id": {"type": "string"}}, + "required": ["execution_id"], + }, + }, + ] + + +def _call_mcp_tool(store: ExecuteService, name: str, arguments: dict[str, Any]) -> str: + if name == "create_execution": + accepted = store.create_execution(arguments) + return json.dumps(accepted.model_dump(mode="json")) + if name == "get_execution": + status = store.get_status(str(arguments.get("execution_id") or "")) + return json.dumps(status.model_dump(mode="json")) + if name == "get_execution_receipt": + receipt = store.get_receipt(str(arguments.get("execution_id") or "")) + return json.dumps(receipt.model_dump(mode="json")) + raise ExecuteServiceError(404, "unknown_tool", f"no MCP tool named {name!r}") + + +def _render_seal_html(seal: SelfSignedSealV1) -> str: + receipt = seal.receipt + outcome = html.escape(str(receipt.get("outcome", ""))) + receipt_id = html.escape(str(receipt.get("receipt_id", ""))) + execution_id = html.escape(str(receipt.get("execution_id", ""))) + digest = html.escape(str(receipt.get("workflow_digest", ""))) + fingerprint = html.escape(seal.issuer_key_fingerprint) + notice = html.escape(seal.notice) + return f""" + + + + Self-signed Execute receipt + + + + +

Local Execute receipt

+

{notice}

+
+
outcome
{outcome}
+
issuer
self_signed
+
key fingerprint
{fingerprint}
+
production seal
false
+
meter USD
0
+
verify host
local
+
receipt id
{receipt_id}
+
execution id
{execution_id}
+
workflow digest
{digest}
+
+ + +""" diff --git a/openadapt_flow/execute/dispatch.py b/openadapt_flow/execute/dispatch.py new file mode 100644 index 00000000..9c7d98e6 --- /dev/null +++ b/openadapt_flow/execute/dispatch.py @@ -0,0 +1,356 @@ +"""Dispatch an admitted Execute request to the local openadapt-flow runner. + +The customer runner is this process. Synthetic MockMed admissions (the test +and ``--seed-mockmed`` path) never launch a browser. A digest-pinned compiled +bundle is replayed in-process under the standard profile. +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Optional + +from openadapt_types.execute import ( + EffectStrengthV1, + ExecuteRequestV1, + ExecuteTerminalOutcomeV1, +) + +from openadapt_flow.execute.models import AdmittedBundle + +Runner = Callable[["AdmittedBundle", "ExecuteRequestV1", Path], "DispatchResult"] + + +class DispatchError(RuntimeError): + """Local replay could not run or could not be classified.""" + + +@dataclass(frozen=True) +class DispatchResult: + """Closed projection of a local run onto the Execute receipt contract.""" + + outcome: ExecuteTerminalOutcomeV1 + authorization_passed: bool + identity_passed: bool + postcondition_passed: bool + effect_passed: bool + minimum_effect_strength: EffectStrengthV1 + observed_effect_strength: EffectStrengthV1 | None + model_used: bool + external_network_used: bool + delivery_uncertain: bool + compensation_effect_verified: bool + workflow_digest: str + evidence_digest: str + + +def default_runner( + admission: AdmittedBundle, + request: ExecuteRequestV1, + run_dir: Path, +) -> DispatchResult: + """Run synthetic MockMed or local governed replay.""" + + if admission.synthetic or not admission.bundle_dir: + return synthetic_mockmed(admission, request) + return live_replay(admission, request, run_dir) + + +def synthetic_mockmed( + admission: AdmittedBundle, request: ExecuteRequestV1 +) -> DispatchResult: + """Project the MockMed tutorial outcomes without a browser. + + Honest environment: ``verified`` at independent-system strength. + Banner-lie / ``break_it`` environment: the screen postcondition passes, + the independent store does not, and the outcome is + ``reconciliation_required`` at $0. That is the tutorial ``--break-it`` + classification, not a production Seal. + """ + + strength = EffectStrengthV1(request.minimum_effect_strength) + if admission.break_it or _request_break_it(request): + return _result( + outcome=ExecuteTerminalOutcomeV1.RECONCILIATION_REQUIRED, + authorization_passed=True, + identity_passed=True, + postcondition_passed=True, + effect_passed=False, + minimum_effect_strength=strength, + observed_effect_strength=None, + delivery_uncertain=True, + workflow_digest=request.workflow_digest, + evidence_tag="mockmed-banner-lie", + ) + return _result( + outcome=ExecuteTerminalOutcomeV1.VERIFIED, + authorization_passed=True, + identity_passed=True, + postcondition_passed=True, + effect_passed=True, + minimum_effect_strength=strength, + observed_effect_strength=EffectStrengthV1.INDEPENDENT_SYSTEM_OF_RECORD, + delivery_uncertain=False, + workflow_digest=request.workflow_digest, + evidence_tag="mockmed-verified", + ) + + +def live_replay( + admission: AdmittedBundle, + request: ExecuteRequestV1, + run_dir: Path, +) -> DispatchResult: + """Replay a compiled bundle in this process under the standard profile.""" + + bundle = Path(admission.bundle_dir or "") + if not bundle.is_dir(): + raise DispatchError(f"admitted bundle_dir is missing: {bundle}") + + from openadapt_flow.ir import Workflow + from openadapt_flow.mockmed.fault_server import serve as serve_mockmed + from openadapt_flow.tutorial import ( + TUTORIAL_BREAK_ENTRY_QUERY, + TUTORIAL_ENTRY_QUERY, + run_tutorial_workflow, + ) + + workflow = Workflow.load(bundle) + run_dir.mkdir(parents=True, exist_ok=True) + stop: Optional[Callable[[], None]] = None + try: + if admission.target_url: + base_url = admission.target_url + entry_query = ( + TUTORIAL_BREAK_ENTRY_QUERY + if admission.break_it + else TUTORIAL_ENTRY_QUERY + ) + report = run_tutorial_workflow( + base_url=base_url, + workflow=workflow, + bundle_dir=bundle, + run_dir=run_dir, + headed=False, + entry_query=entry_query, + ) + else: + base_url, _db, stop = serve_mockmed(port=0) + entry_query = ( + TUTORIAL_BREAK_ENTRY_QUERY + if admission.break_it + else TUTORIAL_ENTRY_QUERY + ) + report = run_tutorial_workflow( + base_url=base_url, + workflow=workflow, + bundle_dir=bundle, + run_dir=run_dir, + headed=False, + entry_query=entry_query, + ) + except Exception as exc: + raise DispatchError(f"local replay failed: {exc}") from exc + finally: + if stop is not None: + stop() + return project_run_report(report, workflow_digest=request.workflow_digest) + + +def project_run_report(report: Any, *, workflow_digest: str) -> DispatchResult: + """Map a local ``RunReport`` onto the Execute terminal taxonomy.""" + + from openadapt_flow.transaction import TransactionOutcome + + txn_raw = getattr(report, "transaction_outcome", None) + try: + txn = TransactionOutcome(str(txn_raw)) if txn_raw is not None else None + except ValueError: + txn = None + coarse = str(getattr(report, "execution_outcome", "") or "") + outcome = _map_outcome(coarse, txn) + envelope = getattr(report, "outcome_envelope", None) + auth_ok, ident_ok, post_ok, effect_ok = _contract_booleans(envelope) + observed = _observed_strength(report) + delivery_uncertain = bool( + txn is TransactionOutcome.RECONCILIATION_REQUIRED + or outcome is ExecuteTerminalOutcomeV1.RECONCILIATION_REQUIRED + ) + compensation = outcome is ExecuteTerminalOutcomeV1.ROLLED_BACK_VERIFIED + if outcome is ExecuteTerminalOutcomeV1.VERIFIED and not ( + auth_ok and ident_ok and post_ok and effect_ok and observed is not None + ): + raise DispatchError( + "local run claimed VERIFIED without complete Execute contracts" + ) + return _result( + outcome=outcome, + authorization_passed=auth_ok, + identity_passed=ident_ok, + postcondition_passed=post_ok, + effect_passed=effect_ok, + minimum_effect_strength=_minimum_strength(report), + observed_effect_strength=observed, + model_used=int(getattr(report, "model_calls", 0) or 0) > 0, + external_network_used=int(getattr(report, "model_calls", 0) or 0) > 0, + delivery_uncertain=delivery_uncertain, + compensation_effect_verified=compensation, + workflow_digest=workflow_digest, + evidence_tag=str(getattr(report, "bundle_content_digest", "") or "run"), + ) + + +def _request_break_it(request: ExecuteRequestV1) -> bool: + params = request.parameters + fault = params.get("fault") + if fault == "optimistic": + return True + flag = params.get("break_it") + return flag is True or flag == "true" + + +def _map_outcome(coarse: str, txn: Any) -> ExecuteTerminalOutcomeV1: + from openadapt_flow.transaction import TransactionOutcome + + if txn is TransactionOutcome.VERIFIED or coarse == "VERIFIED": + return ExecuteTerminalOutcomeV1.VERIFIED + if txn is TransactionOutcome.RECONCILIATION_REQUIRED: + return ExecuteTerminalOutcomeV1.RECONCILIATION_REQUIRED + if txn is TransactionOutcome.HALTED_BEFORE_EFFECT: + return ExecuteTerminalOutcomeV1.HALTED_BEFORE_EFFECT + if txn is TransactionOutcome.REJECTED_POLICY: + return ExecuteTerminalOutcomeV1.REJECTED_POLICY + if txn is TransactionOutcome.FAILED_PLATFORM or coarse == "FAILED": + return ExecuteTerminalOutcomeV1.FAILED_PLATFORM + if txn is TransactionOutcome.ROLLED_BACK: + return ExecuteTerminalOutcomeV1.ROLLED_BACK_VERIFIED + if coarse == "HALTED": + return ExecuteTerminalOutcomeV1.HALTED_BEFORE_EFFECT + if coarse == "COMPLETED_UNVERIFIED": + return ExecuteTerminalOutcomeV1.REJECTED_POLICY + return ExecuteTerminalOutcomeV1.FAILED_PLATFORM + + +def _contract_booleans(envelope: Any) -> tuple[bool, bool, bool, bool]: + if envelope is None: + return False, False, False, False + required = envelope.required_contracts + passed = envelope.passed_contracts + + def _exact(name: str) -> bool: + need = int(getattr(required, name, 0) or 0) + got = int(getattr(passed, name, 0) or 0) + if need == 0: + return True + return got == need + + return ( + _exact("authorization"), + _exact("identity"), + _exact("postcondition"), + _exact("effect"), + ) + + +def _observed_strength(report: Any) -> EffectStrengthV1 | None: + from openadapt_flow.verification import VerificationTier + + mapping = { + int(VerificationTier.INDEPENDENT_SYSTEM): ( + EffectStrengthV1.INDEPENDENT_SYSTEM_OF_RECORD + ), + int(VerificationTier.INDEPENDENT_SESSION): EffectStrengthV1.INDEPENDENT_SESSION, + int(VerificationTier.PERSISTED_STATE_REACQUISITION): ( + EffectStrengthV1.PERSISTED_STATE_REACQUISITION + ), + int(VerificationTier.IMMEDIATE_SCREEN): ( + EffectStrengthV1.IMMEDIATE_SCREEN_CONFIRMATION + ), + } + weakest: int | None = None + for result in getattr(report, "results", []) or []: + for evidence in getattr(result, "effect_evidence", []) or []: + if getattr(evidence, "final_verdict", None) != "confirmed": + continue + tier = getattr(evidence, "verification_tier", None) + if tier is None: + continue + value = int(tier) + weakest = value if weakest is None else max(weakest, value) + if weakest is None: + return None + return mapping.get(weakest) + + +def _minimum_strength(report: Any) -> EffectStrengthV1: + from openadapt_flow.verification import VerificationTier + + raw = getattr(report, "governed_minimum_effect_tier", None) + mapping = { + int(VerificationTier.INDEPENDENT_SYSTEM): ( + EffectStrengthV1.INDEPENDENT_SYSTEM_OF_RECORD + ), + int(VerificationTier.INDEPENDENT_SESSION): EffectStrengthV1.INDEPENDENT_SESSION, + int(VerificationTier.PERSISTED_STATE_REACQUISITION): ( + EffectStrengthV1.PERSISTED_STATE_REACQUISITION + ), + int(VerificationTier.IMMEDIATE_SCREEN): ( + EffectStrengthV1.IMMEDIATE_SCREEN_CONFIRMATION + ), + } + if raw in mapping: + return mapping[int(raw)] + return EffectStrengthV1.INDEPENDENT_SYSTEM_OF_RECORD + + +def _result( + *, + outcome: ExecuteTerminalOutcomeV1, + authorization_passed: bool, + identity_passed: bool, + postcondition_passed: bool, + effect_passed: bool, + minimum_effect_strength: EffectStrengthV1, + observed_effect_strength: EffectStrengthV1 | None, + workflow_digest: str, + evidence_tag: str, + model_used: bool = False, + external_network_used: bool = False, + delivery_uncertain: bool = False, + compensation_effect_verified: bool = False, +) -> DispatchResult: + digest_payload = { + "authorization_passed": authorization_passed, + "identity_passed": identity_passed, + "postcondition_passed": postcondition_passed, + "effect_passed": effect_passed, + "evidence_tag": evidence_tag, + "outcome": outcome.value, + "workflow_digest": workflow_digest, + } + evidence_digest = ( + "sha256:" + + hashlib.sha256( + json.dumps(digest_payload, sort_keys=True, separators=(",", ":")).encode( + "utf-8" + ) + ).hexdigest() + ) + return DispatchResult( + outcome=outcome, + authorization_passed=authorization_passed, + identity_passed=identity_passed, + postcondition_passed=postcondition_passed, + effect_passed=effect_passed, + minimum_effect_strength=minimum_effect_strength, + observed_effect_strength=observed_effect_strength, + model_used=model_used, + external_network_used=external_network_used, + delivery_uncertain=delivery_uncertain, + compensation_effect_verified=compensation_effect_verified, + workflow_digest=workflow_digest, + evidence_digest=evidence_digest, + ) diff --git a/openadapt_flow/execute/keys.py b/openadapt_flow/execute/keys.py new file mode 100644 index 00000000..900493d8 --- /dev/null +++ b/openadapt_flow/execute/keys.py @@ -0,0 +1,110 @@ +"""Local Ed25519 key for self-signed Execute receipts. + +The key lives under the operator's data directory. Nothing here talks to +OpenAdapt Cloud or to ``openadapt.ai``. +""" + +from __future__ import annotations + +import hashlib +import os +import stat +from pathlib import Path + +from cryptography.exceptions import InvalidSignature +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.ed25519 import ( + Ed25519PrivateKey, + Ed25519PublicKey, +) + +_KEY_NAME = "ed25519.pem" +_PUB_NAME = "ed25519.pub" + + +def key_dir(data_dir: Path) -> Path: + return data_dir / "keys" + + +def fingerprint_of(public_key: Ed25519PublicKey) -> str: + raw = public_key.public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ) + return "sha256:" + hashlib.sha256(raw).hexdigest() + + +def load_or_create_private_key(data_dir: Path) -> Ed25519PrivateKey: + """Return the local signing key, generating it on first start.""" + + directory = key_dir(data_dir) + directory.mkdir(parents=True, exist_ok=True) + path = directory / _KEY_NAME + if path.is_file(): + loaded = serialization.load_pem_private_key(path.read_bytes(), password=None) + if not isinstance(loaded, Ed25519PrivateKey): + raise ValueError(f"execute-ref key is not Ed25519: {path}") + return loaded + key = Ed25519PrivateKey.generate() + pem = key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + _atomic_write(path, pem, mode=0o600) + pub_hex = ( + key.public_key() + .public_bytes( + encoding=serialization.Encoding.Raw, + format=serialization.PublicFormat.Raw, + ) + .hex() + .encode("ascii") + + b"\n" + ) + _atomic_write(directory / _PUB_NAME, pub_hex, mode=0o644) + return key + + +def sign_bytes(key: Ed25519PrivateKey, payload: bytes) -> str: + return "ed25519:" + key.sign(payload).hex() + + +def verify_signature( + public_key: Ed25519PublicKey, payload: bytes, signature: str +) -> bool: + if not signature.startswith("ed25519:"): + return False + try: + public_key.verify(bytes.fromhex(signature.split(":", 1)[1]), payload) + except (ValueError, TypeError, InvalidSignature): + return False + return True + + +def load_or_create_token(data_dir: Path, explicit: str | None = None) -> str: + """Return the local bearer token, generating it on first start.""" + + if explicit: + token = explicit.strip() + if not token: + raise ValueError("explicit Execute token must not be empty") + return token + path = data_dir / "token" + if path.is_file(): + token = path.read_text(encoding="utf-8").strip() + if not token: + raise ValueError(f"execute-ref token file is empty: {path}") + return token + token = os.urandom(24).hex() + data_dir.mkdir(parents=True, exist_ok=True) + _atomic_write(path, (token + "\n").encode("utf-8"), mode=0o600) + return token + + +def _atomic_write(path: Path, data: bytes, *, mode: int) -> None: + tmp = path.with_name(path.name + ".tmp") + tmp.write_bytes(data) + os.chmod(tmp, mode | stat.S_IRUSR | stat.S_IWUSR) + tmp.replace(path) + os.chmod(path, mode) diff --git a/openadapt_flow/execute/models.py b/openadapt_flow/execute/models.py new file mode 100644 index 00000000..ae96c78a --- /dev/null +++ b/openadapt_flow/execute/models.py @@ -0,0 +1,111 @@ +"""Local Execute models: admitted bundles and the self-signed seal envelope. + +The portable receipt body is ``ExecuteEvidenceReceiptV1`` from +``openadapt-types``. This module adds only the local issuer wrapper. Extra +keys are forbidden. Screenshot, OCR, parameter, and URL fields have no place +here; see :mod:`openadapt_flow.receipt` for the same allow-list discipline. +""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictFloat, StrictStr + +from openadapt_flow.execute import SELF_SIGNED_NOTICE + +EXECUTE_ADMISSION_SCHEMA: Literal["openadapt.execute-admission/v1"] = ( + "openadapt.execute-admission/v1" +) +SELF_SIGNED_SEAL_SCHEMA: Literal["openadapt.execute-self-signed-seal/v1"] = ( + "openadapt.execute-self-signed-seal/v1" +) + +#: Fields a shareable Execute receipt must never carry. The portable types +#: model already uses ``extra="forbid"``; this set is the regression net for +#: the local projector and the stored JSON. +FORBIDDEN_RECEIPT_KEYS = frozenset( + { + "screenshot", + "screenshots", + "ocr", + "ocr_text", + "typed_value", + "typed_values", + "parameters", + "parameter", + "url", + "hostname", + "coordinate", + "coordinates", + "halt_reason", + "application_name", + "organization_name", + "user_name", + "workflow_name", + "phi", + "image", + "after_png", + "before_png", + "step_intent", + "note", + "record_id", + } +) + + +class _Strict(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class AdmittedBundle(_Strict): + """One digest-pinned qualification an operator has admitted on this machine.""" + + schema_version: Literal["openadapt.execute-admission/v1"] = EXECUTE_ADMISSION_SCHEMA + qualification_id: StrictStr + workflow_version: StrictStr + workflow_digest: StrictStr = Field(pattern=r"^sha256:[0-9a-f]{64}$") + environment_id: StrictStr + minimum_effect_strength: StrictStr + bundle_dir: StrictStr | None = None + target_url: StrictStr | None = None + synthetic: StrictBool = False + break_it: StrictBool = False + policy: StrictStr = "clinical-write" + + +class SelfSignedSealV1(_Strict): + """Local verify envelope around a portable Execute receipt. + + ``production_seal`` is always false. ``issuer`` is always ``self_signed``. + Cloud's OpenAdapt Seal is a different artifact on a different host. + """ + + schema_version: Literal["openadapt.execute-self-signed-seal/v1"] = ( + SELF_SIGNED_SEAL_SCHEMA + ) + issuer: Literal["self_signed"] = "self_signed" + issuer_key_fingerprint: StrictStr = Field(pattern=r"^sha256:[0-9a-f]{64}$") + signature: StrictStr = Field(pattern=r"^ed25519:[0-9a-f]{128}$") + production_seal: Literal[False] = False + verify_host: Literal["local"] = "local" + meter_usd: StrictFloat = Field(ge=0.0, le=0.0) + notice: Literal[ + "Self-signed. Counterparties that require an OpenAdapt Seal still use Cloud." + ] = SELF_SIGNED_NOTICE + receipt: dict[str, Any] + + +def assert_no_forbidden_keys(payload: dict[str, Any]) -> None: + """Refuse a receipt dict that carries a PHI or screenshot field.""" + + extra = FORBIDDEN_RECEIPT_KEYS.intersection(payload) + if extra: + names = ", ".join(sorted(extra)) + raise ValueError(f"execute receipt forbids extra/PHI keys: {names}") + nested = payload.get("contracts") + if isinstance(nested, dict): + extra = FORBIDDEN_RECEIPT_KEYS.intersection(nested) + if extra: + names = ", ".join(sorted(extra)) + raise ValueError(f"execute receipt contracts forbid keys: {names}") diff --git a/openadapt_flow/execute/registry.py b/openadapt_flow/execute/registry.py new file mode 100644 index 00000000..08b4e860 --- /dev/null +++ b/openadapt_flow/execute/registry.py @@ -0,0 +1,104 @@ +"""Directory-backed admitted-bundle registry, pinned by workflow digest.""" + +from __future__ import annotations + +import json +from pathlib import Path + +from openadapt_flow.execute.models import AdmittedBundle + +# Well-known MockMed / tutorial admissions used by tests and ``--seed-mockmed``. +MOCKMED_QUALIFICATION_ID = "qualification_mockmed01" +MOCKMED_WORKFLOW_VERSION = "workflow_tutorial1" +MOCKMED_WORKFLOW_DIGEST = "sha256:" + "c" * 64 +MOCKMED_ENVIRONMENT_OK = "environment_mockmed_ok" +MOCKMED_ENVIRONMENT_LIE = "environment_mockmed_lie" +MOCKMED_EFFECT_STRENGTH = "independent_system_of_record" + + +class AdmissionError(ValueError): + """The request does not match an admitted bundle exactly.""" + + +def admissions_dir(data_dir: Path) -> Path: + return data_dir / "admissions" + + +def load_admissions(data_dir: Path) -> tuple[AdmittedBundle, ...]: + directory = admissions_dir(data_dir) + if not directory.is_dir(): + return () + loaded: list[AdmittedBundle] = [] + for path in sorted(directory.glob("*.json")): + payload = json.loads(path.read_text(encoding="utf-8")) + loaded.append(AdmittedBundle.model_validate(payload)) + return tuple(loaded) + + +def lookup_admission( + data_dir: Path, + *, + qualification_id: str, + workflow_version: str, + workflow_digest: str, + environment_id: str, + minimum_effect_strength: str, +) -> AdmittedBundle: + """Return the unique admission that matches every binding field.""" + + matches = [ + item + for item in load_admissions(data_dir) + if item.qualification_id == qualification_id + and item.workflow_version == workflow_version + and item.workflow_digest == workflow_digest + and item.environment_id == environment_id + and item.minimum_effect_strength == minimum_effect_strength + ] + if not matches: + raise AdmissionError( + "request does not match an admitted bundle exactly " + "(qualification_id, workflow_version, workflow_digest, " + "environment_id, minimum_effect_strength)" + ) + if len(matches) > 1: + raise AdmissionError("multiple admitted bundles match this request") + return matches[0] + + +def write_admission(data_dir: Path, admission: AdmittedBundle) -> Path: + directory = admissions_dir(data_dir) + directory.mkdir(parents=True, exist_ok=True) + slug = f"{admission.qualification_id}__{admission.environment_id}.json" + path = directory / slug + path.write_text( + admission.model_dump_json(indent=2) + "\n", + encoding="utf-8", + ) + return path + + +def seed_mockmed_admissions(data_dir: Path) -> tuple[AdmittedBundle, AdmittedBundle]: + """Write the synthetic MockMed ok + banner-lie admissions.""" + + honest = AdmittedBundle( + qualification_id=MOCKMED_QUALIFICATION_ID, + workflow_version=MOCKMED_WORKFLOW_VERSION, + workflow_digest=MOCKMED_WORKFLOW_DIGEST, + environment_id=MOCKMED_ENVIRONMENT_OK, + minimum_effect_strength=MOCKMED_EFFECT_STRENGTH, + synthetic=True, + break_it=False, + ) + lie = AdmittedBundle( + qualification_id=MOCKMED_QUALIFICATION_ID, + workflow_version=MOCKMED_WORKFLOW_VERSION, + workflow_digest=MOCKMED_WORKFLOW_DIGEST, + environment_id=MOCKMED_ENVIRONMENT_LIE, + minimum_effect_strength=MOCKMED_EFFECT_STRENGTH, + synthetic=True, + break_it=True, + ) + write_admission(data_dir, honest) + write_admission(data_dir, lie) + return honest, lie diff --git a/openadapt_flow/execute/service.py b/openadapt_flow/execute/service.py new file mode 100644 index 00000000..583ee3e9 --- /dev/null +++ b/openadapt_flow/execute/service.py @@ -0,0 +1,378 @@ +"""Durable local Execute store: idempotency, dispatch, self-signed receipts.""" + +from __future__ import annotations + +import json +import threading +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Literal, Optional, cast +from uuid import uuid4 + +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from openadapt_types.execute import ( + ExecuteAcceptedV1, + ExecuteEvidenceContractV1, + ExecuteEvidenceReceiptV1, + ExecuteLifecycleStateV1, + ExecuteRequestV1, + ExecuteStatusV1, + ExecuteTerminalOutcomeV1, +) +from openadapt_types.oracle import oracle_tier_from_effect_strength +from pydantic import ValidationError + +from openadapt_flow.execute.dispatch import ( + DispatchResult, + Runner, + default_runner, +) +from openadapt_flow.execute.keys import ( + fingerprint_of, + load_or_create_private_key, + load_or_create_token, + sign_bytes, +) +from openadapt_flow.execute.models import ( + SelfSignedSealV1, + assert_no_forbidden_keys, +) +from openadapt_flow.execute.registry import ( + AdmissionError, + lookup_admission, + seed_mockmed_admissions, +) + + +class ExecuteServiceError(Exception): + """Typed failure with an HTTP status for the reference server.""" + + def __init__(self, status_code: int, error: str, detail: str) -> None: + super().__init__(detail) + self.status_code = status_code + self.error = error + self.detail = detail + + def body(self) -> dict[str, str]: + return {"error": self.error, "detail": self.detail} + + +class ExecuteService: + """One-operator Execute store on the local filesystem.""" + + def __init__( + self, + data_dir: Path | str, + *, + token: str | None = None, + runner: Runner | None = None, + process_inline: bool = True, + seed_mockmed: bool = False, + ) -> None: + self.data_dir = Path(data_dir).expanduser() + self.data_dir.mkdir(parents=True, exist_ok=True) + self._lock = threading.RLock() + self._key: Ed25519PrivateKey = load_or_create_private_key(self.data_dir) + self.token = load_or_create_token(self.data_dir, token) + self.fingerprint = fingerprint_of(self._key.public_key()) + self.runner = runner or default_runner + self.process_inline = process_inline + if seed_mockmed: + seed_mockmed_admissions(self.data_dir) + (self.data_dir / "executions").mkdir(parents=True, exist_ok=True) + (self.data_dir / "idempotency").mkdir(parents=True, exist_ok=True) + + def create_execution(self, payload: dict[str, Any]) -> ExecuteAcceptedV1: + try: + request = ExecuteRequestV1.model_validate(payload) + except ValidationError as exc: + raise ExecuteServiceError(422, "invalid_request", str(exc)) from exc + canonical = _canonical_request(request) + digest = _sha256_hex(canonical) + with self._lock: + existing = self._read_idempotency(request.idempotency_key) + if existing is not None: + if existing["request_digest"] != digest: + raise ExecuteServiceError( + 409, + "idempotency_conflict", + "idempotency key already bound to a different request", + ) + return ExecuteAcceptedV1(execution_id=existing["execution_id"]) + try: + lookup_admission( + self.data_dir, + qualification_id=request.qualification_id, + workflow_version=request.workflow_version, + workflow_digest=request.workflow_digest, + environment_id=request.environment_id, + minimum_effect_strength=request.minimum_effect_strength.value, + ) + except AdmissionError as exc: + raise ExecuteServiceError( + 422, "qualification_mismatch", str(exc) + ) from exc + execution_id = _new_id("execution") + now = _now() + self._write_json( + self._execution_path(execution_id) / "request.json", + json.loads(canonical), + ) + status = ExecuteStatusV1( + execution_id=execution_id, + state=ExecuteLifecycleStateV1.QUEUED, + updated_at=now, + ) + self._write_status(status) + self._write_json( + self.data_dir / "idempotency" / f"{request.idempotency_key}.json", + {"execution_id": execution_id, "request_digest": digest}, + ) + self._start(execution_id, request) + return ExecuteAcceptedV1(execution_id=execution_id) + + def get_status(self, execution_id: str) -> ExecuteStatusV1: + path = self._execution_path(execution_id) / "status.json" + if not path.is_file(): + raise ExecuteServiceError(404, "not_found", "no such execution") + return ExecuteStatusV1.model_validate(json.loads(path.read_text("utf-8"))) + + def get_receipt(self, execution_id: str) -> ExecuteEvidenceReceiptV1: + status = self.get_status(execution_id) + if status.state is not ExecuteLifecycleStateV1.TERMINAL: + raise ExecuteServiceError( + 409, + "receipt_not_ready", + "execution is not terminal", + ) + path = self._execution_path(execution_id) / "receipt.json" + if not path.is_file(): + raise ExecuteServiceError( + 409, + "receipt_not_ready", + "terminal run still waits for trusted evidence", + ) + payload = json.loads(path.read_text(encoding="utf-8")) + assert_no_forbidden_keys(payload) + return ExecuteEvidenceReceiptV1.model_validate(payload) + + def get_seal(self, seal_id: str) -> SelfSignedSealV1: + execution_id = self._execution_id_for_seal(seal_id) + path = self._execution_path(execution_id) / "seal.json" + if not path.is_file(): + raise ExecuteServiceError(404, "not_found", "no such seal") + payload = json.loads(path.read_text(encoding="utf-8")) + assert_no_forbidden_keys(payload.get("receipt") or {}) + return SelfSignedSealV1.model_validate(payload) + + def _start(self, execution_id: str, request: ExecuteRequestV1) -> None: + if self.process_inline: + self._run(execution_id, request) + return + thread = threading.Thread( + target=self._run, + args=(execution_id, request), + name=f"execute-ref-{execution_id}", + daemon=True, + ) + thread.start() + + def _run(self, execution_id: str, request: ExecuteRequestV1) -> None: + try: + self._write_status( + ExecuteStatusV1( + execution_id=execution_id, + state=ExecuteLifecycleStateV1.RUNNING, + updated_at=_now(), + ) + ) + admission = lookup_admission( + self.data_dir, + qualification_id=request.qualification_id, + workflow_version=request.workflow_version, + workflow_digest=request.workflow_digest, + environment_id=request.environment_id, + minimum_effect_strength=request.minimum_effect_strength.value, + ) + run_dir = self._execution_path(execution_id) / "run" + result = self.runner(admission, request, run_dir) + self._finalize(execution_id, request, result) + except Exception as exc: + failed = _failed_platform_result(request, str(exc)) + try: + self._finalize(execution_id, request, failed) + except Exception: + self._write_status( + ExecuteStatusV1( + execution_id=execution_id, + state=ExecuteLifecycleStateV1.RUNNING, + updated_at=_now(), + ) + ) + + def _finalize( + self, + execution_id: str, + request: ExecuteRequestV1, + result: DispatchResult, + ) -> None: + receipt_id = _new_id("receipt") + issued_at = _now() + contracts = ExecuteEvidenceContractV1( + authorization_passed=result.authorization_passed, + identity_passed=result.identity_passed, + postcondition_passed=result.postcondition_passed, + effect_passed=result.effect_passed, + minimum_effect_strength=result.minimum_effect_strength, + observed_effect_strength=result.observed_effect_strength, + model_used=result.model_used, + external_network_used=result.external_network_used, + ) + receipt = ExecuteEvidenceReceiptV1( + receipt_id=receipt_id, + execution_id=execution_id, + workflow_digest=result.workflow_digest, + workflow_version=request.workflow_version, + qualification_id=request.qualification_id, + environment_id=request.environment_id, + runner_id=_new_id("runner"), + nonce=_new_id("nonce"), + oracle_tier=_oracle_tier(result.observed_effect_strength), + outcome=result.outcome, + contracts=contracts, + delivery_uncertain=result.delivery_uncertain, + compensation_effect_verified=result.compensation_effect_verified, + evidence_digest=result.evidence_digest, + issued_at=issued_at, + ) + payload = receipt.model_dump(mode="json") + assert_no_forbidden_keys(payload) + canonical = _canonical_json(payload) + signature = sign_bytes(self._key, canonical) + seal = SelfSignedSealV1( + issuer_key_fingerprint=self.fingerprint, + signature=signature, + production_seal=False, + meter_usd=0.0, + receipt=payload, + ) + directory = self._execution_path(execution_id) + self._write_json(directory / "receipt.json", payload) + self._write_json(directory / "seal.json", seal.model_dump(mode="json")) + self._write_json( + directory / "seal-index.json", + {"receipt_id": receipt_id, "execution_id": execution_id}, + ) + self._write_status( + ExecuteStatusV1( + execution_id=execution_id, + state=ExecuteLifecycleStateV1.TERMINAL, + terminal_outcome=result.outcome, + evidence_receipt_id=receipt_id, + updated_at=issued_at, + ) + ) + + def _execution_id_for_seal(self, seal_id: str) -> str: + direct = self.data_dir / "executions" / seal_id / "seal.json" + if direct.is_file(): + return seal_id + for status_path in (self.data_dir / "executions").glob("*/status.json"): + try: + status = ExecuteStatusV1.model_validate( + json.loads(status_path.read_text(encoding="utf-8")) + ) + except (OSError, ValidationError, json.JSONDecodeError): + continue + if status.evidence_receipt_id == seal_id: + return status.execution_id + raise ExecuteServiceError(404, "not_found", "no such seal") + + def _execution_path(self, execution_id: str) -> Path: + return self.data_dir / "executions" / execution_id + + def _read_idempotency(self, key: str) -> Optional[dict[str, str]]: + path = self.data_dir / "idempotency" / f"{key}.json" + if not path.is_file(): + return None + payload = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(payload, dict): + return None + return { + "execution_id": str(payload["execution_id"]), + "request_digest": str(payload["request_digest"]), + } + + def _write_status(self, status: ExecuteStatusV1) -> None: + self._write_json( + self._execution_path(status.execution_id) / "status.json", + status.model_dump(mode="json"), + ) + + def _write_json(self, path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_name(path.name + ".tmp") + tmp.write_text( + json.dumps(payload, indent=2, sort_keys=True) + "\n", + encoding="utf-8", + ) + tmp.replace(path) + + +def _oracle_tier(observed: object | None) -> Literal[0, 1, 2, 3]: + return cast( + Literal[0, 1, 2, 3], + int(oracle_tier_from_effect_strength(observed)), + ) + + +def _new_id(prefix: str) -> str: + return f"{prefix}_{uuid4().hex}" + + +def _now() -> str: + return ( + datetime.now(timezone.utc) + .replace(microsecond=0) + .isoformat() + .replace("+00:00", "Z") + ) + + +def _canonical_request(request: ExecuteRequestV1) -> bytes: + return _canonical_json(request.model_dump(mode="json")) + + +def _canonical_json(payload: dict[str, Any]) -> bytes: + return json.dumps( + payload, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=True, + ).encode("utf-8") + + +def _sha256_hex(data: bytes) -> str: + import hashlib + + return hashlib.sha256(data).hexdigest() + + +def _failed_platform_result(request: ExecuteRequestV1, tag: str) -> DispatchResult: + from openadapt_flow.execute.dispatch import _result + + return _result( + outcome=ExecuteTerminalOutcomeV1.FAILED_PLATFORM, + authorization_passed=False, + identity_passed=False, + postcondition_passed=False, + effect_passed=False, + minimum_effect_strength=request.minimum_effect_strength, + observed_effect_strength=None, + workflow_digest=request.workflow_digest, + evidence_tag=f"platform-fault:{tag[:64]}", + ) + + +def default_data_dir() -> Path: + return Path.home() / ".openadapt" / "execute-ref" diff --git a/pyproject.toml b/pyproject.toml index e2e14eb6..1798314c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -101,6 +101,14 @@ console = [ "uvicorn>=0.29", "openadapt-types>=0.13.0,<0.14.0", ] +# Reference Execute server: `openadapt-flow serve-execute`. Same public +# request schema as Cloud Execute, hosted in this process with a local +# self-signed receipt. Not the hosted control plane. +execute = [ + "fastapi>=0.110", + "uvicorn>=0.29", + "openadapt-types>=0.13.0,<0.14.0", +] # WindowsBackend: HTTP client for the WAA (Windows Agent Arena) server. windows = ["requests>=2.31", "pywin32>=312; platform_system == 'Windows'"] # Native macOS window capture/input. Imported lazily; other platforms never diff --git a/tests/test_execute_ref.py b/tests/test_execute_ref.py new file mode 100644 index 00000000..a97bf5c2 --- /dev/null +++ b/tests/test_execute_ref.py @@ -0,0 +1,384 @@ +"""MIT reference Execute server: contract, halt, idempotency, receipt, boundary.""" + +from __future__ import annotations + +import json +import threading +import time +from pathlib import Path +from typing import Any + +import pytest +from pydantic import ValidationError + +pytest.importorskip("fastapi") +pytest.importorskip("openadapt_types") + +from fastapi.testclient import TestClient # noqa: E402 +from openadapt_types.execute import ( # noqa: E402 + ExecuteAcceptedV1, + ExecuteEvidenceReceiptV1, + ExecuteRequestV1, + ExecuteStatusV1, + ExecuteTerminalOutcomeV1, +) + +from openadapt_flow.execute.app import create_app # noqa: E402 +from openadapt_flow.execute.dispatch import ( # noqa: E402 + DispatchResult, + synthetic_mockmed, +) +from openadapt_flow.execute.keys import ( # noqa: E402 + fingerprint_of, + load_or_create_private_key, + verify_signature, +) +from openadapt_flow.execute.models import ( # noqa: E402 + FORBIDDEN_RECEIPT_KEYS, + SelfSignedSealV1, + assert_no_forbidden_keys, +) +from openadapt_flow.execute.registry import ( # noqa: E402 + MOCKMED_EFFECT_STRENGTH, + MOCKMED_ENVIRONMENT_LIE, + MOCKMED_ENVIRONMENT_OK, + MOCKMED_QUALIFICATION_ID, + MOCKMED_WORKFLOW_DIGEST, + MOCKMED_WORKFLOW_VERSION, +) +from openadapt_flow.execute.service import ExecuteService # noqa: E402 +from openadapt_flow.receipt import RunReceipt # noqa: E402 + +AUTH_CTX = { + "actor_id": "caller_agent_12345678", + "authorization_reference": "authorization_12345678", +} + + +def _request(**updates: object) -> dict[str, object]: + fields: dict[str, object] = { + "schema_version": "openadapt.execute-request/v1", + "qualification_id": MOCKMED_QUALIFICATION_ID, + "workflow_version": MOCKMED_WORKFLOW_VERSION, + "workflow_digest": MOCKMED_WORKFLOW_DIGEST, + "environment_id": MOCKMED_ENVIRONMENT_OK, + "parameters": {"date": "2026-08-15", "record": {"id": "12345"}}, + "idempotency_key": "caller_key_12345678", + "authorization_context": AUTH_CTX, + "effect_strength_schema_version": "1", + "minimum_effect_strength": MOCKMED_EFFECT_STRENGTH, + } + fields.update(updates) + return fields + + +def _client(tmp_path: Path, **kwargs: Any) -> tuple[TestClient, ExecuteService]: + kwargs.setdefault("seed_mockmed", True) + kwargs.setdefault("process_inline", True) + app = create_app(tmp_path, token="test-token", **kwargs) + store: ExecuteService = app.state.execute + client = TestClient(app) + client.headers["Authorization"] = "Bearer test-token" + return client, store + + +def test_cli_wires_serve_execute() -> None: + from openadapt_flow.__main__ import build_parser + + parser = build_parser() + args = parser.parse_args( + ["serve-execute", "--port", "8787", "--seed-mockmed", "--data-dir", "/tmp/x"] + ) + assert args.command == "serve-execute" + assert args.port == 8787 + assert args.seed_mockmed is True + assert args.func.__name__ == "_cmd_serve_execute" + + +def test_health_needs_no_token(tmp_path: Path) -> None: + client, store = _client(tmp_path) + bare = TestClient(client.app) + response = bare.get("/health") + assert response.status_code == 200 + body = response.json() + assert body["status"] == "ok" + assert body["issuer"] == "self_signed" + assert body["production_seal"] is False + assert body["issuer_key_fingerprint"] == store.fingerprint + + +def test_post_mockmed_verified_receipt(tmp_path: Path) -> None: + client, store = _client(tmp_path) + accepted = ExecuteAcceptedV1.model_validate( + client.post("/v1/executions", json=_request()).json() + ) + assert accepted.state.value == "queued" + status = ExecuteStatusV1.model_validate( + client.get(f"/v1/executions/{accepted.execution_id}").json() + ) + assert status.state.value == "terminal" + assert status.terminal_outcome is ExecuteTerminalOutcomeV1.VERIFIED + receipt_response = client.get(f"/v1/executions/{accepted.execution_id}/receipt") + receipt = ExecuteEvidenceReceiptV1.model_validate(receipt_response.json()) + assert receipt.execution_id == accepted.execution_id + assert receipt.receipt_id == status.evidence_receipt_id + assert receipt.workflow_digest == MOCKMED_WORKFLOW_DIGEST + assert receipt.outcome is ExecuteTerminalOutcomeV1.VERIFIED + assert receipt_response.headers["X-OpenAdapt-Issuer"] == "self_signed" + assert receipt_response.headers["X-OpenAdapt-Production-Seal"] == "false" + assert ( + receipt_response.headers["X-OpenAdapt-Issuer-Fingerprint"] == store.fingerprint + ) + seal = SelfSignedSealV1.model_validate( + client.get(f"/seals/{receipt.receipt_id}?format=json").json() + ) + assert seal.issuer == "self_signed" + assert seal.production_seal is False + assert seal.meter_usd == 0.0 + assert seal.verify_host == "local" + key = load_or_create_private_key(tmp_path) + assert fingerprint_of(key.public_key()) == seal.issuer_key_fingerprint + canonical = json.dumps( + seal.receipt, sort_keys=True, separators=(",", ":"), ensure_ascii=True + ).encode("utf-8") + assert verify_signature(key.public_key(), canonical, seal.signature) + + +def test_break_it_banner_lie_is_not_a_production_seal(tmp_path: Path) -> None: + client, _store = _client(tmp_path) + accepted = ExecuteAcceptedV1.model_validate( + client.post( + "/v1/executions", + json=_request( + environment_id=MOCKMED_ENVIRONMENT_LIE, + idempotency_key="caller_key_break_it1", + ), + ).json() + ) + status = ExecuteStatusV1.model_validate( + client.get(f"/v1/executions/{accepted.execution_id}").json() + ) + assert status.state.value == "terminal" + assert status.terminal_outcome is ExecuteTerminalOutcomeV1.RECONCILIATION_REQUIRED + receipt = ExecuteEvidenceReceiptV1.model_validate( + client.get(f"/v1/executions/{accepted.execution_id}/receipt").json() + ) + assert receipt.outcome is ExecuteTerminalOutcomeV1.RECONCILIATION_REQUIRED + assert receipt.outcome is not ExecuteTerminalOutcomeV1.VERIFIED + assert receipt.contracts.effect_passed is False + assert receipt.contracts.postcondition_passed is True + assert receipt.delivery_uncertain is True + seal = SelfSignedSealV1.model_validate( + client.get(f"/seals/{receipt.receipt_id}?format=json").json() + ) + assert seal.meter_usd == 0.0 + assert seal.production_seal is False + assert seal.issuer == "self_signed" + html_page = client.get( + f"/seals/{receipt.receipt_id}", headers={"Accept": "text/html"} + ) + assert html_page.status_code == 200 + text = html_page.text + assert "Self-signed" in text + assert "not an OpenAdapt production Seal" in text + assert "0" in text + + +def test_idempotency_returns_the_same_execution(tmp_path: Path) -> None: + client, _store = _client(tmp_path) + body = _request(idempotency_key="caller_key_same_body1") + first = client.post("/v1/executions", json=body) + second = client.post("/v1/executions", json=body) + assert first.status_code == 202 + assert second.status_code == 202 + assert first.json()["execution_id"] == second.json()["execution_id"] + changed = dict(body) + changed["parameters"] = {"date": "2026-08-16"} + conflict = client.post("/v1/executions", json=changed) + assert conflict.status_code == 409 + assert conflict.json()["error"] == "idempotency_conflict" + + +def test_receipt_409_until_terminal(tmp_path: Path) -> None: + gate = threading.Event() + + def runner(admission, request, run_dir): + gate.wait(timeout=5) + return synthetic_mockmed(admission, request) + + client, _store = _client(tmp_path, runner=runner, process_inline=False) + accepted = client.post( + "/v1/executions", json=_request(idempotency_key="caller_key_slowrun01") + ).json() + execution_id = accepted["execution_id"] + early = client.get(f"/v1/executions/{execution_id}/receipt") + assert early.status_code == 409 + gate.set() + deadline = time.time() + 5 + status = None + while time.time() < deadline: + status = client.get(f"/v1/executions/{execution_id}").json() + if status["state"] == "terminal": + break + time.sleep(0.05) + assert status is not None and status["state"] == "terminal" + receipt = client.get(f"/v1/executions/{execution_id}/receipt") + assert receipt.status_code == 200 + ExecuteEvidenceReceiptV1.model_validate(receipt.json()) + + +def test_receipt_refuses_extra_keys_and_screenshots() -> None: + payload = { + "schema_version": "openadapt.execute-evidence-receipt/v1", + "receipt_id": "receipt_12345678", + "execution_id": "execution_12345678", + "workflow_digest": MOCKMED_WORKFLOW_DIGEST, + "workflow_version": MOCKMED_WORKFLOW_VERSION, + "qualification_id": MOCKMED_QUALIFICATION_ID, + "environment_id": MOCKMED_ENVIRONMENT_OK, + "runner_id": "runner_12345678", + "nonce": "nonce_12345678", + "oracle_tier": 2, + "outcome": "verified", + "contracts": { + "authorization_passed": True, + "identity_passed": True, + "postcondition_passed": True, + "effect_passed": True, + "minimum_effect_strength": MOCKMED_EFFECT_STRENGTH, + "observed_effect_strength": MOCKMED_EFFECT_STRENGTH, + "model_used": False, + "external_network_used": False, + }, + "delivery_uncertain": False, + "evidence_digest": "sha256:" + "b" * 64, + "issued_at": "2026-07-29T12:00:00Z", + } + ExecuteEvidenceReceiptV1.model_validate(payload) + with pytest.raises(ValidationError): + ExecuteEvidenceReceiptV1.model_validate({**payload, "screenshot": "x.png"}) + with pytest.raises(ValidationError): + ExecuteEvidenceReceiptV1.model_validate({**payload, "ocr_text": "secret"}) + with pytest.raises(ValueError, match="forbids"): + assert_no_forbidden_keys({**payload, "screenshot": "x.png"}) + for key in FORBIDDEN_RECEIPT_KEYS: + assert key not in ExecuteEvidenceReceiptV1.model_fields + assert RunReceipt.model_config.get("extra") == "forbid" + assert ExecuteEvidenceReceiptV1.model_config.get("extra") == "forbid" + + +def test_stored_receipt_has_no_screenshot_fields(tmp_path: Path) -> None: + client, _store = _client(tmp_path) + execution_id = client.post("/v1/executions", json=_request()).json()["execution_id"] + receipt = client.get(f"/v1/executions/{execution_id}/receipt").json() + assert FORBIDDEN_RECEIPT_KEYS.isdisjoint(receipt) + assert FORBIDDEN_RECEIPT_KEYS.isdisjoint(receipt["contracts"]) + text = json.dumps(receipt) + assert "screenshot" not in text + assert ".png" not in text + assert "Encountersaved" not in text + assert "http://" not in text + seal = client.get(f"/seals/{receipt['receipt_id']}?format=json").json() + assert "screenshot" not in json.dumps(seal) + + +def test_mismatching_qualification_is_refused(tmp_path: Path) -> None: + client, _store = _client(tmp_path) + response = client.post( + "/v1/executions", + json=_request(qualification_id="qualification_unknown1"), + ) + assert response.status_code == 422 + assert response.json()["error"] == "qualification_mismatch" + + +def test_unauthorized_v1_is_401(tmp_path: Path) -> None: + client, _store = _client(tmp_path) + bare = TestClient(client.app) + response = bare.post("/v1/executions", json=_request()) + assert response.status_code == 401 + + +def test_mcp_create_and_read(tmp_path: Path) -> None: + client, _store = _client(tmp_path) + listed = client.post( + "/mcp", + json={"jsonrpc": "2.0", "id": 1, "method": "tools/list", "params": {}}, + ).json() + names = {tool["name"] for tool in listed["result"]["tools"]} + assert names == { + "create_execution", + "get_execution", + "get_execution_receipt", + } + created = client.post( + "/mcp", + json={ + "jsonrpc": "2.0", + "id": 2, + "method": "tools/call", + "params": { + "name": "create_execution", + "arguments": _request(idempotency_key="caller_key_mcp_tool1"), + }, + }, + ).json() + accepted = json.loads(created["result"]["content"][0]["text"]) + execution_id = accepted["execution_id"] + receipt_rpc = client.post( + "/mcp", + json={ + "jsonrpc": "2.0", + "id": 3, + "method": "tools/call", + "params": { + "name": "get_execution_receipt", + "arguments": {"execution_id": execution_id}, + }, + }, + ).json() + receipt = json.loads(receipt_rpc["result"]["content"][0]["text"]) + assert receipt["outcome"] == "verified" + + +def test_source_boundary_has_no_cloud_tenant_or_billing_modules() -> None: + root = Path(__file__).resolve().parents[1] / "openadapt_flow" / "execute" + names = {path.name for path in root.glob("*.py")} + forbidden_files = { + "tenant.py", + "billing.py", + "stripe.py", + "control_plane.py", + "metering.py", + "orgs.py", + } + assert names.isdisjoint(forbidden_files) + joined = "\n".join(path.read_text(encoding="utf-8") for path in root.glob("*.py")) + assert "openadapt_cloud" not in joined + assert "openadapt-cloud" not in joined + assert "stripe" not in joined.lower() + assert "class Tenant" not in joined + assert "class Billing" not in joined + assert "app.openadapt.ai/seals" not in joined + assert "EXECUTE_LANE" not in joined + + +def test_synthetic_break_it_dispatch_is_zero_dollars() -> None: + request = ExecuteRequestV1.model_validate( + _request(environment_id=MOCKMED_ENVIRONMENT_LIE) + ) + from openadapt_flow.execute.models import AdmittedBundle + + admission = AdmittedBundle( + qualification_id=MOCKMED_QUALIFICATION_ID, + workflow_version=MOCKMED_WORKFLOW_VERSION, + workflow_digest=MOCKMED_WORKFLOW_DIGEST, + environment_id=MOCKMED_ENVIRONMENT_LIE, + minimum_effect_strength=MOCKMED_EFFECT_STRENGTH, + synthetic=True, + break_it=True, + ) + result = synthetic_mockmed(admission, request) + assert result.outcome is ExecuteTerminalOutcomeV1.RECONCILIATION_REQUIRED + assert isinstance(result, DispatchResult) + assert result.effect_passed is False diff --git a/uv.lock b/uv.lock index 7405b114..80b02f81 100644 --- a/uv.lock +++ b/uv.lock @@ -2181,6 +2181,11 @@ dev = [ { name = "scikit-image" }, { name = "uvicorn" }, ] +execute = [ + { name = "fastapi" }, + { name = "openadapt-types" }, + { name = "uvicorn" }, +] grounder = [ { name = "anthropic" }, ] @@ -2232,6 +2237,7 @@ requires-dist = [ { name = "cryptography", specifier = ">=42.0" }, { name = "fastapi", marker = "extra == 'console'", specifier = ">=0.110" }, { name = "fastapi", marker = "extra == 'dev'", specifier = ">=0.110" }, + { name = "fastapi", marker = "extra == 'execute'", specifier = ">=0.110" }, { name = "fastapi", marker = "extra == 'service'", specifier = ">=0.110" }, { name = "fastapi", marker = "extra == 'service-mlx'", specifier = ">=0.110" }, { name = "httpx", specifier = ">=0.27" }, @@ -2249,6 +2255,7 @@ requires-dist = [ { name = "openadapt-types", specifier = ">=0.13.0,<0.14.0" }, { name = "openadapt-types", marker = "extra == 'console'", specifier = ">=0.13.0,<0.14.0" }, { name = "openadapt-types", marker = "extra == 'dev'", specifier = ">=0.13.0,<0.14.0" }, + { name = "openadapt-types", marker = "extra == 'execute'", specifier = ">=0.13.0,<0.14.0" }, { name = "openadapt-types", marker = "extra == 'interop'", specifier = ">=0.13.0,<0.14.0" }, { name = "opencv-python", specifier = ">=4.9" }, { name = "pillow", specifier = ">=10.0" }, @@ -2273,10 +2280,11 @@ requires-dist = [ { name = "transformers", marker = "extra == 'service-mlx'", specifier = ">=5.5,<5.15" }, { name = "uvicorn", marker = "extra == 'console'", specifier = ">=0.29" }, { name = "uvicorn", marker = "extra == 'dev'", specifier = ">=0.29" }, + { name = "uvicorn", marker = "extra == 'execute'", specifier = ">=0.29" }, { name = "uvicorn", extras = ["standard"], marker = "extra == 'service'", specifier = ">=0.29" }, { name = "uvicorn", extras = ["standard"], marker = "extra == 'service-mlx'", specifier = ">=0.29" }, ] -provides-extras = ["browser", "dev", "grounder", "grounding", "console", "windows", "macos", "linux", "rdp", "privacy", "hosted", "service", "service-mlx", "capture", "interop"] +provides-extras = ["browser", "dev", "grounder", "grounding", "console", "execute", "windows", "macos", "linux", "rdp", "privacy", "hosted", "service", "service-mlx", "capture", "interop"] [[package]] name = "openadapt-grounding"