diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 27888fc..df583f1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -21,6 +21,14 @@ jobs: - name: Install run: pip install -e ".[dev]" - name: Lint - run: ruff check src tests + run: ruff check src tests scripts - name: Test run: pytest -q + - name: Build release artifacts + if: matrix.python-version == '3.12' + run: python -m build + - name: Verify MIT package boundary + if: matrix.python-version == '3.12' + run: | + python scripts/check_release_artifacts.py dist + python scripts/check_dist.py dist/* diff --git a/README.md b/README.md index d1e6fac..23f47c6 100644 --- a/README.md +++ b/README.md @@ -1,48 +1,32 @@ # OpenAdapt Agent -> [!IMPORTANT] -> **Status: Repurposed (v2, Experimental).** This package is now the -> **agent-facing bridge** for -> [`openadapt-flow`](https://github.com/OpenAdaptAI/openadapt-flow): it -> exposes compiled workflow bundles as **MCP servers** and **Agent -> Skills**, so other agents (Claude Code, Claude Desktop, any MCP client) -> can invoke governed workflows as tools. It is new, unproven code — do -> not treat it as production-ready. -> -> **Legacy v1 users:** the pre-pivot package (execution wrapper for -> model-driven GUI agents: safety gates, HITL confirmation, session -> management, audit logging) is deprecated and its modules were removed -> in v2. Pin the last v1-line release, -> [`openadapt-agent==0.1.0` on PyPI](https://pypi.org/project/openadapt-agent/0.1.0/), -> or browse the pre-pivot source at the last v1 commit on the `main` -> history. Execution responsibilities moved to the governed runtime in -> [`openadapt-flow`](https://github.com/OpenAdaptAI/openadapt-flow). - -[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT) +[![Lifecycle: Beta](https://img.shields.io/badge/lifecycle-Beta-2563eb)](https://github.com/OpenAdaptAI/openadapt-agent) +[![License: MIT](https://img.shields.io/badge/license-MIT-yellow.svg)](LICENSE) [![Python 3.10–3.12](https://img.shields.io/badge/python-3.10%E2%80%933.12-blue)](https://www.python.org/downloads/) -The bridge that makes the OpenAdapt README's promise — *"compiled -workflows can also be emitted as Agent Skills or MCP servers"* — real: +The local agent bridge for +[`openadapt-flow`](https://github.com/OpenAdaptAI/openadapt-flow). +It presents compiled, governed workflows to MCP clients and Agent Skills +without turning an agent into a second automation runtime. -- **`openadapt-agent serve`** — an MCP server (stdio) over a directory of - compiled workflow bundles. Each bundle becomes a `run_` tool with - a JSON schema derived from its declared parameters; every call shells - out to the governed `openadapt-flow run` CLI (fail-closed admission - gates), and returns success **or a structured halt/refusal with an - evidence pointer — never a fabricated success**. Read-only tools - (`list_workflows`, `get_workflow`, `get_run_report`) are always on; - run tools require the explicit `--allow-run` flag. -- **`openadapt-agent emit-skill`** — a Claude Agent Skill folder for one - bundle. Wraps flow's own `emit-skill` and appends MCP-invocation and - halt-semantics guidance for the consuming agent. +`openadapt-flow` remains the authority for execution, policy, identity, +verification, durable pauses, repair, and audit. This package adds two +agent-facing surfaces: -Design, security model, and what v2.0 deliberately does **not** do: -[`docs/DESIGN.md`](docs/DESIGN.md). +- `openadapt-agent serve` exposes compiled workflows and their local + **Needs Attention** queue over MCP stdio. +- `openadapt-agent emit-skill` emits a portable Agent Skill for a + compiled workflow, with the result and halt semantics an agent must + follow. + +Healthy workflow calls run through Flow's governed `run` command. A +halted or refused run is returned as a structured halt or refusal, never +as a fabricated success. ## Install ```bash -pip install "openadapt-agent>=2.0.0.dev0" # not yet published; from source: +# v2 is not yet published; install the current source: pip install git+https://github.com/OpenAdaptAI/openadapt-agent ``` @@ -54,80 +38,223 @@ uvx openadapt-agent serve --bundles /path/to/bundles # read-only uvx openadapt-agent serve --bundles /path/to/bundles --allow-run ``` -Requires Python 3.10–3.12 (inherited from `openadapt-flow`). Installing -pulls in `openadapt-flow` (the governed runtime this package drives) and -the official `mcp` SDK. +Python 3.10–3.12 is supported. The package installs a compatible +`openadapt-flow` runtime and the official MCP SDK. -## Quickstart: compile flow's bundled demo, serve it, call it +## Serve compiled workflows -```bash -# 1. Record + compile flow's canonical demo (MockMed, ships with flow) -openadapt-flow demo-record --out /tmp/rec -openadapt-flow compile /tmp/rec --out /tmp/bundles/triage --name "Demo Triage" +Compile a workflow with Flow, then serve its bundle directory: -# 2. Serve the bundle directory over MCP (stdio) -openadapt-agent serve --bundles /tmp/bundles --allow-run +```bash +openadapt-flow demo-record --out /tmp/recording +openadapt-flow compile /tmp/recording \ + --out /tmp/bundles/triage \ + --name "Demo Triage" + +openadapt-agent serve \ + --bundles /tmp/bundles \ + --runs-dir /tmp/openadapt-runs \ + --allow-run ``` -Register it with Claude Code: +For example, register it with an MCP client that accepts a local stdio +command: ```bash claude mcp add openadapt-workflows -- \ - openadapt-agent serve --bundles /tmp/bundles --allow-run + openadapt-agent serve \ + --bundles /tmp/bundles \ + --runs-dir /tmp/openadapt-runs \ + --allow-run ``` -Claude Code then sees `list_workflows`, `get_workflow`, -`get_run_report`, and `run_demo_triage` (with a `note` parameter -defaulting to the recorded value). Omit `--allow-run` to expose the -read-only tools only. - -> **Expect the demo run to be `refused` — that is the demo.** Run tools -> execute through flow's fail-closed `run` verb, and the freshly compiled -> demo bundle does not satisfy its admission gates (identity arming, -> effect contracts, encryption at rest), so `run_demo_triage` returns a -> structured `refused` outcome carrying the gate coverage report — -> nothing executes. That is the governed behavior this bridge exists to -> preserve. A bundle prepared for a real deployment (armed identity, -> effect contracts, `--config deployment.yaml` / `--policy`) runs to -> `success` or `halt`. To watch the demo actually execute without the -> gates, use flow's permissive demo verb directly: -> `openadapt-flow replay /tmp/bundles/triage`. - -Emit an Agent Skill instead of (or as well as) serving MCP: +The client receives `list_workflows`, `get_workflow`, +`get_run_report`, `list_needs_attention`, `get_attention_item`, and one +typed `run_workflow_` tool per loadable bundle. Declared +parameters are required; recorded demonstration values never appear in +the schema and are not silently reused. Omit `--allow-run` when the +client should be able to inspect workflows but not start them. + +By default every MCP response is safe to render outside the protected +workflow-data boundary. Workflow labels and intents, recorded values, +bundle/run paths, raw reports, observed text, stdout, stderr, and local +exception messages remain on the OpenAdapt machine. The client receives +opaque workflow/run IDs, fixed outcome copy, declared parameter +names/types, and count/boolean metrics. + +Two explicit development switches are deliberately separate: + +- `--allow-protected-export` sends raw local metadata and evidence to the + MCP client. Use it only when that client is trusted and inside the same + protected data boundary. +- `--allow-synthetic-recorded-defaults` lets omitted parameters reuse + recorded values. It requires `--allow-run` and is only for synthetic + demonstrations; production runs require every declared parameter to + prevent a wrong-record action. + +## Connect the attended workflow + +Add Flow's qualified deployment configuration to let a local operator +finish an exception and continue the same durable run: + +```bash +openadapt-agent serve \ + --bundles /opt/openadapt/bundles \ + --runs-dir /var/lib/openadapt/runs \ + --allow-run \ + --allow-attended-actions \ + --config /etc/openadapt/deployment.yaml \ + --headed +``` + +The queue summary is deliberately safe to show in an agent UI: it uses +opaque IDs, typed categories, counts, and signed-capability metadata +without returning observed text, workflow values, reports, or local +paths. Protected evidence stays in the local OpenAdapt operator +experience. + +Attended actions are separate, exact tools: + +| Tool | What happens | +| --- | --- | +| `continue_attention` | The operator confirms they completed the paused task in the live app. Flow revalidates its postconditions and independent effects, checkpoints it as human-completed, and resumes after it. It does not perform the completed action again. | +| `skip_attention` | Flow applies only an already-declared, non-consequential skip. A stale, undeclared, consequential, or ambiguous skip is refused. | +| `teach_attention` | Records an audited request for a corrective demonstration. Flow's existing revision and regression gates decide what can be promoted. | +| `escalate_attention` | Records an audited escalation and leaves the exact durable pause intact for a qualified operator. | + +Every action requires: + +- the opaque queue-item ID; +- the item's exact current capability digest; +- a caller-stable idempotency key; and +- an action-specific `true` confirmation in the tool request. + +Before submitting that request to Flow, the server also opens an MCP +form elicitation with action-specific language and requires the local +operator to accept and confirm it. This is a host-mediated explicit +confirmation signal, not cryptographic proof that a particular person +clicked it or proof of that person's identity. Flow separately records +the effective local OS account as the operator. Clients without form +elicitation cannot execute attended actions through this MCP bridge; the same +Continue, Skip, Teach, and Escalate capabilities remain available through +Flow's attended console/CLI. MCP destructive/idempotent/open-world +annotations give the host an additional approval signal. Neither those +hints nor the elicitation replaces Flow's signed capability, live +revalidation, idempotency, or audit contract. + +The direct fallback uses the same run directory and deployment boundary: ```bash -openadapt-agent emit-skill /tmp/bundles/triage --out ~/.claude/skills +openadapt-flow console \ + --attend \ + --allow-actions \ + --bundles /opt/openadapt/bundles \ + --runs /var/lib/openadapt/runs \ + --config /etc/openadapt/deployment.yaml \ + --headed ``` -## How results come back +Flow rechecks the signed capability, run identity, bundle version, +checkpoint lineage, authorization, live state, and effect evidence at +decision time. Stale capabilities and uncertain delivery are refused. +Retries with the same idempotency key return the prior terminal decision +instead of repeating it. + +With `--allow-attended-actions` but no deployment `--config`, the safe +Teach and Escalate transitions remain available; Continue and Skip are +not registered until Flow can construct the deployment-bound live +verifier and backend. + +## MCP tools + +| Tool | Registration | +| --- | --- | +| `list_workflows` | Always | +| `get_workflow` | Always | +| `get_run_report` | Always | +| `list_needs_attention` | Always | +| `get_attention_item` | Always | +| `run_workflow_` | `--allow-run` | +| `teach_attention`, `escalate_attention` | `--allow-attended-actions` | +| `continue_attention`, `skip_attention` | `--allow-attended-actions` plus a qualified deployment `--config` | + +## Run outcomes -Every `run_` call returns a structured outcome: +Every `run_workflow_` call returns one of these outcomes: -| `status` | meaning | +| `status` | Meaning | | --- | --- | -| `success` | exit 0 **and** the persisted `report.json` marks the run successful | -| `halt` | the run executed and stopped at an unhandled state — includes flow's structured halt observation (where, why, what was observed on screen, what completed first) and pointers to `report.json`/`REPORT.md`. **Not a success.** | -| `refused` | an `openadapt-flow run` admission gate refused the bundle (exit 2); **nothing was executed** | -| `timeout` | the run exceeded the operator-set deadline and was killed; the target may be partially executed | -| `error` | infrastructure problem (CLI missing, evidence inconsistent) | - -`get_run_report` returns the full `report.json` for any run this server -made, so the consuming agent can show the evidence rather than summarize -around it. - -## Honest limits (v2.0, Experimental) - -- **No auth story beyond local stdio.** The server trusts whoever spawned - it; the MCP client's user is the operator of record. No remote - transport, no TLS, no multi-tenancy — do not expose this across a - network boundary. -- **Unproven in production.** Unit tests mock the flow CLI; one local - end-to-end smoke exists. No pilot has run through this bridge. -- **No halt-resolution bridging.** Halts are surfaced with evidence; - fixing them (`openadapt-flow teach` / `approve` / `resume`) is still a - human CLI workflow. -- **Timeout ≠ rollback.** A killed run may leave the target mid-workflow; - the outcome says so, but nothing undoes partial work. +| `success` | The process exited successfully and the persisted report confirms the workflow completed and verified. | +| `halt` | Execution stopped instead of guessing. The workflow is not complete; protected evidence remains local. | +| `refused` | A governed admission gate refused the bundle before execution. Nothing ran. | +| `timeout` | The process exceeded its deadline. The target may be partially executed and must be inspected before retrying. | +| `error` | The CLI, report, or other execution infrastructure was inconsistent. | + +`get_run_report` returns a PHI-safe status and count-only summary for a +run created by this server. The persisted report remains in the local +operator experience unless protected export was explicitly enabled. A +client must never summarize `halt`, `refused`, `timeout`, or `error` as +success. + +## Emit an Agent Skill + +```bash +openadapt-agent emit-skill \ + /tmp/bundles/triage \ + --out ~/.claude/skills +``` + +This wraps Flow's own skill emitter, preserves its portable bundle, and +adds MCP invocation, Needs Attention, and result-handling guidance. +The emitted skill is not a sanitized derivative: it includes the +compiled bundle and Flow-generated workflow guidance. Treat the folder +as protected workflow data and install it only into an agent/client +authorized for that same data boundary. + +## Trust boundary + +This is intentionally a **local stdio bridge**. The process inherits the +local user's OS permissions, and that user is recorded as the operator +for attended decisions. Do not expose its stdin/stdout as an unauthenticated +network service. + +Remote transport, account identity, tenant isolation, fleet policy, and +managed execution belong to OpenAdapt Cloud. They are not duplicated in +this package. + +Other fixed boundaries: + +- Run tools and attended mutations are disabled unless explicitly + enabled when the server starts. +- Target, deployment policy, timeout, and model-egress posture are fixed + at server start rather than supplied by each MCP call. +- Parameters use a mode-`0600` temporary file, not process arguments. +- Workflow and run IDs are opaque in the default MCP surface. +- Recorded parameter values never enter tool schemas. Parameters are + required unless the server is explicitly put into synthetic-default + demo mode. +- Protected reports, names, values, paths, subprocess output, and + exception text stay local unless the operator explicitly enables + protected export for a trusted client in the same data boundary. +- Attended action schemas accept no free-text challenge answers or + protected evidence. +- Attended mutations require protocol-native form elicitation; a boolean + supplied by an autonomous tool caller is not sufficient on its own. + Elicitation is host-mediated confirmation, not cryptographic human-presence + or identity proof. + Clients without elicitation use Flow's existing attended console/CLI; + the capability is not converted to read-only. +- A timeout is not a rollback. Inspect the durable run before retrying. + +See [docs/DESIGN.md](docs/DESIGN.md) for the complete contract. + +## Package history + +The pre-v2 package was a wrapper for model-driven GUI agents. Its +execution responsibilities now live in `openadapt-flow`; the final +legacy release remains available as `openadapt-agent==0.1.0`. The +current package name is retained because this repository bridges both +MCP and Agent Skills; it is not an MCP-only package. ## Install as an MCP server (registries) @@ -141,7 +268,7 @@ distinction and the publish/submission plan. Machine-readable launch manifests live at the repo root: - [`server.json`](server.json) — official [MCP registry](https://github.com/modelcontextprotocol/registry) manifest (PyPI package, `uvx` runtime hint, stdio transport). -- [`smithery.yaml`](smithery.yaml) — [Smithery](https://smithery.ai) launch config (stdio `startCommand` with a `bundlesDir` / `allowRun` config schema). +- [`smithery.yaml`](smithery.yaml) — [Smithery](https://smithery.ai) launch config for bundle/run paths, governed execution, and attended-action opt-ins. - [`llms.txt`](llms.txt) — a concise, link-first summary for AI assistants. Registry-launched installs start **read-only by default**; execution @@ -161,9 +288,13 @@ Documentation: [docs.openadapt.ai](https://docs.openadapt.ai). ```bash pip install -e ".[dev]" -ruff check src tests && pytest +ruff check src tests scripts +pytest -q +python -m build +python scripts/check_release_artifacts.py dist +python scripts/check_dist.py dist/* ``` ## License -MIT — see [LICENSE](LICENSE). +MIT. See [LICENSE](LICENSE). diff --git a/docs/DESIGN.md b/docs/DESIGN.md index 5ff896d..6276e8c 100644 --- a/docs/DESIGN.md +++ b/docs/DESIGN.md @@ -1,179 +1,270 @@ -# openadapt-agent v2 — Design +# OpenAdapt Agent design -**Status: Experimental.** New, unproven code. Nothing in this document -claims production readiness; the honest-limits section below is part of -the design, not a disclaimer bolted on. +**Lifecycle: Beta.** -## What this package is +`openadapt-agent` is the local agent-facing bridge for compiled +`openadapt-flow` workflows. It exposes two complementary interfaces: -`openadapt-agent` v2 is the **agent-facing bridge** for -[openadapt-flow](https://github.com/OpenAdaptAI/openadapt-flow): it takes -a directory of *compiled workflow bundles* (the output of -`openadapt-flow compile` / `induce`) and exposes them to other agents — -Claude Code, Claude Desktop, or any MCP client — as: +1. MCP tools over local stdio. +2. Portable Agent Skills. -1. an **MCP server** (`openadapt-agent serve`, stdio transport), and -2. **Agent Skills** (`openadapt-agent emit-skill`). +It is not a second workflow engine. Flow owns compilation, policy, +identity, verification, durable execution, repair, and audit. -It makes real the main OpenAdapt README's promise that "compiled -workflows can also be emitted as Agent Skills or MCP servers." +## Architecture -v1 of this package (a legacy execution wrapper for model-driven GUI -agents: safety gates, HITL confirmation, session management) is -deprecated and its modules were removed in v2; the last v1-line release -remains on PyPI as `openadapt-agent 0.1.0`. +```text +MCP client / Agent Skill + │ + │ local stdio + exact JSON schemas + ▼ +openadapt_agent.mcp + │ + ▼ +openadapt_agent.bridge + ├── bundle discovery and typed run tools + ├── PHI-safe Needs Attention projection + ├── action-specific operator decisions + └── structured success / halt / refusal results + │ + ├── new run ──────────────► openadapt-flow run subprocess + │ fail-closed admission + execution + │ + └── attended decision ────► openadapt-flow durable API + signed capability + idempotency + + live revalidation + audit +``` -## Relationship to openadapt-flow's own emitters (wrap, don't duplicate) +The MCP adapter is intentionally thin. Tool descriptions and dispatch +live in a transport-independent bridge so they can be tested without +starting stdio. -flow ≥1.9 already ships `openadapt_flow.emit` with two emitters and CLI -verbs: +## Why the package keeps both MCP and Agent Skills -| flow surface | what it produces | gap for agent use | -| --- | --- | --- | -| `emit-skill` (`emit_skill`) | portable skill folder: `SKILL.md` + copied `bundle/`, CLI `replay` invocation | no MCP guidance; invokes the *permissive* `replay` path; no halt-semantics instructions for the consuming agent | -| `emit-mcp` (`emit_mcp_server`) | standalone per-bundle FastMCP `server.py` that runs the `Replayer` **in-process** | single workflow per server; permissive replay path (no `run` admission gates); web/Playwright backend only; always executable (no read-only mode); result carries only a success flag — no structured halt evidence | +Flow's skill emitter produces the portable workflow folder. This package +wraps it and appends agent-facing invocation, halt, and attended-action +guidance without regenerating the workflow. The resulting skill includes +the compiled bundle and is protected workflow data, not a sanitized +artifact; operators install it only into an agent authorized for the +same data boundary. -This package **wraps** rather than duplicates: +Flow's per-bundle MCP emitter has a different purpose: a self-contained +demo server for one workflow. `openadapt-agent serve` is the governed +multi-bundle local surface. It uses Flow's `run` admission path and +returns durable evidence rather than invoking permissive replay. -- `openadapt-agent emit-skill` calls flow's `emit_skill()` verbatim and - then *appends* two sections to the generated `SKILL.md`: how to invoke - the workflow through the MCP server, and the halt semantics the - consuming agent must respect. flow's frontmatter, step list, parameter - table, and CLI usage are untouched. -- `openadapt-agent serve` does **not** build on `emit_mcp_server`, - deliberately: that emitter's design point is a self-contained, - in-process, single-workflow demo server. The bridge's design point is - the opposite — multi-workflow, out-of-process, **governed**. The two - coexist; the DESIGN gap table above is the justification. If flow later - grows a governed multi-bundle server, this package should shrink to a - wrapper around it. +The repository therefore remains `openadapt-agent`, not +`openadapt-mcp`: MCP is one transport; Agent Skills are an equally +supported surface. -## Architecture +## Tool registration -``` -MCP client (Claude Code / Desktop, stdio) - │ tools/list, tools/call - ▼ -openadapt_agent.mcp thin adapter: bridge ⇄ mcp SDK low-level Server - ▼ -openadapt_agent.bridge tool specs + dispatch (no mcp imports; unit-testable) - ├─ openadapt_agent.bundles discovery + metadata via Workflow.load (flow IR) - └─ openadapt_agent.runner one subprocess per run: - openadapt-flow run BUNDLE --run-dir … --params-file … [--url/--config/--policy …] - └── flow's fail-closed admission gates + replayer -``` +The server always registers: + +- `list_workflows` +- `get_workflow` +- `get_run_report` +- `list_needs_attention` +- `get_attention_item` + +The first three return PHI-safe projections: opaque IDs, parameter +names/types, availability, status, and count/boolean metrics. They do not +return workflow labels, recorded values, step intents, report bodies, +observed text, local paths, subprocess output, or exception text. The +attention tools use the same boundary plus typed categories, artifact +IDs, and non-authorizing capability metadata. + +`--allow-run` registers one `run_workflow_` tool per loadable +workflow. Every declared parameter is required by default, and recorded +demonstration values never enter the schema. Missing or unknown +parameters are rejected before the subprocess starts. A per-call URL is +accepted only if the operator separately enabled `--allow-url-override`. + +Two server-start modes are intentionally separate from ordinary +operation: + +- `--allow-protected-export` includes raw labels, values, intents, local + paths, reports, stdout/stderr, and detailed local errors. It is for an + explicitly trusted MCP client inside the same protected data boundary. +- `--allow-synthetic-recorded-defaults` permits omitted parameters to + use demonstrated values. It requires run authority and is only for + synthetic demos; it never places those values in a tool schema. + +`--allow-attended-actions` registers Teach and Escalate. Continue and +Skip are registered only when a deployment configuration lets Flow +construct its bound live executor. + +## Governed runs + +Each run tool shells out to the `openadapt-flow` installed in the same +Python environment. The command uses Flow's fail-closed `run` verb, so +its certification, identity, effects, encryption, integrity, and egress +gates remain authoritative. + +The bridge reports success only when both conditions hold: + +1. Flow exits with code 0. +2. The persisted report has `success: true`. + +Exit 1 is a halt. Exit 2 is a governed refusal before execution. A +timeout is explicitly uncertain rather than a rollback. Report evidence +always outranks a process exit code. + +Parameters are passed in a mode-`0600` temporary JSON file and removed +after the run. Server-fixed target, policy, deployment configuration, +timeout, and extra Flow arguments cannot be changed by an MCP call. + +The runner retains detailed reports and subprocess diagnostics locally, +but its default MCP projection contains only an opaque workflow/run ID, +fixed status copy, and schema-limited metrics. Success, halt, refusal, +timeout, and infrastructure-error paths share this same projection, so +an exception cannot accidentally turn into an egress channel. + +## Needs Attention + +Flow creates a signed attended capability for a specific durable pause. +The capability binds: + +- run and pause identity; +- workflow and bundle version; +- exact step or interpreter cursor; +- checkpoint lineage and expected next transition; +- verification and delivery state; +- allowed actions and expiration. + +Only the capability digest and allowed-action summary cross the +agent-facing queue projection. The HMAC, protected evidence, and local +paths stay inside the Flow runtime boundary. + +Each mutation is an action-specific MCP tool with +`additionalProperties: false`. Its payload contains only: + +- opaque attention ID; +- exact capability digest; +- stable idempotency key; +- one explicit boolean operator confirmation. + +There is no field for challenge answers, credentials, screenshots, +observed text, or arbitrary approval prose. + +The payload's confirmation boolean is not treated as proof of human +presence. Before dispatch, the MCP server performs a second, +action-specific form elicitation and requires an explicit accept plus +confirmation from the local operator. Elicitation is a host-mediated +explicit-confirmation signal, not cryptographic proof that a particular +person clicked, nor identity proof. Flow separately records the effective +local OS account as the operator. A client that does not advertise +form elicitation cannot execute attended mutations through MCP; the +operator uses Flow's existing attended console/CLI instead, where all +four capabilities remain available. This is a transport authorization +choice, not a read-only conversion. Tool annotations also mark Continue +and Skip as destructive, idempotent, and open-world so the host can apply +its own approval policy. Annotations and elicitation do not replace +Flow's signed capability, live revalidation, idempotency, or durable +audit. + +### Continue + +Continue means the human already completed the paused task in the live +application. Flow: + +1. reloads and validates the exact signed pause under a filesystem + lease; +2. verifies the human-completed postconditions and independent effects + in the deployment-bound live session; +3. commits a human-completed checkpoint; +4. resumes from the next transition. + +The completed action is never actuated again. If delivery may have +crossed the boundary without a terminal receipt, retries are refused +until reconciliation. + +### Skip + +Skip is not a generic bypass. It exists only when the signed capability +and compiled workflow declare a safe, non-consequential skip. Flow +rechecks that guard against current state. Consequential, stale, +ambiguous, or undeclared skips are refused. + +### Teach + +Teach records an audited request for a corrective demonstration. The +durable pause remains intact. The existing Flow teach/revision pipeline +owns capture, regression evaluation, evidence banking, and promotion; +the MCP client cannot directly rewrite a bundle. + +### Escalate + +Escalate records a durable, audited request for qualified assistance and +leaves the pause available for later resolution. + +## Idempotency and thread ownership + +Flow persists attended decisions before crossing a delivery boundary. +Repeating the same action with the same idempotency key returns its +terminal decision. Reusing a key for different content is refused. + +Continue and Skip use a persistent deployment-bound backend. Some +backends, including Playwright, are thread-affine and their synchronous +APIs cannot run inside MCP's asyncio event loop. Flow's public +`AttendedActionService` therefore creates, uses, and closes the live +executor on one dedicated non-async owner thread. The bridge submits +exact signed requests through that service while the event loop remains +responsive. Run subprocesses and read-only projections use ordinary +worker threads. + +Flow serializes live attended actions and applies per-pause filesystem +leases. A second process cannot silently duplicate an in-flight +decision. + +## Identity and transport + +The MCP server uses local stdio. The process inherits the OS user's +permissions, and the effective local OS account is recorded as the +attended operator. POSIX uses the effective UID account; Windows uses +the process/thread token-backed `GetUserNameW` API rather than the +caller-controlled `USERNAME` environment variable. A blank operator +identity fails closed. + +This process must not be port-forwarded or exposed as an unauthenticated +network service. OpenAdapt Cloud owns remote authentication, +multi-tenancy, tenant-scoped authorization, fleet policy, and managed +transport. + +## Dependency boundary + +The attended bridge uses Flow's public durable action contract and public +`AttendedActionService`, constructed from a public `DeploymentConfig`. +The package pins the Flow minor release containing that contract so a +later refactor cannot silently change backend construction, thread +ownership, or cleanup semantics. + +Agent applies only explicit server-start URL, visibility, and egress +overrides to that typed deployment config. Replayer, backend, policy, +verification, owner-thread, resume, and audit logic are never copied +into this repository. + +## Test and release contract + +Tests cover: + +- exact tool registration and schemas; +- opaque workflow/run IDs, required parameters, and no recorded defaults + in MCP schemas; +- adversarial PHI/secret/path/exception strings across workflow + discovery, success, halt, refusal, timeout, error, and report lookup; +- explicit protected-export and synthetic-default modes; +- MCP form elicitation and action annotations; +- PHI-safe queue projections and path traversal refusal; +- stale capability, unknown field, and false-confirmation refusal; +- idempotent Continue without re-actuation; +- Teach and Escalate without a live service; +- delegation to Flow's public service context; +- compatibility with Flow's public, thread-owned attended service; +- success/halt/refusal/timeout outcome mapping; +- MCP serialization and thread ownership; +- Agent Skill emission. -Execution is **never reimplemented**. Every run tool call is a subprocess -invocation of `openadapt-flow run` — flow's fail-closed deployment verb — -so certification, identity arming, effect contracts, encryption-at-rest, -and integrity pinning apply exactly as they would from a terminal. The -bridge has no code path that could bypass them. - -### Tools - -Read-only (always registered): - -- `list_workflows` — discovered bundles (slug, name, params, step count, - load errors), plus whether run tools are enabled. -- `get_workflow` — step intents, declared params with recorded example - values, and certification status (evaluated via `openadapt-flow - certify` when the operator configured `--policy`/`--config`; honestly - `null` otherwise). -- `get_run_report` — the persisted `report.json` for a `run_id` this - server produced. Constrained to the server's runs directory (single - path component, resolved containment check). - -Run tools (registered only under `--allow-run`): - -- `run_` per loadable bundle. Input schema is derived from the - bundle's declared parameters: each is an optional string whose default - is the recorded example value (the same fallback the flow CLI applies); - `additionalProperties: false`. A `url` property exists only when the - operator passed `--allow-url-override`. - -### Outcome mapping (honesty invariant) - -`openadapt-flow run` exit-code contract → MCP result `status`: - -| exit | meaning | status | -| --- | --- | --- | -| 0 + report.success | executed, every step verified | `success` | -| 0, no/failed report | inconsistent evidence | `error` / `halt` (report outranks exit code; never success) | -| 1 | executed and stopped | `halt` — includes the structured `halt` observation (state, reason, PHI-scrubbed observed texts, completed intents) or the failing step, plus `run_dir`/`report_path` evidence pointers | -| 2 | governed refusal — nothing executed | `refused` — stdout tail carries the gate coverage report | -| timeout | process killed at operator-set deadline | `timeout` — with a warning that the target may be partially executed | - -The invariant, enforced in `classify_outcome` and covered by tests: -**`success` requires both exit 0 and a persisted report with -`success: true`.** A halt is surfaced as a halt with an evidence pointer, -never as success, and never silently retried. - -## Security model - -- **Execution is opt-in.** Without `--allow-run` the server registers - read-only tools only; a forged `run_*` call is refused by dispatch as - well (belt and braces). -- **The operator fixes the blast radius at start time.** Target URL, - deployment config, policy, timeout, extra run args, and the flow CLI - itself are server flags — an MCP client cannot alter them per call. - Per-call URL override requires the separate `--allow-url-override`. -- **Governance lives in flow, not here.** The server cannot bypass - policy/identity/effect gates because it only ever invokes the governed - CLI; there is no in-process replay path. -- **Parameter hygiene.** Params travel via a mode-0600 temp file passed - as `--params-file` (flow's managed-runner mechanism), never argv, and - the file is deleted after the run. Unknown parameters are rejected - before any subprocess starts. -- **Report access is scoped.** `get_run_report` accepts a single path - component and verifies the resolved path stays inside the server's runs - directory. -- **Operator of record.** The server performs no authentication. The MCP - client's user is the operator of record for anything a run writes; - stdio transport means the server runs with that user's local - privileges, on that user's machine. -- **Per-call timeout.** Every run subprocess is killed at the configured - deadline and reported as `timeout` (not success). Note flow may leave - the target mid-workflow; the outcome says so. -- **Encrypted bundles** stay encrypted: metadata loads through - `Workflow.load` (key from `OPENADAPT_BUNDLE_KEY`), and a bundle that - cannot be loaded is listed with its `load_error` and gets no run tool. - -## What v2.0 does NOT do - -- **No remote transport.** stdio only. No SSE/HTTP, no TLS story, no - network authentication or multi-tenant isolation. Do not port-forward - this at a network boundary. -- **No auth beyond the local user.** Anyone who can spawn the process can - call the tools it registers. -- **No scheduling, queueing, or concurrency control.** One subprocess per - tool call; concurrent calls are concurrent flow runs. -- **No execution semantics of its own.** No retries, no healing, no - branching — all of that is flow's job. -- **No teach/resume/approve bridging yet.** Halts are surfaced with - evidence; resolving them (`openadapt-flow teach`, `approve`, `resume`) - remains a human CLI workflow. -- **No production track record.** This code is new and unproven; it has - unit tests with the flow CLI mocked and a local end-to-end smoke, and - nothing more. - -## Testing strategy - -- Unit tests mock `subprocess.run` with a stub that emulates the flow - CLI's exit-code/report contract; fixture bundles are written through - flow's own IR (`Workflow.save`) so schema drift breaks tests here - instead of hiding. -- Covered: schema generation from a fixture bundle, run→success mapping, - run→halt mapping (exit 1 and exit-0-with-failed-report both), exit 2 → - structured refusal, timeout handling, `--allow-run` and - `--allow-url-override` gating, run-report path containment, skill - emission golden-file. -- The MCP layer is exercised by constructing the low-level server and - invoking its `tools/list` handler; the stdio transport is covered by a - manual smoke script (`scripts/smoke_client.py`), not CI. -- What the real-CLI smoke has actually proven so far: flow's bundled demo - compiled → served → `tools/list` + `list_workflows` + `get_workflow` - over real stdio, and `run_demo_triage` → structured `refused` carrying - the real gate coverage report (the demo bundle cannot pass the - fail-closed gates, by design). The `success`/`halt` mappings are proven - only against the stubbed CLI contract, not yet against a - gate-satisfying real deployment bundle. +CI runs on Python 3.10, 3.11, and 3.12. It also builds the wheel and +sdist, verifies MIT metadata and license inclusion, and refuses package +artifacts containing repository-only copyleft benchmark material. diff --git a/docs/DISTRIBUTION.md b/docs/DISTRIBUTION.md index af5e328..647c699 100644 --- a/docs/DISTRIBUTION.md +++ b/docs/DISTRIBUTION.md @@ -1,6 +1,6 @@ # Distribution & discoverability — openadapt-agent -**Status: Experimental (v2).** This document describes how the package is +**Status: Beta (v2).** This document describes how the package is made installable and discoverable as an MCP server, the security-relevant distinction between the *public capability* and a *user's private bundle*, and the exact founder-owned steps to publish and list it. The @@ -15,7 +15,7 @@ server". Keep them separate. | | Public / official | Private / per-user | | --- | --- | --- | | **What it is** | the `openadapt-agent` package and its MCP server *program* | a user's compiled `openadapt-flow` workflow *bundle* | -| **What it exposes** | the OpenAdapt capability: inspect a compiled bundle, and (opt-in) run it under governance | one customer's specific recorded workflow (its steps, parameters, recorded example values) | +| **What it exposes** | PHI-safe workflow and Needs Attention projections plus opt-in governed run and attended-action tools | one customer's specific recorded workflow (its steps, parameters, recorded example values) | | **Where it lives** | PyPI + MCP registries (official, Smithery, mcp.so, Glama, PulseMCP) | the operator's own disk, passed at launch via `--bundles` | | **Ships in the package?** | yes — code only | **never** — no bundle is embedded in the wheel, `server.json`, or any registry listing | @@ -29,12 +29,15 @@ inside their trust boundary. The registry listing advertises the **Read-only by default when registry-launched.** The `server.json` and `smithery.yaml` launch the server WITHOUT `--allow-run`, so a one-click -install from a directory yields the inspection tools only -(`list_workflows`, `get_workflow`, `get_run_report`). Executing a -workflow (`run_`) is a deliberate operator act: they add -`--allow-run` (and typically `--policy`/`--config`) when they run the -server against their own bundles. This matches the security model in -[`DESIGN.md`](DESIGN.md). +install from a directory yields PHI-safe inspection and Needs Attention +tools only (`list_workflows`, `get_workflow`, `get_run_report`, +`list_needs_attention`, and `get_attention_item`). Executing a workflow +(`run_workflow_`) is a deliberate operator act: they add +`--allow-run` (and typically `--policy`/`--config`). Teach and Escalate +require `--allow-attended-actions`; Continue and Skip additionally require +a qualified deployment `--config`. Clients without MCP form elicitation use +Flow's attended console/CLI, where all four capabilities remain available. +This matches the security model in [`DESIGN.md`](DESIGN.md). > There is intentionally **no** hosted, multi-tenant "official OpenAdapt > workflow server" that exposes OpenAdapt-operated workflows to the @@ -52,19 +55,20 @@ and [`../smithery.yaml`](../smithery.yaml). - **Name (reverse-DNS, official registry):** `io.github.OpenAdaptAI/openadapt-agent` - **Display name:** OpenAdapt Agent (openadapt-flow bridge) - **PyPI package:** `openadapt-agent` -- **Version:** `2.0.0.dev0` (Experimental; pick the final release version at publish — see step 3.1) -- **Description:** Expose compiled openadapt-flow workflow bundles as governed MCP tools (Experimental). +- **Version:** `2.0.0.dev0` (Beta development release; pick the final release version at publish — see step 3.1) +- **Description:** Local Beta bridge for governed openadapt-flow workflows and attended actions. - **Homepage / docs:** https://docs.openadapt.ai - **Repository:** https://github.com/OpenAdaptAI/openadapt-agent - **License:** MIT - **Transport:** stdio -- **Run command (uvx):** `uvx openadapt-agent serve --bundles [--allow-run]` -- **Config:** `--bundles` (required, dir), `--allow-run` (opt-in execution), `OPENADAPT_BUNDLE_KEY` (optional secret, for encrypted bundles) +- **Run command (uvx):** `uvx openadapt-agent serve --bundles [--allow-run] [--allow-attended-actions]` +- **Config:** `--bundles` (required), `--runs-dir`, `--allow-run`, `--allow-attended-actions`, qualified `--config` for Continue/Skip, and optional secret `OPENADAPT_BUNDLE_KEY` - **Tools:** - - `list_workflows` — list exposed bundles (slug, name, params, step count, load errors). - - `get_workflow` — inspect one workflow (step intents, declared params with recorded example values, certification status). - - `get_run_report` — fetch the persisted `report.json` for a prior run (halt/success evidence). - - `run_` — execute a bundle via the governed `openadapt-flow run` CLI (only when `--allow-run`). Returns `success` | `halt` | `refused` | `timeout` | `error`. + - `list_workflows` / `get_workflow` — PHI-safe structural bundle projections with opaque IDs. + - `get_run_report` — PHI-safe status and count summary; raw evidence stays local unless protected export was explicitly enabled. + - `list_needs_attention` / `get_attention_item` — PHI-safe durable-pause cards and current signed-capability metadata. + - `run_workflow_` — execute through the governed `openadapt-flow run` CLI when `--allow-run`; returns `success` | `halt` | `refused` | `timeout` | `error`. + - `continue_attention` / `skip_attention` / `teach_attention` / `escalate_attention` — exact, elicited attended decisions under Flow's capability, idempotency, verification, and audit contract. - **Categories/tags:** mcp, agent-skills, automation, workflow, gui, governed, healthcare, rpa ## 3. Release automation vs. founder one-time actions @@ -138,7 +142,7 @@ No secret is ever committed; these live only in repo Settings -> Secrets. `2.0.0.dev0` is a dev pre-release; `pip install openadapt-agent` will not select it without `--pre`. For a real launch choose one of: - `2.0.0rc1` (recommended first public: pre-release, opt-in via `--pre`), or -- `2.0.0` (final; drop the Experimental-only caveats only when honest to). +- `2.0.0` (final; remove development-release caveats only when the evidence supports it). Whatever you choose, set it in **three places that a CI test pins together**: `pyproject.toml` `version`, `src/openadapt_agent/__init__.py` @@ -182,7 +186,9 @@ add the ownership marker it names to the package metadata and re-run. - "Add Server" -> point at `github.com/OpenAdaptAI/openadapt-agent`. - Smithery reads `smithery.yaml` (stdio `startCommand`, `configSchema`, `commandFunction`). Confirm the generated config form shows - `bundlesDir` (required), `allowRun` (default off), `bundleKey` (secret). + `bundlesDir` (required), `runsDir`, `allowRun` (default off), + `allowAttendedActions` (default off), optional `deploymentConfig`, + `headed`, and `bundleKey` (secret). - Publish. Smithery hosts the connection config; it does not host bundles. > **Why manual:** Smithery has no public "create listing" API keyed to an diff --git a/llms.txt b/llms.txt index 9effad0..ef7f02f 100644 --- a/llms.txt +++ b/llms.txt @@ -1,22 +1,32 @@ # openadapt-agent -> The agent-facing bridge for [openadapt-flow](https://github.com/OpenAdaptAI/openadapt-flow). It exposes compiled workflow bundles to other agents (Claude Code, Claude Desktop, any MCP client) as an **MCP server** and as **Agent Skills**. Every execution shells out to the governed `openadapt-flow run` CLI, so flow's fail-closed admission gates (certification, identity arming, effect contracts, encryption at rest, integrity pinning) cannot be bypassed. A halted or refused run is always returned as a structured halt/refusal with an evidence pointer, never as a fabricated success. Status: Experimental (v2). +> The local agent bridge for [openadapt-flow](https://github.com/OpenAdaptAI/openadapt-flow). It exposes compiled workflows and PHI-safe Needs Attention items to MCP clients and emits Agent Skills without becoming a second automation runtime. Healthy execution uses Flow's governed `run` command; a halted or refused run is returned as that exact non-success outcome. Flow remains the authority for policy, identity, verification, durable pauses, attended decisions, repair, and audit. Status: Beta (v2). ## What it provides -- `openadapt-agent serve --bundles [--allow-run]`: an MCP server (stdio) over a directory of compiled bundles. Tools: `list_workflows`, `get_workflow`, `get_run_report` (always on, read-only) and `run_` (only when the operator passes `--allow-run`), whose JSON input schema is derived from the bundle's declared parameters. -- `openadapt-agent emit-skill --out `: a Claude Agent Skill folder for one bundle. Wraps flow's own `emit-skill` and appends MCP-invocation and halt-semantics guidance. +- `openadapt-agent serve --bundles [--allow-run]`: a local MCP stdio server. `list_workflows`, `get_workflow`, `get_run_report`, `list_needs_attention`, and `get_attention_item` are always available as PHI-safe read-only projections. `run_workflow_` tools require `--allow-run`. +- `--allow-attended-actions` adds exact Teach and Escalate tools for signed durable pauses. With a qualified Flow `--config`, the same server also exposes Continue and Skip through Flow's deployment-bound live verifier and deterministic resume path. +- `openadapt-agent emit-skill --out ` wraps Flow's skill emitter and appends MCP, halt, and attended-action guidance. + +## Attended actions + +- `continue_attention`: after the local operator completes the paused task, Flow verifies the exact outcome and resumes without actuating that task again. +- `skip_attention`: applies only a compiled, non-consequential skip that Flow revalidates at decision time. +- `teach_attention`: records an audited corrective-demonstration request; Flow's revision and regression gates still decide promotion. +- `escalate_attention`: records escalation and preserves the exact durable pause. + +All mutations require an exact capability digest, a stable idempotency key, action-specific confirmation, and MCP form elicitation. Clients without form elicitation use Flow's attended console/CLI; the capabilities are not removed or converted to read-only. ## Run outcomes (honesty invariant) - `success`: exit 0 AND the persisted `report.json` marks the run successful. Only then did the workflow complete. -- `halt`: the run executed and stopped at an unhandled state (halt instead of guessing). NOT a success; evidence in `report.json`/`REPORT.md`. +- `halt`: execution stopped instead of guessing. It is not a success; protected evidence remains in the local operator experience. - `refused`: a governed admission gate refused the bundle before execution; nothing ran. - `timeout` / `error`: killed at the deadline / infrastructure problem. ## Public capability vs private artifact -- The **package/MCP server is the public, official OpenAdapt capability**: inspect and (opt-in) run a compiled bundle under governance. It is what gets published to PyPI and listed in MCP registries. +- The **package/MCP server is the public OpenAdapt capability**: inspect and, with explicit operator flags, run or attend a compiled workflow under governance. The v2 package is not yet published. - A user's **compiled workflow bundle is their private artifact**. It is supplied at server start via `--bundles` and is never embedded in the package, the `server.json`, or any registry listing. ## Key documents diff --git a/pyproject.toml b/pyproject.toml index 02ba633..6184479 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "hatchling.build" [project] name = "openadapt-agent" version = "2.0.0.dev0" -description = "Agent-facing bridge for openadapt-flow: expose compiled workflow bundles as MCP servers and Agent Skills (Experimental)" +description = "Local agent bridge for governed OpenAdapt workflows, attended actions, MCP, and Agent Skills" readme = "README.md" # openadapt-flow pins <3.13 (rapidocr-onnxruntime); this package inherits it. requires-python = ">=3.10,<3.13" @@ -17,7 +17,7 @@ maintainers = [ {name = "OpenAdaptAI", email = "contact@openadapt.ai"} ] classifiers = [ - "Development Status :: 3 - Alpha", + "Development Status :: 4 - Beta", "Intended Audience :: Developers", "License :: OSI Approved :: MIT License", "Programming Language :: Python :: 3", @@ -32,9 +32,9 @@ keywords = ["mcp", "agent-skills", "gui", "automation", "workflow", "openadapt", dependencies = [ # Governed workflow compiler/runtime this package bridges. Execution # shells out to its CLI; bundle metadata loads via its IR. - "openadapt-flow>=1.9,<2", + "openadapt-flow>=1.18.1,<2", # Official Model Context Protocol SDK (stdio server transport). - "mcp>=1.2", + "mcp>=1.28,<2", # Structured concurrency runtime the mcp SDK already uses; we call # anyio.run / to_thread directly. "anyio>=4.0", @@ -42,6 +42,7 @@ dependencies = [ [project.optional-dependencies] dev = [ + "build>=1.2.0", "pytest>=8.0.0", "ruff>=0.4.0", # Parses smithery.yaml in the distribution-artifact guard tests. diff --git a/scripts/check_dist.py b/scripts/check_dist.py new file mode 100644 index 0000000..7c1b3c3 --- /dev/null +++ b/scripts/check_dist.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +"""Fail a release if its archives violate the OpenAdapt package boundary.""" + +from __future__ import annotations + +import argparse +import tarfile +import zipfile +from pathlib import Path + +_FORBIDDEN_PATH_MARKERS = ( + "openimis", + "benchmark/openimis", + "benchmarks/openimis", + "adversary_corpus", + "identity_roc", + "pixel_verify_cert", + "oracle_recipe", + "real_emr", + "held_out_corpus", + "/private/", +) +_COPYLEFT_MARKERS = ( + b"GNU " + b"AFFERO GENERAL PUBLIC LICENSE", + b"GNU " + b"GENERAL PUBLIC LICENSE", + b"SPDX-License-Identifier: " + b"AGPL-", + b"SPDX-License-Identifier: " + b"GPL-", +) + + +class ArtifactError(RuntimeError): + """Distribution archive failed the release boundary.""" + + +def _members(path: Path) -> dict[str, bytes]: + if path.suffix == ".whl": + with zipfile.ZipFile(path) as archive: + return { + info.filename: archive.read(info) + for info in archive.infolist() + if not info.is_dir() + } + if path.name.endswith(".tar.gz"): + with tarfile.open(path, "r:gz") as archive: + return { + member.name: extracted.read() + for member in archive.getmembers() + if member.isfile() and (extracted := archive.extractfile(member)) is not None + } + raise ArtifactError(f"unsupported artifact type: {path}") + + +def check(path: Path) -> None: + members = _members(path) + if not members: + raise ArtifactError(f"{path}: empty archive") + + lowered_names = {name: name.lower() for name in members} + forbidden = sorted( + name + for name, lowered in lowered_names.items() + if any(marker in lowered for marker in _FORBIDDEN_PATH_MARKERS) + ) + if forbidden: + raise ArtifactError(f"{path}: repository-only benchmark material in artifact: {forbidden}") + + metadata = [ + body + for name, body in members.items() + if name.endswith(".dist-info/METADATA") or name.endswith("/PKG-INFO") + ] + if not metadata or not any( + b"License-Expression: MIT" in body or b"License: MIT" in body for body in metadata + ): + raise ArtifactError(f"{path}: package metadata does not declare MIT") + + license_files = [ + body + for name, body in members.items() + if Path(name).name.upper() in {"LICENSE", "LICENSE.TXT", "LICENSE.MD"} + ] + if not license_files or not any( + b"Permission is hereby granted, free of charge" in body for body in license_files + ): + raise ArtifactError(f"{path}: MIT LICENSE is missing") + + copied_copyleft = [] + for name, body in members.items(): + if name.endswith((".dist-info/METADATA", "/PKG-INFO")): + continue + body_upper = body.upper() + if any(marker.upper() in body_upper for marker in _COPYLEFT_MARKERS): + copied_copyleft.append(name) + copied_copyleft.sort() + if copied_copyleft: + raise ArtifactError(f"{path}: copied copyleft material in MIT artifact: {copied_copyleft}") + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("artifacts", nargs="+", type=Path) + args = parser.parse_args() + for artifact in args.artifacts: + check(artifact) + print(f"PASS {artifact}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/smoke_client.py b/scripts/smoke_client.py index 7f5ffd9..c058c32 100644 --- a/scripts/smoke_client.py +++ b/scripts/smoke_client.py @@ -4,9 +4,9 @@ python scripts/smoke_client.py --bundles /path/to/bundles [--allow-run] -Performs tools/list plus the read-only ``list_workflows`` and -``get_workflow`` calls over the stdio transport and prints the results. -It never invokes a run tool — executing a workflow is an operator action. +Performs tools/list plus ``list_workflows``, ``get_workflow``, and the +PHI-safe ``list_needs_attention`` call over stdio. It never invokes a +run or attended-action tool. """ from __future__ import annotations @@ -50,11 +50,18 @@ async def main() -> int: print("\nlist_workflows:") print(json.dumps(payload, indent=2)) - workflows = [w for w in payload["workflows"] if not w["load_error"]] + attention = await session.call_tool("list_needs_attention", {}) + print("\nlist_needs_attention:") + print(attention.content[0].text) + + workflows = [w for w in payload["workflows"] if w["available"]] if workflows: - slug = workflows[0]["slug"] - detail = await session.call_tool("get_workflow", {"workflow": slug}) - print(f"\nget_workflow({slug!r}):") + workflow_id = workflows[0]["id"] + detail = await session.call_tool( + "get_workflow", + {"workflow": workflow_id}, + ) + print(f"\nget_workflow({workflow_id!r}):") print(detail.content[0].text) return 0 diff --git a/server.json b/server.json index 6eac93a..c64b33f 100644 --- a/server.json +++ b/server.json @@ -2,7 +2,7 @@ "$schema": "https://static.modelcontextprotocol.io/schemas/2025-12-11/server.schema.json", "name": "io.github.OpenAdaptAI/openadapt-agent", "title": "OpenAdapt Agent (openadapt-flow bridge)", - "description": "Expose compiled openadapt-flow workflow bundles as governed MCP tools (Experimental).", + "description": "Local Beta bridge for governed openadapt-flow workflows and attended actions.", "websiteUrl": "https://docs.openadapt.ai", "repository": { "url": "https://github.com/OpenAdaptAI/openadapt-agent", @@ -47,7 +47,7 @@ ], "_meta": { "io.modelcontextprotocol.registry/publisher-provided": { - "notes": "Read-only by default. The server registers execution (run_*) tools only when the operator adds the --allow-run flag; a registry-launched install stays inspection-only unless that flag is appended. See docs/DISTRIBUTION.md." + "notes": "PHI-safe inspection and Needs Attention tools are available by default. Run tools require --allow-run. Continue, Skip, Teach, and Escalate require --allow-attended-actions; Continue and Skip additionally require a qualified deployment --config. See docs/DISTRIBUTION.md." } } } diff --git a/smithery.yaml b/smithery.yaml index 5f7561a..8c6ae19 100644 --- a/smithery.yaml +++ b/smithery.yaml @@ -1,4 +1,4 @@ -# Smithery configuration for openadapt-agent (Experimental v2). +# Smithery configuration for openadapt-agent (Beta v2). # # Exposes compiled openadapt-flow workflow bundles as governed MCP tools # over stdio. The bundle directory is the operator's OWN private artifact: @@ -22,15 +22,43 @@ startCommand: bundles). This is your own private artifact. allowRun: type: boolean - title: Allow execution (run_* tools) + title: Allow governed workflow execution default: false description: >- When false (default), only the read-only tools (list_workflows, - get_workflow, get_run_report) are registered. When true, each - loadable bundle also gets a run_ tool that executes through - the governed `openadapt-flow run` CLI (fail-closed admission - gates). A halt or refusal is always surfaced as such, never as a - fabricated success. + get_workflow, get_run_report, list_needs_attention, and + get_attention_item) are registered. When true, each loadable bundle + also gets a run_workflow_ tool that executes through the + governed `openadapt-flow run` CLI. A halt or refusal is never + presented as success. + runsDir: + type: string + title: Local runs directory + default: runs + description: >- + Local directory containing governed run reports, durable pauses, + and Needs Attention state. Protected evidence remains here. + allowAttendedActions: + type: boolean + title: Allow attended decisions + default: false + description: >- + Register governed Teach and Escalate actions for signed pauses. + When deploymentConfig is also set, register Continue and Skip + through Flow's live verifier and deterministic resume path. + deploymentConfig: + type: string + title: Flow deployment config + description: >- + Optional path to a qualified openadapt-flow deployment YAML. It is + required for Continue and Skip and is fixed when the server starts. + headed: + type: boolean + title: Keep attended browser visible + default: false + description: >- + Keep a deployment-configured browser session visible to the local + operator. Required for browser Continue and Skip. bundleKey: type: string title: Bundle decryption key @@ -43,10 +71,17 @@ startCommand: args: [ 'serve', '--bundles', config.bundlesDir, - ...(config.allowRun ? ['--allow-run'] : []) + '--runs-dir', config.runsDir || 'runs', + ...(config.allowRun ? ['--allow-run'] : []), + ...(config.allowAttendedActions ? ['--allow-attended-actions'] : []), + ...(config.deploymentConfig ? ['--config', config.deploymentConfig] : []), + ...(config.headed ? ['--headed'] : []) ], env: config.bundleKey ? { OPENADAPT_BUNDLE_KEY: config.bundleKey } : {} }) exampleConfig: bundlesDir: /path/to/compiled/bundles allowRun: false + runsDir: runs + allowAttendedActions: false + headed: false diff --git a/src/openadapt_agent/__init__.py b/src/openadapt_agent/__init__.py index 140af64..b3cc2d5 100644 --- a/src/openadapt_agent/__init__.py +++ b/src/openadapt_agent/__init__.py @@ -1,8 +1,8 @@ -"""openadapt-agent — the agent-facing bridge for openadapt-flow workflows. +"""openadapt-agent — the local agent bridge for openadapt-flow workflows. -**Status: Experimental (v2).** This package exposes compiled +This package exposes compiled `openadapt-flow `_ workflow -bundles to other agents: +bundles and PHI-safe Needs Attention items to other agents: - an **MCP server** (``openadapt-agent serve`` / ``python -m openadapt_agent.mcp``) that presents each bundle as a tool whose @@ -11,14 +11,12 @@ gates; and - an **Agent Skills emitter** (``openadapt-agent emit-skill``) that wraps flow's own ``emit-skill`` and appends agent-facing MCP invocation and - halt-semantics guidance. + halt and attended-action guidance. -A halted or refused run is always surfaced as a structured halt/refusal -with an evidence pointer (the run's ``report.json``), never as success. - -Version 1.x of this package (a legacy execution wrapper for model-driven -GUI agents) is deprecated; see the README for the pointer to the last -v0.1.x release. +A halted or refused run is always surfaced as a PHI-safe structured +halt/refusal, never as success. Protected report evidence remains local +unless the operator explicitly enables protected export for a trusted +client in the same data boundary. """ __version__ = "2.0.0.dev0" diff --git a/src/openadapt_agent/attended.py b/src/openadapt_agent/attended.py new file mode 100644 index 0000000..4cfe20a --- /dev/null +++ b/src/openadapt_agent/attended.py @@ -0,0 +1,376 @@ +"""Local, capability-bound bridge to openadapt-flow's attended run contract. + +This module does not implement pause, resume, verification, or audit semantics. +It projects Flow's PHI-safe Needs Attention DTOs and submits exact +``AttendedActionRequest`` values back to Flow's durable engine API. +""" + +from __future__ import annotations + +import logging +import os +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional + +__all__ = [ + "AttendedBridge", + "AttendedBridgeError", + "AttendedTool", +] + +_ATTENTION_ID_RE = re.compile(r"^[0-9a-f]{24}$") +_CAPABILITY_DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$") +_IDEMPOTENCY_KEY_RE = re.compile(r"^[A-Za-z0-9._:-]{16,200}$") +_LOG = logging.getLogger(__name__) +_SAFE_DECISION_MESSAGES = { + "completed": ( + "Flow verified the attended outcome and completed the admitted " + "transition. Review the local run report for protected details." + ), + "halted": ( + "Flow admitted the attended transition, then halted safely at a later " + "condition. Review the refreshed local Needs Attention queue." + ), + "refused": ( + "Flow refused the action because the signed pause, authorization, or " + "current live state did not verify. Reload the item before deciding " + "what to do next." + ), + "needs_demonstration": ( + "The governed request for a corrective demonstration was recorded. " + "The durable pause remains available." + ), + "escalated": ( + "The escalation was recorded and the durable pause remains available " + "for qualified assistance." + ), +} + + +def _windows_operator_identity() -> str: + """Read the effective Windows token username without environment lookup.""" + try: + import ctypes + + buffer = ctypes.create_unicode_buffer(257) + size = ctypes.c_ulong(len(buffer)) + if ctypes.windll.advapi32.GetUserNameW(buffer, ctypes.byref(size)): + return buffer.value + except (AttributeError, OSError, ValueError): + pass + return "" + + +def _local_operator_identity() -> str: + """Use the effective OS account, never caller-controlled environment vars.""" + if os.name == "nt": + return _windows_operator_identity() + try: + import pwd + + return pwd.getpwuid(os.geteuid()).pw_name + except (ImportError, KeyError, OSError): + return "" + + +class AttendedBridgeError(Exception): + """A bounded attended-operation error safe to return to a local MCP user.""" + + +@dataclass(frozen=True) +class AttendedTool: + """One action-specific MCP tool contract.""" + + action: str + confirmation: str + disposition: str + description: str + + +ATTENDED_TOOLS: dict[str, AttendedTool] = { + "continue_attention": AttendedTool( + action="continue", + confirmation="human_completed", + disposition="completed_by_operator", + description=( + "After the local operator completes the paused task in the live " + "application, ask OpenAdapt to verify the exact postconditions and " + "independent effects, checkpoint it as human-completed, and resume " + "without actuating that task again." + ), + ), + "skip_attention": AttendedTool( + action="skip", + confirmation="confirmed_not_applicable", + disposition="not_applicable", + description=( + "Apply the workflow's already-declared non-consequential skip path. " + "Flow rechecks the exact guard and refuses any undeclared, " + "consequential, stale, or ambiguous skip." + ), + ), + "teach_attention": AttendedTool( + action="teach", + confirmation="request_demonstration", + disposition="teach_requested", + description=( + "Record an audited request for a corrective demonstration. The " + "existing governed teach/revision gate still decides whether the " + "demonstration is accepted, banked as progress, or refused." + ), + ), + "escalate_attention": AttendedTool( + action="escalate", + confirmation="request_assistance", + disposition="needs_assistance", + description=( + "Record an audited escalation while preserving the exact durable " + "pause for a qualified operator." + ), + ), +} + + +def action_input_schema(tool: AttendedTool) -> dict[str, Any]: + """Exact, free-text-free schema for one attended action tool.""" + return { + "type": "object", + "properties": { + "attention_id": { + "type": "string", + "pattern": _ATTENTION_ID_RE.pattern, + "description": "Opaque id returned by list_needs_attention.", + }, + "capability_digest": { + "type": "string", + "pattern": _CAPABILITY_DIGEST_RE.pattern, + "description": ( + "Exact current capability digest returned for this attention " + "item. A stale or changed digest is refused." + ), + }, + "idempotency_key": { + "type": "string", + "minLength": 16, + "maxLength": 200, + "pattern": _IDEMPOTENCY_KEY_RE.pattern, + "description": ( + "Caller-stable request id. Reuse the exact same value only " + "when retrying this exact action." + ), + }, + tool.confirmation: { + "type": "boolean", + "const": True, + "description": ( + "Explicit local-operator confirmation required before this " + "governed decision is submitted." + ), + }, + }, + "required": [ + "attention_id", + "capability_digest", + "idempotency_key", + tool.confirmation, + ], + "additionalProperties": False, + } + + +class AttendedBridge: + """Expose Flow's local Needs Attention and attended-action APIs safely.""" + + def __init__( + self, + runs_dir: Path, + *, + allow_actions: bool = False, + service: Optional[Any] = None, + operator: Optional[str] = None, + ) -> None: + self.runs_dir = Path(runs_dir) + self.allow_actions = allow_actions + self.service = service + self.operator = (operator if operator is not None else _local_operator_identity()).strip() + if allow_actions and not self.operator: + raise AttendedBridgeError( + "attended actions require a server-derived local operator identity" + ) + + @property + def live_actions_ready(self) -> bool: + """Continue/Skip need Flow's deployment-bound live executor.""" + return self.allow_actions and self.service is not None + + def enabled_action_tools(self) -> tuple[str, ...]: + if not self.allow_actions: + return () + tools = ["teach_attention", "escalate_attention"] + if self.live_actions_ready: + tools[0:0] = ["continue_attention", "skip_attention"] + return tuple(tools) + + def list(self) -> dict[str, Any]: + from openadapt_flow.console.attention import list_attention + + items = ( + list_attention(self.runs_dir) + if self.runs_dir.is_dir() and not self.runs_dir.is_symlink() + else [] + ) + return { + "open_count": len(items), + "actions_enabled": self.allow_actions, + "live_actions_ready": self.live_actions_ready, + "items": [item.model_dump(mode="json") for item in items], + } + + def _resolve(self, attention_id: str) -> tuple[Path, Any]: + if not isinstance(attention_id, str) or not _ATTENTION_ID_RE.fullmatch(attention_id): + raise AttendedBridgeError("attention_id must be an exact opaque queue id") + if not self.runs_dir.is_dir() or self.runs_dir.is_symlink(): + raise AttendedBridgeError("the configured runs directory is unavailable") + + from openadapt_flow.console.attention import resolve_attention + + resolved = resolve_attention(self.runs_dir, attention_id) + if resolved is None: + raise AttendedBridgeError("no current attention item has that id; reload the queue") + return resolved + + def get(self, attention_id: str) -> dict[str, Any]: + _path, item = self._resolve(attention_id) + return item.model_dump(mode="json") + + def for_run_dir(self, run_dir: Path | str | None) -> Optional[dict[str, Any]]: + """Return the PHI-safe attention projection for a just-finished run.""" + if not run_dir: + return None + if self.runs_dir.is_symlink(): + return None + original = Path(run_dir) + if original.is_symlink(): + return None + try: + root = self.runs_dir.resolve(strict=True) + path = original.resolve(strict=True) + path.relative_to(root) + except (FileNotFoundError, OSError, ValueError): + return None + if path.is_symlink(): + return None + from openadapt_flow.console.attention import attention_item + + item = attention_item(self.runs_dir, path) + return item.model_dump(mode="json") if item is not None else None + + def act(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]: + tool = ATTENDED_TOOLS.get(tool_name) + if tool is None: + raise AttendedBridgeError(f"unknown attended action tool: {tool_name}") + if tool_name not in self.enabled_action_tools(): + if tool.action in {"continue", "skip"} and self.allow_actions: + raise AttendedBridgeError( + "Continue/Skip require a deployment-bound live executor " + "configured when this server starts" + ) + raise AttendedBridgeError( + "attended actions are disabled: restart with --allow-attended-actions" + ) + + expected = { + "attention_id", + "capability_digest", + "idempotency_key", + tool.confirmation, + } + unknown = set(arguments) - expected + missing = expected - set(arguments) + if unknown or missing: + raise AttendedBridgeError( + "attended action arguments do not match the exact tool schema" + ) + if arguments[tool.confirmation] is not True: + raise AttendedBridgeError(f"{tool.confirmation} must be explicitly true") + capability_digest = arguments["capability_digest"] + idempotency_key = arguments["idempotency_key"] + if ( + not isinstance(capability_digest, str) + or not _CAPABILITY_DIGEST_RE.fullmatch(capability_digest) + or not isinstance(idempotency_key, str) + or not _IDEMPOTENCY_KEY_RE.fullmatch(idempotency_key) + ): + raise AttendedBridgeError("capability_digest or idempotency_key is malformed") + + path, item = self._resolve(arguments["attention_id"]) + capability = item.capability or {} + if capability.get("digest") != capability_digest: + raise AttendedBridgeError("the attention item changed; reload its exact capability") + if tool.action not in (capability.get("allowed_actions") or []): + raise AttendedBridgeError(f"the exact pause capability does not allow {tool.action}") + + from openadapt_flow.runtime.durable import ( + ApprovalRequired, + AttendedActionRefused, + AttendedActionRequest, + ResumeRefused, + execute_attended_action, + ) + + request = AttendedActionRequest( + capability_digest=capability_digest, + idempotency_key=idempotency_key, + action=tool.action, + disposition=tool.disposition, + ) + try: + if self.service is not None: + decision = self.service.execute( + path, + request, + operator=self.operator, + ) + else: + decision = execute_attended_action( + path, + request, + operator=self.operator, + ) + except (ApprovalRequired, AttendedActionRefused, ResumeRefused) as exc: + _LOG.info( + "Flow refused attended action %s for attention item %s: %s", + tool.action, + item.id, + exc, + ) + return { + "attention_id": item.id, + "action": tool.action, + "status": "refused", + "success": False, + "message": _SAFE_DECISION_MESSAGES["refused"], + } + except Exception as exc: + _LOG.exception("attended action failed without a safe terminal decision") + raise AttendedBridgeError( + "attended execution did not return a safe terminal decision; " + "inspect the durable audit and reconcile live state before retrying" + ) from exc + return { + "attention_id": item.id, + "decision_id": decision.decision_id, + "action": decision.action, + "status": decision.status, + "success": decision.status == "completed" and decision.report_success is True, + "message": _SAFE_DECISION_MESSAGES.get( + decision.status, + "Flow recorded the attended decision. Review the protected " + "local run evidence for details.", + ), + "report_success": decision.report_success, + "next_transition": decision.next_transition, + "transition_receipt_digest": decision.transition_receipt_digest, + } diff --git a/src/openadapt_agent/bridge.py b/src/openadapt_agent/bridge.py index 7f248cc..8f4caaa 100644 --- a/src/openadapt_agent/bridge.py +++ b/src/openadapt_agent/bridge.py @@ -6,35 +6,48 @@ Safety model (documented in ``docs/DESIGN.md``): -- Read-only tools (``list_workflows`` / ``get_workflow`` / - ``get_run_report``) are always registered. -- ``run_`` tools are registered ONLY when the operator started the +- PHI-safe read-only tools (workflow inspection, run summaries, and Needs + Attention projections) are always registered. +- ``run_`` tools are registered ONLY when the operator started the server with ``--allow-run``; even then, every call shells out to the governed ``openadapt-flow run`` CLI, so flow's fail-closed admission gates cannot be bypassed from here. -- The MCP client's user is the operator of record for anything a run - writes; the server itself performs no authentication. +- Needs Attention projections are read-only and PHI-safe. Attended mutations + are registered only under ``--allow-attended-actions`` and are submitted to + Flow's signed capability/idempotency/audit contract. +- Remote authentication is outside this local stdio process. The local OS + user is the operator of record for attended decisions. """ from __future__ import annotations import json +import logging from dataclasses import dataclass from pathlib import Path -from typing import Optional +from typing import Any, Optional +from openadapt_agent.attended import ( + ATTENDED_TOOLS, + AttendedBridge, + AttendedBridgeError, + action_input_schema, +) from openadapt_agent.bundles import ( WorkflowInfo, discover_bundles, tool_input_schema, ) -from openadapt_agent.runner import FlowRunner, RunnerConfig, is_safe_run_id +from openadapt_agent.runner import ( + FlowRunner, + RunnerConfig, + is_safe_run_id, + public_report_summary, +) __all__ = ["AgentBridge", "BridgeError", "ToolSpec"] -_EXPERIMENTAL_NOTE = ( - "openadapt-agent is EXPERIMENTAL (v2.0.0.dev0): unproven in production." -) +_LOG = logging.getLogger(__name__) class BridgeError(Exception): @@ -46,6 +59,21 @@ class ToolSpec: name: str description: str input_schema: dict + annotations: Optional[dict[str, Any]] = None + + +_READ_ONLY_ANNOTATIONS = { + "readOnlyHint": True, + "destructiveHint": False, + "idempotentHint": True, + "openWorldHint": False, +} +_RUN_ANNOTATIONS = { + "readOnlyHint": False, + "destructiveHint": True, + "idempotentHint": False, + "openWorldHint": True, +} class AgentBridge: @@ -58,14 +86,32 @@ def __init__( *, allow_run: bool = False, runner: Optional[FlowRunner] = None, + allow_attended_actions: bool = False, + attended_service: Optional[object] = None, + attended: Optional[AttendedBridge] = None, + allow_protected_export: bool = False, + allow_recorded_defaults: bool = False, ): self.bundles_dir = Path(bundles_dir) self.allow_run = allow_run + self.allow_protected_export = allow_protected_export + self.allow_recorded_defaults = allow_recorded_defaults self.runner_config = runner_config self.runner = runner or FlowRunner(runner_config) - self.workflows: dict[str, WorkflowInfo] = { - info.slug: info for info in discover_bundles(self.bundles_dir) - } + self.attended = attended or AttendedBridge( + runner_config.runs_dir, + allow_actions=allow_attended_actions, + service=attended_service, + ) + self.workflows: dict[str, WorkflowInfo] = {} + for info in discover_bundles(self.bundles_dir): + base_id = info.public_id + workflow_id = base_id + suffix = 2 + while workflow_id in self.workflows: + workflow_id = f"{base_id}_{suffix}" + suffix += 1 + self.workflows[workflow_id] = info # -- tool surface ------------------------------------------------------ @@ -74,42 +120,47 @@ def list_tool_specs(self) -> list[ToolSpec]: ToolSpec( name="list_workflows", description=( - "List the compiled openadapt-flow workflows this server " - "exposes (name, parameters, step count, whether run tools " - f"are enabled). {_EXPERIMENTAL_NOTE}" + "List workflows by opaque id with parameter names/types, " + "step count, availability, and whether run tools are " + "enabled. Demonstration labels, values, intents, paths, " + "and load exceptions stay local by default." ), input_schema={ "type": "object", "properties": {}, "additionalProperties": False, }, + annotations=_READ_ONLY_ANNOTATIONS, ), ToolSpec( name="get_workflow", description=( - "Inspect one workflow: step intents, declared parameters " - "with recorded example values, and certification status " - "(evaluated via `openadapt-flow certify` when the server " - "was configured with a policy)." + "Inspect one workflow's PHI-safe structural metadata and " + "certification result by opaque id. Recorded values, raw " + "intents, names, paths, and exception text stay local by " + "default." ), input_schema={ "type": "object", "properties": { "workflow": { "type": "string", - "description": "Workflow slug from list_workflows.", + "description": "Opaque id from list_workflows.", } }, "required": ["workflow"], "additionalProperties": False, }, + annotations=_READ_ONLY_ANNOTATIONS, ), ToolSpec( name="get_run_report", description=( - "Fetch the persisted report.json for a previous run made " - "through this server (by run_id from a run tool result). " - "This is the halt/success evidence trail." + "Fetch a PHI-safe status and count-only summary of a " + "persisted run by opaque run id. The raw report, observed " + "text, local paths, stdout, and stderr stay in the local " + "operator experience unless protected export was " + "explicitly enabled when the server started." ), input_schema={ "type": "object", @@ -122,23 +173,80 @@ def list_tool_specs(self) -> list[ToolSpec]: "required": ["run_id"], "additionalProperties": False, }, + annotations=_READ_ONLY_ANNOTATIONS, + ), + ToolSpec( + name="list_needs_attention", + description=( + "List PHI-safe local halt cards and their exact currently " + "allowed attended actions. Raw workflow names, observed " + "text, parameters, reports, and filesystem paths are not " + "returned." + ), + input_schema={ + "type": "object", + "properties": {}, + "additionalProperties": False, + }, + annotations=_READ_ONLY_ANNOTATIONS, + ), + ToolSpec( + name="get_attention_item", + description=( + "Reload one PHI-safe halt card by its opaque id before an " + "operator decision. Use the newly returned capability " + "digest; stale capabilities are refused." + ), + input_schema={ + "type": "object", + "properties": { + "attention_id": { + "type": "string", + "pattern": "^[0-9a-f]{24}$", + "description": ("Opaque id returned by list_needs_attention."), + } + }, + "required": ["attention_id"], + "additionalProperties": False, + }, + annotations=_READ_ONLY_ANNOTATIONS, ), ] + for tool_name in self.attended.enabled_action_tools(): + tool = ATTENDED_TOOLS[tool_name] + specs.append( + ToolSpec( + name=tool_name, + description=tool.description, + input_schema=action_input_schema(tool), + annotations={ + "readOnlyHint": False, + "destructiveHint": tool.action in {"continue", "skip"}, + "idempotentHint": True, + "openWorldHint": tool.action in {"continue", "skip"}, + }, + ) + ) if self.allow_run: - for slug, info in sorted(self.workflows.items()): + for workflow_id, info in sorted(self.workflows.items()): if not info.ok: continue n_steps = len(info.step_intents) + workflow_copy = ( + f"the compiled workflow {info.name!r}" + if self.allow_protected_export + else "the selected compiled workflow" + ) specs.append( ToolSpec( - name=f"run_{slug}", + name=f"run_{workflow_id}", description=( - f"Execute the compiled workflow {info.name!r} " - f"({n_steps} steps) via the governed " + f"Execute {workflow_copy} ({n_steps} steps) via the governed " "`openadapt-flow run` CLI (fail-closed admission " - "gates). Returns a structured outcome: status is " - "'success', 'halt' (run stopped with evidence in " - "report.json — NOT a success), 'refused' " + "gates). Returns a PHI-safe structured outcome: " + "status is 'success', 'halt' (run stopped with " + "protected evidence retained locally — NOT a " + "success), 'refused' " "(admission gate refused; nothing executed), " "'timeout', or 'error'. Never treat a non-success " "status as success." @@ -146,7 +254,9 @@ def list_tool_specs(self) -> list[ToolSpec]: input_schema=tool_input_schema( info, allow_url_override=self.runner_config.allow_url_override, + allow_recorded_defaults=self.allow_recorded_defaults, ), + annotations=_RUN_ANNOTATIONS, ) ) return specs @@ -161,101 +271,194 @@ def dispatch(self, name: str, arguments: Optional[dict]) -> dict: return self._get_workflow(arguments.get("workflow", "")) if name == "get_run_report": return self._get_run_report(arguments.get("run_id", "")) + if name == "list_needs_attention": + return self.attended.list() + if name == "get_attention_item": + try: + return self.attended.get(arguments.get("attention_id", "")) + except AttendedBridgeError as exc: + raise BridgeError(str(exc)) from exc + if name in ATTENDED_TOOLS: + try: + return self.attended.act(name, arguments) + except AttendedBridgeError as exc: + raise BridgeError(str(exc)) from exc if name.startswith("run_"): return self._run(name[len("run_") :], arguments) - raise BridgeError(f"unknown tool: {name}") + raise BridgeError("unknown tool name") def _list_workflows(self) -> dict: - return { - "experimental": True, + result = { + "schema_version": 1, + "lifecycle": "beta", "run_tools_enabled": self.allow_run, - "bundles_dir": str(self.bundles_dir), + "protected_export_enabled": self.allow_protected_export, + "synthetic_recorded_defaults_enabled": self.allow_recorded_defaults, "workflows": [ - { - "slug": info.slug, - "name": info.name or None, - "steps": len(info.step_intents), - "params": info.params, - "encrypted": info.encrypted, - "load_error": info.load_error, - } - for info in self.workflows.values() + self._workflow_projection(workflow_id, info) + for workflow_id, info in self.workflows.items() ], "note": ( - "Run tools are disabled unless the operator started the " - "server with --allow-run." + "Run tools are disabled unless the operator started the server with --allow-run." if not self.allow_run else None ), } + if self.allow_protected_export: + result["protected"] = {"bundles_dir": str(self.bundles_dir)} + return result - def _require_workflow(self, slug: str) -> WorkflowInfo: - info = self.workflows.get(slug) + def _workflow_projection( + self, + workflow_id: str, + info: WorkflowInfo, + ) -> dict: + result = { + "id": workflow_id, + "available": info.ok, + "step_count": len(info.step_intents) if info.ok else None, + "parameters": [ + { + "name": name, + "type": "string", + "required": not self.allow_recorded_defaults, + } + for name in sorted(info.params) + ], + "encrypted": info.encrypted, + } + if self.allow_protected_export: + result["protected"] = { + "slug": info.slug, + "name": info.name or None, + "bundle_dir": str(info.bundle_dir), + "recorded_params": info.params, + "step_intents": info.step_intents, + "load_error": info.load_error, + } + return result + + def _require_workflow(self, workflow_id: str) -> WorkflowInfo: + info = self.workflows.get(workflow_id) if info is None: - raise BridgeError( - f"unknown workflow {slug!r}; call list_workflows for the " - "available slugs" - ) + raise BridgeError("unknown workflow id; reload list_workflows") return info - def _get_workflow(self, slug: str) -> dict: - info = self._require_workflow(slug) + def _get_workflow(self, workflow_id: str) -> dict: + info = self._require_workflow(workflow_id) + result = self._workflow_projection(workflow_id, info) + result.update( + { + "schema_version": info.schema_version, + "certification": self._certification_projection(info), + } + ) + return result + + def _certification_projection(self, info: WorkflowInfo) -> dict: if not info.ok: + if info.load_error: + _LOG.warning( + "workflow %s could not be loaded locally: %s", + info.public_id, + info.load_error, + ) return { - "slug": info.slug, - "bundle_dir": str(info.bundle_dir), - "load_error": info.load_error, + "certified": None, + "message": "The local bundle could not be loaded safely.", } - return { - "slug": info.slug, - "name": info.name, - "bundle_dir": str(info.bundle_dir), - "schema_version": info.schema_version, - "encrypted": info.encrypted, - "params": info.params, - "steps": info.step_intents, - "certification": self.runner.certify(info.bundle_dir), - } + certification = self.runner.certify(info.bundle_dir) + certified = certification.get("certified") + if certified is True: + message = "The configured policy certification passed." + elif certified is False: + message = "The configured policy certification did not pass." + else: + message = "Certification was not evaluated by this server." + result = {"certified": certified, "message": message} + if self.allow_protected_export: + result["protected"] = certification + return result def _get_run_report(self, run_id: str) -> dict: if not run_id or not is_safe_run_id(run_id): raise BridgeError("run_id must be a single path component") - run_dir = (Path(self.runner_config.runs_dir) / run_id).resolve() - runs_root = Path(self.runner_config.runs_dir).resolve() + runs_candidate = Path(self.runner_config.runs_dir) + run_candidate = runs_candidate / run_id + if runs_candidate.is_symlink() or run_candidate.is_symlink(): + raise BridgeError("the configured run evidence boundary is unavailable") + run_dir = run_candidate.resolve() + runs_root = runs_candidate.resolve() if runs_root not in run_dir.parents: raise BridgeError("run_id resolves outside the server's runs directory") report_path = run_dir / "report.json" - if not report_path.is_file(): - raise BridgeError(f"no report.json for run_id {run_id!r}") + if report_path.is_symlink() or not report_path.is_file(): + raise BridgeError("no local report exists for that run id") try: report = json.loads(report_path.read_text()) except (OSError, json.JSONDecodeError) as exc: - raise BridgeError(f"report.json unreadable: {exc}") from exc - return {"run_id": run_id, "run_dir": str(run_dir), "report": report} + _LOG.exception("protected local report could not be read") + raise BridgeError("the local report could not be read safely") from exc + success = report.get("success") + status = "success" if success is True else ("halt" if success is False else "error") + messages = { + "success": "The persisted local report confirms success.", + "halt": ( + "The persisted local report confirms the run did not complete. " + "Review protected evidence in the local operator experience." + ), + "error": ( + "The persisted local report has no trustworthy terminal status. Review it locally." + ), + } + result = { + "schema_version": 1, + "run_id": run_id, + "status": status, + "success": status == "success", + "message": messages[status], + "summary": public_report_summary(report), + } + if self.allow_protected_export: + result["protected"] = { + "run_dir": str(run_dir), + "report": report, + } + return result - def _run(self, slug: str, arguments: dict) -> dict: - info = self._require_workflow(slug) + def _run(self, workflow_id: str, arguments: dict) -> dict: + info = self._require_workflow(workflow_id) if not self.allow_run: raise BridgeError( - "run tools are disabled: the operator did not start the " - "server with --allow-run" + "run tools are disabled: the operator did not start the server with --allow-run" ) if not info.ok: - raise BridgeError( - f"workflow {slug!r} could not be loaded: {info.load_error}" - ) + if info.load_error: + _LOG.warning( + "workflow %s could not be loaded locally: %s", + info.public_id, + info.load_error, + ) + raise BridgeError("the selected workflow could not be loaded safely") + arguments = dict(arguments) url_override = arguments.pop("url", None) unknown = set(arguments) - set(info.params) - if unknown: - raise BridgeError( - f"unknown parameter(s) {sorted(unknown)!r}; declared " - f"parameters: {sorted(info.params)!r}" - ) - params = {k: str(v) for k, v in arguments.items()} + missing = set(info.params) - set(arguments) + invalid_types = any(not isinstance(value, str) for value in arguments.values()) or ( + url_override is not None and not isinstance(url_override, str) + ) + if unknown or (missing and not self.allow_recorded_defaults) or invalid_types: + raise BridgeError("arguments do not match the declared workflow parameter schema") + params = dict(arguments) outcome = self.runner.run( - workflow=info.slug, + workflow=workflow_id, bundle_dir=info.bundle_dir, params=params, url_override=url_override, ) - return outcome.to_dict() + result = outcome.to_dict( + include_protected=self.allow_protected_export, + ) + if outcome.status == "halt": + result["needs_attention"] = self.attended.for_run_dir(outcome.run_dir) + return result diff --git a/src/openadapt_agent/bundles.py b/src/openadapt_agent/bundles.py index 5676021..b846ef7 100644 --- a/src/openadapt_agent/bundles.py +++ b/src/openadapt_agent/bundles.py @@ -11,9 +11,11 @@ from __future__ import annotations +import hashlib import re from dataclasses import dataclass, field from pathlib import Path +from typing import Any __all__ = [ "WorkflowInfo", @@ -22,6 +24,7 @@ "load_workflow_info", "slugify", "tool_input_schema", + "workflow_public_id", ] _BUNDLE_MARKERS = ("workflow.json", "workflow.json.enc") @@ -35,8 +38,9 @@ def is_bundle_dir(path: Path) -> bool: def slugify(name: str) -> str: """Lowercase, underscore-separated identifier for tool names. - Mirrors flow's skill-slug convention but uses underscores so the - result is also a valid ``run_`` MCP tool-name fragment. + Mirrors flow's skill-slug convention. The slug remains local metadata + unless protected export is explicitly enabled; default MCP tool names + use :func:`workflow_public_id`. """ slug = re.sub(r"[^a-z0-9]+", "_", name.lower()).strip("_") return slug or "workflow" @@ -47,6 +51,7 @@ class WorkflowInfo: """Metadata for one discovered bundle (or the error loading it).""" slug: str + public_id: str bundle_dir: Path name: str = "" params: dict[str, str] = field(default_factory=dict) @@ -77,12 +82,14 @@ def load_workflow_info(bundle_dir: Path, slug: str | None = None) -> WorkflowInf except Exception as exc: # fail-soft per bundle, loud in the listing return WorkflowInfo( slug=slug or slugify(bundle_dir.name), + public_id=workflow_public_id(bundle_dir), bundle_dir=bundle_dir, encrypted=encrypted, load_error=f"{type(exc).__name__}: {exc}", ) return WorkflowInfo( slug=slug or slugify(wf.name), + public_id=workflow_public_id(bundle_dir, workflow=wf), bundle_dir=bundle_dir, name=wf.name, params=dict(wf.params), @@ -123,23 +130,54 @@ def discover_bundles(root: Path) -> list[WorkflowInfo]: return infos -def tool_input_schema(info: WorkflowInfo, *, allow_url_override: bool = False) -> dict: +def workflow_public_id( + bundle_dir: Path, + *, + workflow: Any = None, +) -> str: + """Stable opaque MCP identifier without exporting bundle labels or paths.""" + digest = None + manifest = getattr(workflow, "manifest", None) + if manifest is not None: + digest = getattr(manifest, "content_digest", None) + if isinstance(digest, str) and digest: + material = digest.encode("utf-8") + elif workflow is not None: + material = workflow.model_dump_json().encode("utf-8") + else: + hasher = hashlib.sha256() + for marker in _BUNDLE_MARKERS: + path = Path(bundle_dir) / marker + if not path.is_file() or path.is_symlink(): + continue + hasher.update(marker.encode("ascii")) + try: + hasher.update(path.read_bytes()) + except OSError: + continue + material = hasher.digest() + opaque = hashlib.sha256(b"openadapt-agent-workflow-id-v1\0" + material).hexdigest()[:24] + return f"workflow_{opaque}" + + +def tool_input_schema( + info: WorkflowInfo, + *, + allow_url_override: bool = False, + allow_recorded_defaults: bool = False, +) -> dict: """JSON schema for a workflow's run tool, derived from declared params. - Every workflow parameter becomes an optional string property whose - default is the recorded example value (the same fallback the flow CLI - applies when a ``--param`` is omitted). ``url`` is only present when - the operator started the server with ``--allow-url-override``. + Recorded demonstration values never enter the schema. Parameters are + required unless the operator explicitly enabled synthetic recorded-default + reuse when starting the server. ``url`` is present only under the separate + URL-override opt-in. """ properties: dict[str, dict] = {} - for name, example in info.params.items(): + for name in info.params: properties[name] = { "type": "string", - "description": ( - f"Value for the {name!r} workflow parameter " - f"(default: the recorded example value {example!r})." - ), - "default": example, + "description": f"Value for the declared {name!r} workflow parameter.", } if allow_url_override: properties["url"] = { @@ -152,6 +190,6 @@ def tool_input_schema(info: WorkflowInfo, *, allow_url_override: bool = False) - return { "type": "object", "properties": properties, - "required": [], + "required": [] if allow_recorded_defaults else sorted(info.params), "additionalProperties": False, } diff --git a/src/openadapt_agent/cli.py b/src/openadapt_agent/cli.py index 68df116..6b64cf5 100644 --- a/src/openadapt_agent/cli.py +++ b/src/openadapt_agent/cli.py @@ -2,13 +2,14 @@ Subcommands: -- ``serve`` — expose a directory of compiled openadapt-flow bundles as an - MCP server over stdio. Read-only tools are always on; ``run_*`` tools - require the explicit ``--allow-run`` flag. +- ``serve`` — expose compiled openadapt-flow bundles and the local Needs + Attention queue over MCP stdio. PHI-safe read-only tools are always on; + workflow runs and attended decisions require separate operator flags. - ``emit-skill`` — emit a Claude Agent Skill folder for one bundle (wraps ``openadapt-flow emit-skill`` and appends MCP + halt guidance). -EXPERIMENTAL: see the README's honest-limits section before deploying. +The bridge is local stdio. Remote identity, tenancy, and transport are provided +by OpenAdapt Cloud rather than added to this process. """ from __future__ import annotations @@ -29,15 +30,12 @@ def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="openadapt-agent", description=( - "EXPERIMENTAL agent-facing bridge for openadapt-flow: expose " - "compiled workflow bundles as MCP tools and Agent Skills. " - "Execution always shells out to the governed `openadapt-flow " - "run` CLI." + "Agent-facing bridge for openadapt-flow: expose compiled workflow " + "bundles, Needs Attention, and governed operator decisions as " + "local MCP tools and Agent Skills." ), ) - parser.add_argument( - "--version", action="version", version=f"%(prog)s {__version__}" - ) + parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}") sub = parser.add_subparsers(dest="command", required=True) p = sub.add_parser( @@ -60,6 +58,35 @@ def build_parser() -> argparse.ArgumentParser: "this flag no MCP client can execute anything." ), ) + p.add_argument( + "--allow-protected-export", + action="store_true", + help=( + "DANGER: export raw workflow labels, recorded values, intents, " + "local paths, reports, stdout, stderr, and exception detail to the " + "MCP client. Off by default; enable only for an explicitly trusted " + "local client inside the protected data boundary." + ), + ) + p.add_argument( + "--allow-synthetic-recorded-defaults", + action="store_true", + help=( + "DEMO ONLY: let omitted workflow parameters reuse recorded values. " + "Use only with synthetic demonstrations; production requires every " + "declared parameter so a run cannot silently target the recorded " + "customer or record." + ), + ) + p.add_argument( + "--allow-attended-actions", + action="store_true", + help=( + "Register governed Teach/Escalate tools for signed durable pauses. " + "With --config, also register Continue/Skip through Flow's " + "deployment-bound live verifier and deterministic resume path." + ), + ) p.add_argument( "--url", default=None, @@ -110,6 +137,22 @@ def build_parser() -> argparse.ArgumentParser: "may contain spaces, e.g. 'openadapt-flow'" ), ) + p.add_argument( + "--headed", + action="store_true", + help=( + "Keep a deployment-configured attended web session visible to the " + "local operator (required for web Continue/Skip)." + ), + ) + p.add_argument( + "--allow-model-grounding", + action="store_true", + help=( + "Explicit CLI opt-in for configured off-box model grounding. A " + "deployment config may also opt in; otherwise verification remains local." + ), + ) p.add_argument( "--extra-run-arg", action="append", @@ -131,9 +174,7 @@ def build_parser() -> argparse.ArgumentParser: ), ) p.add_argument("bundle", help="Workflow bundle directory") - p.add_argument( - "--out", required=True, help="Parent directory for the skill folder" - ) + p.add_argument("--out", required=True, help="Parent directory for the skill folder") p.set_defaults(func=_cmd_emit_skill) return parser @@ -141,14 +182,27 @@ def build_parser() -> argparse.ArgumentParser: def _cmd_serve(args: argparse.Namespace) -> int: from openadapt_agent.bridge import AgentBridge + from openadapt_agent.flow_service import open_attended_service from openadapt_agent.mcp import serve from openadapt_agent.runner import default_flow_cli + if args.allow_attended_actions and args.flow_cli: + print( + "serve: attended actions require the openadapt-flow installed in " + "this interpreter; --flow-cli cannot select a different runtime", + file=sys.stderr, + ) + return 2 + if args.allow_synthetic_recorded_defaults and not args.allow_run: + print( + "serve: --allow-synthetic-recorded-defaults requires --allow-run", + file=sys.stderr, + ) + return 2 + runner_config = RunnerConfig( - flow_cli=( - tuple(shlex.split(args.flow_cli)) if args.flow_cli else default_flow_cli() - ), + flow_cli=(tuple(shlex.split(args.flow_cli)) if args.flow_cli else default_flow_cli()), runs_dir=Path(args.runs_dir), url=args.url, deployment_config=args.config, @@ -158,20 +212,40 @@ def _cmd_serve(args: argparse.Namespace) -> int: extra_run_args=tuple(args.extra_run_arg), ) try: - bridge = AgentBridge( - Path(args.bundles), runner_config, allow_run=args.allow_run - ) - except FileNotFoundError as exc: + with open_attended_service( + enabled=args.allow_attended_actions, + deployment_config=args.config, + url=args.url, + headed=args.headed, + allow_model_grounding=args.allow_model_grounding, + ) as attended_service: + bridge = AgentBridge( + Path(args.bundles), + runner_config, + allow_run=args.allow_run, + allow_attended_actions=args.allow_attended_actions, + attended_service=attended_service, + allow_protected_export=args.allow_protected_export, + allow_recorded_defaults=args.allow_synthetic_recorded_defaults, + ) + n = len(bridge.workflows) + print( + f"openadapt-agent {__version__}: serving {n} workflow(s) " + "over local stdio; run tools " + f"{'enabled' if args.allow_run else 'disabled'}; attended " + f"decisions {'enabled' if args.allow_attended_actions else 'disabled'}; " + "live Continue/Skip " + f"{'ready' if bridge.attended.live_actions_ready else 'not configured'}; " + "protected MCP export " + f"{'ENABLED' if args.allow_protected_export else 'disabled'}; " + "synthetic recorded defaults " + f"{'ENABLED' if args.allow_synthetic_recorded_defaults else 'disabled'}", + file=sys.stderr, + ) + serve(bridge) + except (FileNotFoundError, RuntimeError) as exc: print(f"serve: {exc}", file=sys.stderr) return 2 - n = len(bridge.workflows) - print( - f"openadapt-agent {__version__} (EXPERIMENTAL): serving {n} " - f"workflow(s) over stdio; run tools " - f"{'ENABLED' if args.allow_run else 'disabled (read-only)'}", - file=sys.stderr, - ) - serve(bridge) return 0 diff --git a/src/openadapt_agent/flow_service.py b/src/openadapt_agent/flow_service.py new file mode 100644 index 0000000..056e796 --- /dev/null +++ b/src/openadapt_agent/flow_service.py @@ -0,0 +1,51 @@ +"""Construct Flow's public persistent attended-action service.""" + +from __future__ import annotations + +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Iterator, Optional + +__all__ = ["open_attended_service"] + + +@contextmanager +def open_attended_service( + *, + enabled: bool, + deployment_config: Optional[str], + url: Optional[str], + headed: bool, + allow_model_grounding: bool, +) -> Iterator[Optional[Any]]: + """Open Flow's thread-owned service from one public DeploymentConfig. + + Agent never constructs a backend, Replayer, verifier, owner thread, or + continuation. Flow owns the visible session and the complete attended + decision lifecycle behind ``AttendedActionService.execute``. + """ + if not enabled or not deployment_config: + yield None + return + + from openadapt_flow.deployment import load_deployment + from openadapt_flow.runtime.durable import AttendedActionService + + deployment = load_deployment(Path(deployment_config)) + backend = deployment.backend + runtime = deployment.runtime + if url is not None: + backend = backend.model_copy(update={"url": url}) + if headed: + backend = backend.model_copy(update={"headed": True}) + if allow_model_grounding: + runtime = runtime.model_copy(update={"allow_model_grounding": True}) + deployment = deployment.model_copy( + update={ + "backend": backend, + "runtime": runtime, + } + ) + + with AttendedActionService(deployment) as service: + yield service diff --git a/src/openadapt_agent/mcp.py b/src/openadapt_agent/mcp.py index 019692a..94b6a7f 100644 --- a/src/openadapt_agent/mcp.py +++ b/src/openadapt_agent/mcp.py @@ -13,6 +13,7 @@ from __future__ import annotations import json +import logging from typing import Any import anyio @@ -20,11 +21,67 @@ from mcp.server.lowlevel import Server from mcp.server.stdio import stdio_server +from openadapt_agent.attended import ATTENDED_TOOLS from openadapt_agent.bridge import AgentBridge, BridgeError __all__ = ["build_server", "serve"] SERVER_NAME = "openadapt-agent" +_LOG = logging.getLogger(__name__) + +_CONFIRMATION_COPY = { + "continue_attention": ( + "A person must already have completed the paused task in the visible " + "application. Confirm that OpenAdapt may verify that outcome and resume " + "the governed workflow after it. The completed task will not be performed again." + ), + "skip_attention": ( + "Confirm that this paused task is not applicable. OpenAdapt will skip " + "only if the compiled workflow and exact signed capability declare that " + "skip safe; otherwise it will refuse." + ), + "teach_attention": ( + "Confirm that OpenAdapt should record an audited request for a corrective " + "demonstration. This does not directly rewrite or promote the workflow." + ), + "escalate_attention": ( + "Confirm that OpenAdapt should record an audited escalation and preserve " + "the exact durable pause for qualified assistance." + ), +} + + +async def _confirm_attended_action(server: Server, name: str) -> None: + """Require a second, protocol-native human confirmation before mutation.""" + context = server.request_context + session = context.session + params = session.client_params + elicitation = params.capabilities.elicitation if params is not None else None + if elicitation is None or elicitation.form is None: + raise BridgeError( + "attended actions require an MCP client with form elicitation so " + "the local operator can confirm this exact decision; use Flow's " + "attended console/CLI when the client does not support elicitation" + ) + result = await session.elicit_form( + _CONFIRMATION_COPY[name], + { + "type": "object", + "properties": { + "confirmed": { + "type": "boolean", + "title": "I confirm this attended action", + } + }, + "required": ["confirmed"], + }, + related_request_id=context.request_id, + ) + content = result.content or {} + if result.action != "accept" or content.get("confirmed") is not True: + raise BridgeError( + "the local operator declined or cancelled; no attended action was submitted" + ) def build_server(bridge: AgentBridge) -> Server: @@ -32,14 +89,19 @@ def build_server(bridge: AgentBridge) -> Server: server: Server = Server( SERVER_NAME, instructions=( - "EXPERIMENTAL bridge exposing compiled openadapt-flow workflow " - "bundles as tools. run_* tools execute via the governed " + "Local bridge exposing compiled openadapt-flow workflow bundles " + "and PHI-safe Needs Attention items as tools. run_* tools execute via the governed " "`openadapt-flow run` CLI and return a structured outcome: only " "status 'success' means the workflow completed and verified. " - "'halt' means the run stopped with evidence (fetch it with " - "get_run_report); 'refused' means an admission gate refused the " - "bundle and nothing executed. Never report a halted, refused, or " - "timed-out run as a success." + "'halt' means the run stopped and protected evidence remains in " + "the local operator experience; get_run_report returns only a " + "PHI-safe status/count summary by default. 'refused' means an " + "admission gate refused the bundle and nothing executed. " + "Continue/Skip require a human action " + "plus protocol-native operator elicitation, an exact signed " + "capability, live revalidation, and a stable idempotency key; they " + "never re-actuate the human-completed step. " + "Never report a halted, refused, or timed-out run as a success." ), ) @@ -50,6 +112,11 @@ async def _list_tools() -> list[types.Tool]: name=spec.name, description=spec.description, inputSchema=spec.input_schema, + annotations=( + types.ToolAnnotations(**spec.annotations) + if spec.annotations is not None + else None + ), ) for spec in bridge.list_tool_specs() ] @@ -57,18 +124,44 @@ async def _list_tools() -> list[types.Tool]: @server.call_tool() async def _call_tool(name: str, arguments: dict[str, Any] | None): try: - # dispatch() shells out to the flow CLI (blocking); keep the - # event loop responsive by running it in a worker thread. - result = await anyio.to_thread.run_sync( - lambda: bridge.dispatch(name, dict(arguments or {})) - ) + if name in ATTENDED_TOOLS: + await _confirm_attended_action(server, name) + + def call() -> dict[str, Any]: + return bridge.dispatch(name, dict(arguments or {})) + + # CLI runs and filesystem projections are blocking. Live attended + # actions synchronously submit to their own non-async backend-owner + # thread, so every MCP call can leave the event loop responsive. + result = await anyio.to_thread.run_sync(call) except BridgeError as exc: - return [ - types.TextContent( - type="text", - text=json.dumps({"error": str(exc)}), - ) - ] + return types.CallToolResult( + content=[ + types.TextContent( + type="text", + text=json.dumps({"error": str(exc)}), + ) + ], + isError=True, + ) + except Exception: + _LOG.exception("MCP tool dispatch failed inside the protected boundary") + return types.CallToolResult( + content=[ + types.TextContent( + type="text", + text=json.dumps( + { + "error": ( + "Local tool execution failed safely. Inspect " + "the protected local logs before retrying." + ) + } + ), + ) + ], + isError=True, + ) return [types.TextContent(type="text", text=json.dumps(result, indent=2))] return server @@ -77,9 +170,7 @@ async def _call_tool(name: str, arguments: dict[str, Any] | None): async def _run_stdio(bridge: AgentBridge) -> None: server = build_server(bridge) async with stdio_server() as (read_stream, write_stream): - await server.run( - read_stream, write_stream, server.create_initialization_options() - ) + await server.run(read_stream, write_stream, server.create_initialization_options()) def serve(bridge: AgentBridge) -> None: diff --git a/src/openadapt_agent/runner.py b/src/openadapt_agent/runner.py index 629d131..e910b96 100644 --- a/src/openadapt_agent/runner.py +++ b/src/openadapt_agent/runner.py @@ -1,12 +1,12 @@ """Shell out to the governed ``openadapt-flow run`` CLI and map the outcome. -This module is the only execution path in openadapt-agent. It NEVER -reimplements replay: every run is a subprocess invocation of +This module is the only new-run path in openadapt-agent. It NEVER +reimplements replay: every ``run_*`` call is a subprocess invocation of ``openadapt-flow run`` (the fail-closed deployment verb), so flow's admission gates (certification, identity arming, effect contracts, encryption, integrity pinning) apply exactly as they would from a terminal. -Exit-code contract of ``openadapt-flow run`` (v1.9/v1.10): +Exit-code contract of ``openadapt-flow run``: - ``0`` — run executed and every step verified: **success**. - ``1`` — run executed and stopped: **halt** (evidence in the run @@ -22,12 +22,12 @@ from __future__ import annotations import json +import logging import os import re import subprocess import sys import tempfile -import time import uuid from dataclasses import dataclass, field from pathlib import Path @@ -40,10 +40,28 @@ "classify_outcome", "default_flow_cli", "is_safe_run_id", + "public_report_summary", ] _RUN_ID_RE = re.compile(r"^[A-Za-z0-9._-]+$") _TAIL_CHARS = 4000 +_LOG = logging.getLogger(__name__) +_PUBLIC_MESSAGES = { + "success": ("The governed run completed and its persisted report confirms success."), + "halt": ( + "The governed run stopped safely instead of guessing. Review the local " + "Needs Attention experience for protected evidence." + ), + "refused": ("A governed admission check refused the run. The target workflow did not start."), + "timeout": ( + "The run exceeded its deadline. Its live effect is uncertain; inspect " + "the protected local run before retrying." + ), + "error": ( + "The run could not produce a trustworthy terminal result. Inspect the " + "protected local logs and run evidence." + ), +} def default_flow_cli() -> tuple[str, ...]: @@ -88,38 +106,88 @@ class RunOutcome: stdout_tail: str = "" stderr_tail: str = "" - def to_dict(self) -> dict: - return { + def to_dict(self, *, include_protected: bool = False) -> dict: + """Project an MCP-safe result; raw local evidence is explicit opt-in.""" + result = { + "schema_version": 1, "status": self.status, "success": self.status == "success", - "workflow": self.workflow, + "workflow_id": self.workflow, "run_id": self.run_id, - "run_dir": self.run_dir, - "report_path": self.report_path, - "exit_code": self.exit_code, - "detail": self.detail, - "halt": self.halt, - "summary": self.summary, - "stdout_tail": self.stdout_tail, - "stderr_tail": self.stderr_tail, + "message": _PUBLIC_MESSAGES.get( + self.status, + _PUBLIC_MESSAGES["error"], + ), + "summary": _sanitize_public_summary(self.summary), } + if include_protected: + result["protected"] = { + "workflow": self.workflow, + "run_dir": self.run_dir, + "report_path": self.report_path, + "exit_code": self.exit_code, + "detail": self.detail, + "halt": self.halt, + "stdout_tail": self.stdout_tail, + "stderr_tail": self.stderr_tail, + } + return result def _tail(text: str) -> str: return text[-_TAIL_CHARS:] if text else "" +def _sanitize_public_summary(values: dict) -> dict: + summary: dict[str, int | float | bool] = {} + for key in ( + "steps_total", + "steps_ok", + "steps_skipped", + "heal_count", + "model_calls", + ): + value = values.get(key) + if isinstance(value, int) and not isinstance(value, bool) and value >= 0: + summary[key] = value + total_ms = values.get("total_ms") + if isinstance(total_ms, (int, float)) and not isinstance(total_ms, bool) and total_ms >= 0: + summary["total_ms"] = total_ms + screenshots_egress = values.get("screenshots_may_leave_box") + if isinstance(screenshots_egress, bool): + summary["screenshots_may_leave_box"] = screenshots_egress + return summary + + +def public_report_summary(report: dict) -> dict: + """Return count/boolean metrics only; never report labels or text.""" + results = report.get("results") + results = results if isinstance(results, list) else [] + summary: dict[str, int | float | bool] = { + "steps_total": len(results), + "steps_ok": sum( + 1 for result in results if isinstance(result, dict) and result.get("ok") is True + ), + "steps_skipped": sum( + 1 for result in results if isinstance(result, dict) and result.get("skipped") is True + ), + } + for key in ("heal_count", "model_calls"): + value = report.get(key) + if isinstance(value, int) and not isinstance(value, bool) and value >= 0: + summary[key] = value + total_ms = report.get("total_ms") + if isinstance(total_ms, (int, float)) and not isinstance(total_ms, bool) and total_ms >= 0: + summary["total_ms"] = total_ms + screenshots_egress = report.get("screenshots_may_leave_box") + if isinstance(screenshots_egress, bool): + summary["screenshots_may_leave_box"] = screenshots_egress + return _sanitize_public_summary(summary) + + def _report_summary(report: dict) -> dict: return { - "workflow_name": report.get("workflow_name"), - "steps_total": len(report.get("results") or []), - "steps_ok": sum(1 for r in (report.get("results") or []) if r.get("ok")), - "heal_count": report.get("heal_count"), - "model_calls": report.get("model_calls"), - "total_ms": report.get("total_ms"), - "terminal_outcome": report.get("terminal_outcome"), - "screenshots_may_leave_box": report.get("screenshots_may_leave_box"), - "governed_authorization_id": report.get("governed_authorization_id"), + **public_report_summary(report), } @@ -151,7 +219,7 @@ def classify_outcome( Pure function (no I/O) so the mapping is unit-testable. The invariant: ``status == "success"`` requires BOTH exit code 0 AND a persisted report whose ``success`` flag is true — anything else surfaces as a - halt, refusal, or error with evidence pointers. + halt, refusal, or error while protected evidence remains local. """ outcome = RunOutcome( status="error", @@ -225,8 +293,7 @@ def classify_outcome( ) else: outcome.detail = ( - "Run exited nonzero before a report.json was written; see " - "stdout_tail / stderr_tail." + "Run exited nonzero before a report.json was written; see stdout_tail / stderr_tail." ) return outcome @@ -283,13 +350,11 @@ def run( runs_root = Path(self.config.runs_dir) runs_root.mkdir(parents=True, exist_ok=True) - run_id = f"{workflow}-{time.strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:6]}" + run_id = f"run-{uuid.uuid4().hex[:24]}" run_dir = runs_root / run_id report_path = run_dir / "report.json" - params_fd, params_name = tempfile.mkstemp( - prefix="openadapt_agent_params_", suffix=".json" - ) + params_fd, params_name = tempfile.mkstemp(prefix="openadapt_agent_params_", suffix=".json") params_file = Path(params_name) try: with os.fdopen(params_fd, "w") as fh: @@ -331,6 +396,16 @@ def run( "install openadapt-flow in the server's environment." ), ) + except (OSError, subprocess.SubprocessError, ValueError) as exc: + _LOG.exception("governed Flow subprocess failed locally") + return RunOutcome( + status="error", + workflow=workflow, + run_id=run_id, + run_dir=str(run_dir), + report_path=(str(report_path) if report_path.exists() else None), + detail=f"{type(exc).__name__}: {exc}", + ) finally: try: params_file.unlink(missing_ok=True) @@ -384,6 +459,12 @@ def certify(self, bundle_dir: Path) -> dict: return {"certified": None, "detail": "certify timed out"} except FileNotFoundError: return {"certified": None, "detail": "openadapt-flow CLI not found"} + except (OSError, subprocess.SubprocessError, ValueError) as exc: + _LOG.exception("Flow certification subprocess failed locally") + return { + "certified": None, + "detail": f"{type(exc).__name__}: {exc}", + } return { "certified": proc.returncode == 0, "exit_code": proc.returncode, diff --git a/src/openadapt_agent/skill.py b/src/openadapt_agent/skill.py index 937e1ab..d08a585 100644 --- a/src/openadapt_agent/skill.py +++ b/src/openadapt_agent/skill.py @@ -7,37 +7,37 @@ 1. how to invoke the workflow through this package's MCP server (``openadapt-agent serve``) instead of the raw CLI, and -2. the halt semantics an agent must respect (exit codes, run report - evidence, and the rule that a halt is never a success). +2. the halt and attended-action semantics an agent must respect. """ from __future__ import annotations from pathlib import Path -from openadapt_agent.bundles import load_workflow_info, slugify +from openadapt_agent.bundles import load_workflow_info __all__ = ["emit_agent_skill"] _APPENDIX_TEMPLATE = """\ -## Invoking via MCP (openadapt-agent, Experimental) +## Invoking via MCP (openadapt-agent) -If the operator runs the EXPERIMENTAL `openadapt-agent` MCP server over -this skill's bundle directory: +If the operator runs the local `openadapt-agent` MCP server over this +skill's bundle directory: ```bash openadapt-agent serve --bundles {bundle_ref} --allow-run ``` -this workflow appears as the MCP tool `run_{slug}` (parameters: -{param_list}). Prefer the MCP tool when it is available: it executes +this workflow appears as the opaque MCP tool `run_{workflow_id}` +(required parameters: {param_list}). Recorded demonstration values are +never placed in the tool schema and every declared parameter is required +by default. Prefer the MCP tool when it is available: it executes through the governed `openadapt-flow run` path (fail-closed admission gates) rather than the permissive `replay` demo path, and it returns a -structured outcome instead of raw CLI output. Read-only tools -`list_workflows`, `get_workflow`, and `get_run_report` are always -available; `run_{slug}` exists only when the operator started the server -with `--allow-run`. +PHI-safe structured outcome instead of raw CLI output. Inspection and +PHI-safe Needs Attention tools are always available; `run_{workflow_id}` exists only +when the operator started the server with `--allow-run`. ## Halt semantics (IMPORTANT) @@ -48,16 +48,42 @@ - **halt** (exit code 1 / MCP status `halt`) — the run executed and STOPPED at an unhandled state. This is the system working as designed (halt instead of guessing), but it is NOT a success: the workflow's - effect did not fully happen. Evidence lives in the run directory's - `report.json` (structured `halt` record: where it stopped, what it - observed, what completed before) and `REPORT.md`. Surface the halt and - its evidence to the user; do not retry blindly and never claim success. + effect did not fully happen. Protected evidence remains in the local + OpenAdapt operator experience; default MCP results contain only opaque + IDs, fixed messages, and count/boolean metrics. Surface the halt to the + user; do not retry blindly and never claim success. - **governed refusal** (exit code 2 / MCP status `refused`, `openadapt-flow run` only) — an admission gate refused the bundle before execution; NOTHING was executed. The printed coverage report names the failing gate. Never summarize a halted, refused, or timed-out run as if it succeeded. + +## Needs Attention + +When a run halts, `list_needs_attention` and `get_attention_item` return +an opaque, PHI-safe queue card. Do not ask for or place credentials, +challenge answers, screenshots, observed text, or other protected values +in an attended-action payload. + +If the operator enabled attended actions, use only the exact action tool +that matches their explicit decision: + +- `continue_attention` only after the local operator says they completed + the paused task in the live application. Flow revalidates the outcome + and resumes after it; it never performs that completed action again. +- `skip_attention` only for an allowed, declared skip. +- `teach_attention` to request a corrective demonstration. +- `escalate_attention` to preserve the pause for qualified assistance. + +Reload the item immediately before acting, pass its exact capability +digest, and use one stable idempotency key for retries of that same +request. The MCP server will separately elicit the person's confirmation; +a client without form elicitation must direct the operator to Flow's +attended console/CLI instead. Elicitation is host-mediated explicit +confirmation, not cryptographic human-presence or identity proof. Never +infer or answer that elicitation yourself, auto-retry an uncertain +delivery, or substitute one action for another. """ @@ -77,15 +103,11 @@ def emit_agent_skill(bundle_dir: Path | str, out_dir: Path | str) -> Path: skill_dir = emit_skill(bundle_dir, out_dir) info = load_workflow_info(Path(bundle_dir)) - slug = slugify(info.name or Path(bundle_dir).name) - appendix = _APPENDIX_TEMPLATE.format( - slug=slug, + workflow_id=info.public_id, bundle_ref="/bundle", param_list=_format_params(info.params), ) skill_md = skill_dir / "SKILL.md" - skill_md.write_text( - skill_md.read_text(encoding="utf-8") + appendix, encoding="utf-8" - ) + skill_md.write_text(skill_md.read_text(encoding="utf-8") + appendix, encoding="utf-8") return skill_dir diff --git a/tests/golden/skill_appendix.md b/tests/golden/skill_appendix.md index 1a27752..5549685 100644 --- a/tests/golden/skill_appendix.md +++ b/tests/golden/skill_appendix.md @@ -1,21 +1,22 @@ -## Invoking via MCP (openadapt-agent, Experimental) +## Invoking via MCP (openadapt-agent) -If the operator runs the EXPERIMENTAL `openadapt-agent` MCP server over -this skill's bundle directory: +If the operator runs the local `openadapt-agent` MCP server over this +skill's bundle directory: ```bash openadapt-agent serve --bundles /bundle --allow-run ``` -this workflow appears as the MCP tool `run_demo_triage` (parameters: -`note`). Prefer the MCP tool when it is available: it executes +this workflow appears as the opaque MCP tool `run_` +(required parameters: `note`). Recorded demonstration values are +never placed in the tool schema and every declared parameter is required +by default. Prefer the MCP tool when it is available: it executes through the governed `openadapt-flow run` path (fail-closed admission gates) rather than the permissive `replay` demo path, and it returns a -structured outcome instead of raw CLI output. Read-only tools -`list_workflows`, `get_workflow`, and `get_run_report` are always -available; `run_demo_triage` exists only when the operator started the server -with `--allow-run`. +PHI-safe structured outcome instead of raw CLI output. Inspection and +PHI-safe Needs Attention tools are always available; `run_` exists only +when the operator started the server with `--allow-run`. ## Halt semantics (IMPORTANT) @@ -26,13 +27,39 @@ A run has exactly one of these outcomes — report it faithfully: - **halt** (exit code 1 / MCP status `halt`) — the run executed and STOPPED at an unhandled state. This is the system working as designed (halt instead of guessing), but it is NOT a success: the workflow's - effect did not fully happen. Evidence lives in the run directory's - `report.json` (structured `halt` record: where it stopped, what it - observed, what completed before) and `REPORT.md`. Surface the halt and - its evidence to the user; do not retry blindly and never claim success. + effect did not fully happen. Protected evidence remains in the local + OpenAdapt operator experience; default MCP results contain only opaque + IDs, fixed messages, and count/boolean metrics. Surface the halt to the + user; do not retry blindly and never claim success. - **governed refusal** (exit code 2 / MCP status `refused`, `openadapt-flow run` only) — an admission gate refused the bundle before execution; NOTHING was executed. The printed coverage report names the failing gate. Never summarize a halted, refused, or timed-out run as if it succeeded. + +## Needs Attention + +When a run halts, `list_needs_attention` and `get_attention_item` return +an opaque, PHI-safe queue card. Do not ask for or place credentials, +challenge answers, screenshots, observed text, or other protected values +in an attended-action payload. + +If the operator enabled attended actions, use only the exact action tool +that matches their explicit decision: + +- `continue_attention` only after the local operator says they completed + the paused task in the live application. Flow revalidates the outcome + and resumes after it; it never performs that completed action again. +- `skip_attention` only for an allowed, declared skip. +- `teach_attention` to request a corrective demonstration. +- `escalate_attention` to preserve the pause for qualified assistance. + +Reload the item immediately before acting, pass its exact capability +digest, and use one stable idempotency key for retries of that same +request. The MCP server will separately elicit the person's confirmation; +a client without form elicitation must direct the operator to Flow's +attended console/CLI instead. Elicitation is host-mediated explicit +confirmation, not cryptographic human-presence or identity proof. Never +infer or answer that elicitation yourself, auto-retry an uncertain +delivery, or substitute one action for another. diff --git a/tests/test_attended_bridge.py b/tests/test_attended_bridge.py new file mode 100644 index 0000000..aee6031 --- /dev/null +++ b/tests/test_attended_bridge.py @@ -0,0 +1,513 @@ +"""Needs Attention and attended decisions stay on Flow's exact engine contract.""" + +from __future__ import annotations + +import json +import os + +import pytest + +from openadapt_agent.attended import AttendedBridgeError +from openadapt_agent.bridge import AgentBridge, BridgeError +from openadapt_agent.flow_service import open_attended_service +from openadapt_agent.runner import RunnerConfig + + +@pytest.fixture() +def paused_attention(tmp_path): + from openadapt_flow.ir import ( + ActionKind, + HaltObservation, + Postcondition, + PostconditionKind, + RunReport, + Step, + StepResult, + Workflow, + ) + from openadapt_flow.runtime.durable import ( + CheckpointStore, + PendingEscalation, + RunManifest, + issue_attended_capability, + ) + + workflow = Workflow( + name="Attended reference", + steps=[ + Step( + id="human", + intent="complete the local challenge", + action=ActionKind.KEY, + key="Enter", + expect=[ + Postcondition( + kind=PostconditionKind.TEXT_PRESENT, + text="DONE", + ) + ], + ), + Step( + id="next", + intent="continue deterministically", + action=ActionKind.KEY, + key="Tab", + ), + ], + ) + bundle = tmp_path / "bundles" / "attended" + workflow.save(bundle) + runs = tmp_path / "runs" + run = runs / "run-one" + store = CheckpointStore(run) + store.write_manifest( + RunManifest( + run_id="run-instance-a", + workflow_name=workflow.name, + bundle_dir=str(bundle), + params={}, + ) + ) + pending = PendingEscalation( + workflow_name=workflow.name, + step_index=0, + step_id="human", + intent="complete the local challenge", + category="human_required", + reason="local human presence required", + proposed_options=["complete locally", "resume"], + resume_from_index=0, + ) + store.write_pending(pending) + failed = StepResult( + step_id="human", + intent="complete the local challenge", + ok=False, + error="local human presence required", + ) + RunReport( + workflow_name=workflow.name, + started_at="2026-07-19T00:00:00+00:00", + success=False, + results=[failed], + halt=HaltObservation( + state_id="human", + intent="complete the local challenge", + reason="local human presence required", + ), + ).save(run) + capability = issue_attended_capability( + run, + store=store, + pending=pending, + workflow=workflow, + result=failed, + ) + return { + "bundle": bundle, + "bundles": bundle.parent, + "runs": runs, + "run": run, + "store": store, + "capability": capability, + } + + +class ResultExecutor: + def __init__(self): + self.calls = 0 + + def continue_run(self, _run_dir, capability, _approval): + from openadapt_flow.runtime.durable import AttendedExecutionResult + + self.calls += 1 + return AttendedExecutionResult( + status="completed", + message="human outcome verified; deterministic continuation completed", + report_success=True, + resumed_from=capability.step_id, + next_transition=capability.expected_next_transition, + ) + + def skip_run(self, run_dir, capability, approval): + return self.continue_run(run_dir, capability, approval) + + +class FailingExecutor(ResultExecutor): + def continue_run(self, _run_dir, _capability, _approval): + self.calls += 1 + raise RuntimeError("backend transport failed after delivery began") + + +class DirectFlowService: + """Test double that still executes Flow's exact public action function.""" + + def __init__(self, executor): + self.executor = executor + + def execute(self, run_dir, request, *, operator): + from openadapt_flow.runtime.durable import execute_attended_action + + return execute_attended_action( + run_dir, + request, + operator=operator, + executor=self.executor, + ) + + +def make_bridge(paused, *, allow_actions=False, service=None): + return AgentBridge( + paused["bundles"], + RunnerConfig( + flow_cli=("openadapt-flow-stub",), + runs_dir=paused["runs"], + ), + allow_attended_actions=allow_actions, + attended_service=service, + ) + + +def item_and_args(bridge, *, action="continue"): + item = bridge.dispatch("list_needs_attention", {})["items"][0] + confirmation = { + "continue": "human_completed", + "skip": "confirmed_not_applicable", + "teach": "request_demonstration", + "escalate": "request_assistance", + }[action] + return item, { + "attention_id": item["id"], + "capability_digest": item["capability"]["digest"], + "idempotency_key": f"stable-request-{action}-0001", + confirmation: True, + } + + +def test_needs_attention_is_phi_safe_and_always_readable(paused_attention): + bridge = make_bridge(paused_attention) + specs = {spec.name for spec in bridge.list_tool_specs()} + assert {"list_needs_attention", "get_attention_item"} <= specs + assert not (set(("continue_attention", "skip_attention")) & specs) + + listing = bridge.dispatch("list_needs_attention", {}) + assert listing["open_count"] == 1 + assert listing["actions_enabled"] is False + serialized = json.dumps(listing) + assert "Attended reference" not in serialized + assert "local human presence required" not in serialized + assert str(paused_attention["run"]) not in serialized + + item = listing["items"][0] + assert item["human_required"] is True + assert item["capability"]["allowed_actions"] == [ + "continue", + "teach", + "escalate", + ] + assert bridge.dispatch("get_attention_item", {"attention_id": item["id"]}) == item + for invalid in ("../run-one", "a/b", "", "0" * 23): + with pytest.raises(BridgeError): + bridge.dispatch("get_attention_item", {"attention_id": invalid}) + with pytest.raises(BridgeError): + bridge.dispatch("get_attention_item", {"attention_id": 123}) + + +def test_continue_is_exactly_bound_and_idempotent_without_reactuation( + paused_attention, +): + executor = ResultExecutor() + bridge = make_bridge( + paused_attention, + allow_actions=True, + service=DirectFlowService(executor), + ) + specs = {spec.name for spec in bridge.list_tool_specs()} + assert { + "continue_attention", + "skip_attention", + "teach_attention", + "escalate_attention", + } <= specs + continue_schema = next( + spec.input_schema for spec in bridge.list_tool_specs() if spec.name == "continue_attention" + ) + assert continue_schema["additionalProperties"] is False + assert continue_schema["properties"]["human_completed"]["const"] is True + assert "challenge_answer" not in continue_schema["properties"] + item, arguments = item_and_args(bridge) + + first = bridge.dispatch("continue_attention", dict(arguments)) + second = bridge.dispatch("continue_attention", dict(arguments)) + assert first == second + assert first["status"] == "completed" + assert first["success"] is True + assert executor.calls == 1 + decisions = json.loads((paused_attention["run"] / "attended_decisions.json").read_text())[ + "decisions" + ] + assert [decision["status"] for decision in decisions] == [ + "prepared", + "delivery_started", + "completed", + ] + + +def test_stale_capability_extra_fields_and_false_confirmation_never_execute( + paused_attention, +): + executor = ResultExecutor() + bridge = make_bridge( + paused_attention, + allow_actions=True, + service=DirectFlowService(executor), + ) + _item, arguments = item_and_args(bridge) + + wrong = dict(arguments, capability_digest="sha256:" + "0" * 64) + with pytest.raises(BridgeError, match="changed"): + bridge.dispatch("continue_attention", wrong) + extra = dict(arguments, challenge_answer="never-crosses-this-boundary") + with pytest.raises(BridgeError, match="exact tool schema"): + bridge.dispatch("continue_attention", extra) + unconfirmed = dict(arguments, human_completed=False) + with pytest.raises(BridgeError, match="explicitly true"): + bridge.dispatch("continue_attention", unconfirmed) + + pending = paused_attention["store"].read_pending() + paused_attention["store"].write_pending(pending.model_copy(update={"step_id": "different"})) + refused = bridge.dispatch("continue_attention", dict(arguments)) + assert refused["status"] == "refused" + assert refused["success"] is False + assert executor.calls == 0 + assert not (paused_attention["run"] / "approval.json").exists() + + +def test_flow_refusal_details_do_not_cross_the_phi_safe_bridge(paused_attention): + class ProtectedRefusalService: + def execute(self, _run_dir, _request, *, operator): + assert operator + from openadapt_flow.runtime.durable import AttendedActionRefused + + raise AttendedActionRefused("protected patient name and local workflow details") + + bridge = make_bridge( + paused_attention, + allow_actions=True, + service=ProtectedRefusalService(), + ) + _item, arguments = item_and_args(bridge) + + result = bridge.dispatch("continue_attention", arguments) + serialized = json.dumps(result) + assert result["status"] == "refused" + assert "protected patient" not in serialized + assert "local workflow" not in serialized + assert "signed pause" in result["message"] + + +def test_uncertain_delivery_is_audited_and_never_automatically_retried( + paused_attention, +): + executor = FailingExecutor() + bridge = make_bridge( + paused_attention, + allow_actions=True, + service=DirectFlowService(executor), + ) + _item, arguments = item_and_args(bridge) + + with pytest.raises(BridgeError, match="reconcile live state"): + bridge.dispatch("continue_attention", dict(arguments)) + retry = bridge.dispatch("continue_attention", dict(arguments)) + + assert retry["status"] == "refused" + assert retry["success"] is False + assert executor.calls == 1 + decisions = json.loads((paused_attention["run"] / "attended_decisions.json").read_text())[ + "decisions" + ] + assert [decision["status"] for decision in decisions] == [ + "prepared", + "delivery_started", + "delivery_uncertain", + ] + + +def test_teach_and_escalate_are_audited_without_a_live_executor(paused_attention): + bridge = make_bridge(paused_attention, allow_actions=True, service=None) + specs = {spec.name for spec in bridge.list_tool_specs()} + assert {"teach_attention", "escalate_attention"} <= specs + assert "continue_attention" not in specs + + _item, teach_args = item_and_args(bridge, action="teach") + teach = bridge.dispatch("teach_attention", teach_args) + assert teach["status"] == "needs_demonstration" + assert teach["success"] is False + + _item, escalate_args = item_and_args(bridge, action="escalate") + escalated = bridge.dispatch("escalate_attention", escalate_args) + assert escalated["status"] == "escalated" + assert escalated["success"] is False + assert paused_attention["store"].read_pending() is not None + + with pytest.raises(BridgeError, match="deployment-bound"): + bridge.dispatch( + "continue_attention", + { + **teach_args, + "idempotency_key": "stable-request-continue-0002", + "human_completed": True, + "request_demonstration": True, + }, + ) + + +def test_public_service_wrapper_applies_explicit_deployment_overrides( + monkeypatch, +): + import openadapt_flow.deployment as deployment_mod + import openadapt_flow.runtime.durable as durable_mod + from openadapt_flow.deployment import DeploymentConfig + + captured = {} + + class FakeService: + def __init__(self, deployment): + captured["deployment"] = deployment + + def __enter__(self): + captured["entered"] = True + return self + + def __exit__(self, *_args): + captured["closed"] = True + + monkeypatch.setattr( + deployment_mod, + "load_deployment", + lambda _path: DeploymentConfig(), + ) + monkeypatch.setattr( + durable_mod, + "AttendedActionService", + FakeService, + raising=False, + ) + with open_attended_service( + enabled=True, + deployment_config="deployment.yaml", + url="https://example.invalid/app", + headed=True, + allow_model_grounding=True, + ) as service: + assert isinstance(service, FakeService) + + deployment = captured["deployment"] + assert deployment.backend.url == "https://example.invalid/app" + assert deployment.backend.headed is True + assert deployment.runtime.allow_model_grounding is True + assert captured["entered"] is True + assert captured["closed"] is True + + +def test_exact_flow_public_service_context_execute_and_close( + monkeypatch, + paused_attention, +): + from openadapt_flow.deployment import DeploymentConfig + from openadapt_flow.runtime.durable import ( + AttendedActionRequest, + AttendedActionService, + ) + + closed = [] + + class FakeBackend: + def close(self): + closed.append(True) + + monkeypatch.setattr( + "openadapt_flow.backends.factory.build_backend", + lambda _config: FakeBackend(), + ) + deployment = DeploymentConfig( + backend={ + "kind": "windows", + "agent_url": "http://127.0.0.1:5001", + } + ) + request = AttendedActionRequest( + capability_digest=paused_attention["capability"].digest, + idempotency_key="public-service-contract-0001", + action="teach", + disposition="teach_requested", + ) + + with AttendedActionService(deployment) as service: + decision = service.execute( + paused_attention["run"], + request, + operator="local-staff", + ) + assert decision.status == "needs_demonstration" + service.close() + assert closed == [True] + + +def test_attended_bridge_requires_local_operator_identity(paused_attention): + from openadapt_agent.attended import AttendedBridge + + with pytest.raises(AttendedBridgeError, match="operator identity"): + AttendedBridge( + paused_attention["runs"], + allow_actions=True, + operator="", + ) + + +@pytest.mark.skipif(os.name == "nt", reason="POSIX effective-uid identity contract") +def test_operator_identity_ignores_spoofed_posix_environment( + monkeypatch, + paused_attention, +): + import pwd + + from openadapt_agent.attended import AttendedBridge + + monkeypatch.setenv("USER", "spoofed-agent") + monkeypatch.setenv("LOGNAME", "spoofed-agent") + bridge = AttendedBridge(paused_attention["runs"], allow_actions=True) + assert bridge.operator == pwd.getpwuid(os.geteuid()).pw_name + assert bridge.operator != "spoofed-agent" + + +def test_windows_operator_identity_uses_token_api_not_environment(monkeypatch): + import ctypes + from types import SimpleNamespace + + from openadapt_agent.attended import _windows_operator_identity + + class Advapi: + @staticmethod + def GetUserNameW(buffer, _size): + buffer.value = "token-backed-user" + return 1 + + monkeypatch.setenv("USERNAME", "spoofed-agent") + monkeypatch.setattr( + ctypes, + "windll", + SimpleNamespace(advapi32=Advapi()), + raising=False, + ) + assert _windows_operator_identity() == "token-backed-user" + + +def test_symlinked_runs_root_is_not_scanned(paused_attention, tmp_path): + from openadapt_agent.attended import AttendedBridge + + alias = tmp_path / "runs-alias" + alias.symlink_to(paused_attention["runs"], target_is_directory=True) + bridge = AttendedBridge(alias) + assert bridge.list()["items"] == [] diff --git a/tests/test_bridge.py b/tests/test_bridge.py index 050240b..f6764f3 100644 --- a/tests/test_bridge.py +++ b/tests/test_bridge.py @@ -1,8 +1,9 @@ -"""Bridge tool surface: schema generation, --allow-run gating, report access.""" +"""Bridge tools keep protected workflow and run evidence local by default.""" from __future__ import annotations import json +import re import pytest from conftest import FlowCliStub @@ -13,22 +14,48 @@ from openadapt_agent.runner import RunnerConfig -def make_bridge(bundles_root, runner_config, allow_run=False): - return AgentBridge(bundles_root, runner_config, allow_run=allow_run) +def make_bridge( + bundles_root, + runner_config, + allow_run=False, + **kwargs, +): + return AgentBridge( + bundles_root, + runner_config, + allow_run=allow_run, + **kwargs, + ) + + +def workflow_id(bridge: AgentBridge) -> str: + assert len(bridge.workflows) == 1 + return next(iter(bridge.workflows)) + + +def run_tool(bridge: AgentBridge) -> str: + return f"run_{workflow_id(bridge)}" -def test_schema_from_fixture_bundle(bundles_root, runner_config): +def test_schema_uses_opaque_id_requires_params_and_exports_no_examples( + bundles_root, + runner_config, +): bridge = make_bridge(bundles_root, runner_config, allow_run=True) - specs = {s.name: s for s in bridge.list_tool_specs()} - spec = specs["run_demo_triage"] + workflow = workflow_id(bridge) + assert re.fullmatch(r"workflow_[0-9a-f]{24}", workflow) + spec = {item.name: item for item in bridge.list_tool_specs()}[f"run_{workflow}"] schema = spec.input_schema assert schema["type"] == "object" assert schema["additionalProperties"] is False - assert schema["required"] == [] - assert schema["properties"]["note"]["type"] == "string" - assert schema["properties"]["note"]["default"] == "Follow-up in 2 weeks" - # URL override NOT exposed unless the operator opted in. - assert "url" not in schema["properties"] + assert schema["required"] == ["note"] + assert schema["properties"]["note"] == { + "type": "string", + "description": "Value for the declared 'note' workflow parameter.", + } + serialized = json.dumps(spec.__dict__) + assert "Follow-up in 2 weeks" not in serialized + assert "Demo Triage" not in serialized def test_url_property_only_with_override_flag(bundle_dir): @@ -41,115 +68,329 @@ def test_url_property_only_with_override_flag(bundle_dir): assert "url" not in without_url["properties"] -def test_read_only_tools_without_allow_run(bundles_root, runner_config): +def test_read_only_tools_and_workflow_listing_are_phi_safe( + bundles_root, + runner_config, +): bridge = make_bridge(bundles_root, runner_config, allow_run=False) - names = [s.name for s in bridge.list_tool_specs()] - assert names == ["list_workflows", "get_workflow", "get_run_report"] + names = [spec.name for spec in bridge.list_tool_specs()] + assert names == [ + "list_workflows", + "get_workflow", + "get_run_report", + "list_needs_attention", + "get_attention_item", + ] listing = bridge.dispatch("list_workflows", {}) + assert listing["lifecycle"] == "beta" assert listing["run_tools_enabled"] is False - assert listing["workflows"][0]["slug"] == "demo_triage" - # Calling a run tool anyway is refused loudly. + assert listing["protected_export_enabled"] is False + item = listing["workflows"][0] + assert re.fullmatch(r"workflow_[0-9a-f]{24}", item["id"]) + assert item["step_count"] == 2 + assert item["parameters"] == [{"name": "note", "type": "string", "required": True}] + serialized = json.dumps(listing) + for protected in ( + "Demo Triage", + "Follow-up in 2 weeks", + "Open the patient chart", + str(bundles_root), + ): + assert protected not in serialized + with pytest.raises(BridgeError, match="--allow-run"): - bridge.dispatch("run_demo_triage", {}) + bridge.dispatch(f"run_{item['id']}", {"note": "fresh value"}) -def test_get_workflow_reports_metadata_and_uncertified_status( - bundles_root, runner_config +def test_get_workflow_returns_structural_metadata_not_recorded_content( + bundles_root, + runner_config, ): bridge = make_bridge(bundles_root, runner_config) - info = bridge.dispatch("get_workflow", {"workflow": "demo_triage"}) - assert info["name"] == "Demo Triage" - assert info["steps"] == ["Open the patient chart", "Type the triage note"] - assert info["params"] == {"note": "Follow-up in 2 weeks"} - # No policy configured -> certification honestly "not evaluated". + workflow = workflow_id(bridge) + info = bridge.dispatch("get_workflow", {"workflow": workflow}) + assert info["id"] == workflow + assert info["step_count"] == 2 + assert info["parameters"][0]["name"] == "note" assert info["certification"]["certified"] is None + serialized = json.dumps(info) + assert "Demo Triage" not in serialized + assert "Follow-up in 2 weeks" not in serialized + assert "Open the patient chart" not in serialized + assert str(bundles_root) not in serialized -def test_run_dispatch_success_and_report_roundtrip( - monkeypatch, bundles_root, runner_config, success_report +def test_run_success_and_report_roundtrip_are_phi_safe( + monkeypatch, + bundles_root, + runner_config, + success_report, ): + secret = "Jane Roe MRN-9911 bearer sk_live_secret" + success_report["workflow_name"] = secret + success_report["results"][0]["intent"] = secret + success_report["stdout"] = secret stub = FlowCliStub(exit_code=0, report=success_report) + stub.stdout = secret + stub.stderr = secret monkeypatch.setattr(runner_mod.subprocess, "run", stub) bridge = make_bridge(bundles_root, runner_config, allow_run=True) - result = bridge.dispatch("run_demo_triage", {"note": "custom note"}) + result = bridge.dispatch(run_tool(bridge), {"note": "custom note"}) assert result["status"] == "success" - # The persisted report is retrievable via the read-only tool. + assert result["success"] is True + assert re.fullmatch(r"run-[0-9a-f]{24}", result["run_id"]) + assert "protected" not in result + assert secret not in json.dumps(result) + assert str(runner_config.runs_dir) not in json.dumps(result) + fetched = bridge.dispatch("get_run_report", {"run_id": result["run_id"]}) - assert fetched["report"]["success"] is True + assert fetched["status"] == "success" + assert fetched["summary"]["steps_ok"] == 2 + assert "protected" not in fetched + assert secret not in json.dumps(fetched) + assert str(runner_config.runs_dir) not in json.dumps(fetched) -def test_run_dispatch_halt_surfaces_evidence( - monkeypatch, bundles_root, runner_config, halt_report +def test_run_halt_returns_safe_card_not_raw_evidence( + monkeypatch, + bundles_root, + runner_config, + halt_report, ): + secret = "Jane Roe MRN-9911 Unexpected secret dialog" + halt_report["workflow_name"] = secret + halt_report["halt"]["reason"] = secret + halt_report["halt"]["observed_texts"] = [secret] + halt_report["results"][1]["error"] = secret stub = FlowCliStub(exit_code=1, report=halt_report) + stub.stdout = secret + stub.stderr = secret monkeypatch.setattr(runner_mod.subprocess, "run", stub) bridge = make_bridge(bundles_root, runner_config, allow_run=True) - result = bridge.dispatch("run_demo_triage", {}) + result = bridge.dispatch(run_tool(bridge), {"note": "fresh note"}) assert result["status"] == "halt" assert result["success"] is False - assert result["halt"]["observed_texts"] == ["Unexpected dialog: Save changes?"] - assert result["report_path"].endswith("report.json") + assert "halt" not in result + assert "protected" not in result + assert "needs_attention" in result + assert secret not in json.dumps(result) + assert str(runner_config.runs_dir) not in json.dumps(result) -def test_unknown_parameter_rejected(monkeypatch, bundles_root, runner_config): +def test_missing_or_unknown_parameters_refuse_before_execution( + monkeypatch, + bundles_root, + runner_config, +): stub = FlowCliStub(exit_code=0) monkeypatch.setattr(runner_mod.subprocess, "run", stub) bridge = make_bridge(bundles_root, runner_config, allow_run=True) - with pytest.raises(BridgeError, match="unknown parameter"): - bridge.dispatch("run_demo_triage", {"nope": "x"}) + tool = run_tool(bridge) + with pytest.raises(BridgeError, match="declared workflow parameter schema"): + bridge.dispatch(tool, {}) + with pytest.raises(BridgeError, match="declared workflow parameter schema"): + bridge.dispatch(tool, {"note": "fresh", "secret-unknown": "x"}) + with pytest.raises(BridgeError, match="declared workflow parameter schema"): + bridge.dispatch(tool, {"note": {"unexpected": "object"}}) assert stub.calls == [] -def test_get_run_report_traversal_refused(bundles_root, runner_config): +def test_synthetic_recorded_default_reuse_is_separate_explicit_mode( + monkeypatch, + bundles_root, + runner_config, + success_report, +): + stub = FlowCliStub(exit_code=0, report=success_report) + monkeypatch.setattr(runner_mod.subprocess, "run", stub) + bridge = make_bridge( + bundles_root, + runner_config, + allow_run=True, + allow_recorded_defaults=True, + ) + spec = {item.name: item for item in bridge.list_tool_specs()}[run_tool(bridge)] + assert spec.input_schema["required"] == [] + assert "default" not in spec.input_schema["properties"]["note"] + assert "Follow-up in 2 weeks" not in json.dumps(spec.__dict__) + assert bridge.dispatch(run_tool(bridge), {})["status"] == "success" + + +def test_protected_export_is_explicit_and_preserves_local_detail( + monkeypatch, + bundles_root, + runner_config, + halt_report, +): + secret = "Jane Roe MRN-9911 protected halt" + halt_report["halt"]["reason"] = secret + stub = FlowCliStub(exit_code=1, report=halt_report) + stub.stderr = secret + monkeypatch.setattr(runner_mod.subprocess, "run", stub) + bridge = make_bridge( + bundles_root, + runner_config, + allow_run=True, + allow_protected_export=True, + ) + listing = bridge.dispatch("list_workflows", {}) + assert listing["protected_export_enabled"] is True + assert listing["protected"]["bundles_dir"] == str(bundles_root) + workflow = workflow_id(bridge) + detail = bridge.dispatch("get_workflow", {"workflow": workflow}) + assert detail["protected"]["name"] == "Demo Triage" + assert detail["protected"]["recorded_params"]["note"] == "Follow-up in 2 weeks" + + result = bridge.dispatch(f"run_{workflow}", {"note": "fresh"}) + assert secret in json.dumps(result["protected"]) + report = bridge.dispatch("get_run_report", {"run_id": result["run_id"]}) + assert secret in json.dumps(report["protected"]) + assert str(runner_config.runs_dir) in json.dumps(report["protected"]) + + +def test_get_run_report_traversal_and_symlink_are_refused( + bundles_root, + runner_config, + tmp_path, +): bridge = make_bridge(bundles_root, runner_config) for bad in ["../secrets", "a/b", "..", ""]: with pytest.raises(BridgeError): bridge.dispatch("get_run_report", {"run_id": bad}) + runs = runner_config.runs_dir + runs.mkdir() + outside = tmp_path / "outside" + outside.mkdir() + (outside / "report.json").write_text('{"success": true}') + (runs / ("run-" + "a" * 24)).symlink_to( + outside, + target_is_directory=True, + ) + with pytest.raises(BridgeError, match="boundary"): + bridge.dispatch( + "get_run_report", + {"run_id": "run-" + "a" * 24}, + ) + -def test_unknown_workflow_and_tool(bundles_root, runner_config): +def test_unreadable_report_exception_text_stays_local( + bundles_root, + runner_config, +): + secret = "Jane Roe MRN-9911 sk_live_secret" + run_id = "run-" + "b" * 24 + run = runner_config.runs_dir / run_id + run.mkdir(parents=True) + (run / "report.json").write_text("{not-json " + secret) bridge = make_bridge(bundles_root, runner_config) - with pytest.raises(BridgeError, match="unknown workflow"): - bridge.dispatch("get_workflow", {"workflow": "nope"}) - with pytest.raises(BridgeError, match="unknown tool"): - bridge.dispatch("frobnicate", {}) + with pytest.raises(BridgeError) as error: + bridge.dispatch("get_run_report", {"run_id": run_id}) + assert secret not in str(error.value) + assert str(run) not in str(error.value) + assert "could not be read safely" in str(error.value) + + +def test_unknown_workflow_and_tool_do_not_echo_untrusted_names( + bundles_root, + runner_config, +): + secret = "Jane Roe sk_live_secret" + bridge = make_bridge(bundles_root, runner_config) + with pytest.raises(BridgeError) as workflow_error: + bridge.dispatch("get_workflow", {"workflow": secret}) + assert secret not in str(workflow_error.value) + with pytest.raises(BridgeError) as tool_error: + bridge.dispatch(secret, {}) + assert secret not in str(tool_error.value) -def test_unloadable_bundle_listed_with_error_not_run(tmp_path, runner_config): - bad = tmp_path / "bundles" / "broken" + +def test_unloadable_bundle_is_safe_by_default_and_not_runnable( + tmp_path, + runner_config, +): + secret = "Jane Roe MRN-9911 sk_live_secret" + bad = tmp_path / f"bundles-{secret}" / "broken" bad.mkdir(parents=True) - (bad / "workflow.json").write_text("{not json") - bridge = AgentBridge(tmp_path / "bundles", runner_config, allow_run=True) + (bad / "workflow.json").write_text("{not json " + secret) + bridge = AgentBridge(bad.parent, runner_config, allow_run=True) listing = bridge.dispatch("list_workflows", {}) - assert listing["workflows"][0]["load_error"] - names = [s.name for s in bridge.list_tool_specs()] - assert not any(n.startswith("run_") for n in names) + assert listing["workflows"][0]["available"] is False + assert secret not in json.dumps(listing) + assert str(bad) not in json.dumps(listing) + names = [spec.name for spec in bridge.list_tool_specs()] + assert not any(name.startswith("run_") for name in names) def test_tool_result_is_json_serializable( - monkeypatch, bundles_root, runner_config, halt_report + monkeypatch, + bundles_root, + runner_config, + halt_report, ): stub = FlowCliStub(exit_code=1, report=halt_report) monkeypatch.setattr(runner_mod.subprocess, "run", stub) bridge = make_bridge(bundles_root, runner_config, allow_run=True) - result = bridge.dispatch("run_demo_triage", {}) - json.dumps(result) # must not raise + result = bridge.dispatch(run_tool(bridge), {"note": "fresh"}) + json.dumps(result) -def test_certify_forwarded_when_policy_configured( - monkeypatch, bundles_root, tmp_path +def test_certify_result_is_fixed_copy_unless_protected_export_enabled( + monkeypatch, + bundles_root, + tmp_path, ): + secret = "Jane Roe MRN-9911 unverified write" config = RunnerConfig( flow_cli=("openadapt-flow-stub",), runs_dir=tmp_path / "runs", policy="clinical-write", ) stub = FlowCliStub(exit_code=2) - stub.stdout = "certify FAILED: unverified write" + stub.stdout = secret monkeypatch.setattr(runner_mod.subprocess, "run", stub) bridge = AgentBridge(bundles_root, config) - info = bridge.dispatch("get_workflow", {"workflow": "demo_triage"}) + workflow = workflow_id(bridge) + info = bridge.dispatch("get_workflow", {"workflow": workflow}) assert info["certification"]["certified"] is False + assert secret not in json.dumps(info) cmd = stub.calls[0] assert cmd[1] == "certify" assert cmd[cmd.index("--policy") + 1] == "clinical-write" + + +def test_workflow_names_intents_recorded_values_and_paths_never_default_export( + tmp_path, +): + from openadapt_flow.ir import ActionKind, Step, Workflow + + secret = "Jane Roe MRN-9911 sk_live_secret" + bundle = tmp_path / secret / "bundle" + Workflow( + name=secret, + params={"patient_id": secret}, + steps=[ + Step( + id="step", + intent=secret, + action=ActionKind.KEY, + key="Tab", + ) + ], + ).save(bundle) + config = RunnerConfig( + flow_cli=("openadapt-flow-stub",), + runs_dir=tmp_path / "runs", + ) + bridge = AgentBridge(bundle, config, allow_run=True) + workflow = workflow_id(bridge) + surfaces = [ + bridge.dispatch("list_workflows", {}), + bridge.dispatch("get_workflow", {"workflow": workflow}), + [spec.__dict__ for spec in bridge.list_tool_specs()], + ] + serialized = json.dumps(surfaces) + assert secret not in serialized + assert str(bundle) not in serialized + assert "patient_id" in serialized diff --git a/tests/test_cli.py b/tests/test_cli.py new file mode 100644 index 0000000..a398e8c --- /dev/null +++ b/tests/test_cli.py @@ -0,0 +1,58 @@ +"""CLI keeps one exact Flow runtime across runs and attended decisions.""" + +from __future__ import annotations + +from openadapt_agent.cli import build_parser, main + + +def test_attended_flags_parse_as_server_fixed_configuration(tmp_path): + args = build_parser().parse_args( + [ + "serve", + "--bundles", + str(tmp_path / "bundles"), + "--runs-dir", + str(tmp_path / "runs"), + "--allow-run", + "--allow-attended-actions", + "--allow-protected-export", + "--allow-synthetic-recorded-defaults", + "--config", + "deployment.yaml", + "--headed", + ] + ) + assert args.allow_run is True + assert args.allow_attended_actions is True + assert args.allow_protected_export is True + assert args.allow_synthetic_recorded_defaults is True + assert args.config == "deployment.yaml" + assert args.headed is True + + +def test_custom_flow_cli_is_refused_when_attended_actions_are_enabled(tmp_path, capsys): + result = main( + [ + "serve", + "--bundles", + str(tmp_path / "bundles"), + "--allow-attended-actions", + "--flow-cli", + "different-openadapt-flow", + ] + ) + assert result == 2 + assert "cannot select a different runtime" in capsys.readouterr().err + + +def test_synthetic_recorded_defaults_require_run_authority(tmp_path, capsys): + result = main( + [ + "serve", + "--bundles", + str(tmp_path / "bundles"), + "--allow-synthetic-recorded-defaults", + ] + ) + assert result == 2 + assert "requires --allow-run" in capsys.readouterr().err diff --git a/tests/test_distribution.py b/tests/test_distribution.py index 418fa72..702dc3a 100644 --- a/tests/test_distribution.py +++ b/tests/test_distribution.py @@ -3,8 +3,9 @@ These files are how the package is listed in MCP registries. The tests pin them to the package's real identity so a version bump or a rename cannot silently desync the registry manifests, and they encode the -security-relevant invariant that a registry-launched server is read-only -by default (execution requires the operator to add --allow-run). +security-relevant invariant that a registry-launched server is PHI-safe and +read-only by default. Workflow execution and attended decisions require +separate operator opt-ins. """ from __future__ import annotations @@ -74,8 +75,8 @@ def test_serve_is_the_subcommand_and_bundles_is_required() -> None: def test_registry_launch_is_read_only_by_default() -> None: """A one-click registry install must NOT auto-enable execution. - --allow-run is deliberately absent from server.json/smithery defaults; - the operator opts into run_* tools explicitly. + --allow-run is deliberately absent from server.json defaults and false in + Smithery. Attended decisions have their own false-by-default switch. """ args = _server_json()["packages"][0]["packageArguments"] assert not any( @@ -85,6 +86,7 @@ def test_registry_launch_is_read_only_by_default() -> None: smithery = yaml.safe_load(SMITHERY_YAML.read_text(encoding="utf-8")) props = smithery["startCommand"]["configSchema"]["properties"] assert props["allowRun"]["default"] is False + assert props["allowAttendedActions"]["default"] is False # --allow-run is not forced into the required config. assert "allowRun" not in smithery["startCommand"]["configSchema"].get( "required", [] @@ -107,6 +109,10 @@ def test_smithery_stdio_command_wires_bundles_and_allow_run() -> None: assert "openadapt-agent" in command_fn assert "--bundles" in command_fn assert "--allow-run" in command_fn + assert "--allow-attended-actions" in command_fn + assert "--runs-dir" in command_fn + assert "--config" in command_fn + assert "--headed" in command_fn assert "OPENADAPT_BUNDLE_KEY" in command_fn @@ -116,7 +122,13 @@ def test_llms_txt_lists_the_tool_surface() -> None: "list_workflows", "get_workflow", "get_run_report", - "run_", + "list_needs_attention", + "get_attention_item", + "run_workflow_", + "continue_attention", + "skip_attention", + "teach_attention", + "escalate_attention", "docs.openadapt.ai", ): assert token in text diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index dd1ab65..825da11 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -2,11 +2,17 @@ from __future__ import annotations +import threading +from types import SimpleNamespace + import anyio import mcp.types as types +import pytest +import openadapt_agent.mcp as mcp_mod from openadapt_agent.bridge import AgentBridge -from openadapt_agent.mcp import build_server +from openadapt_agent.bridge import BridgeError +from openadapt_agent.mcp import _confirm_attended_action, build_server def test_server_builds_and_lists_bridge_tools(bundles_root, runner_config): @@ -22,10 +28,18 @@ async def list_tools(): names = [t.name for t in tools] assert "list_workflows" in names assert "get_run_report" in names - assert "run_demo_triage" in names - run_tool = next(t for t in tools if t.name == "run_demo_triage") + assert "list_needs_attention" in names + run_names = [name for name in names if name.startswith("run_workflow_")] + assert len(run_names) == 1 + run_tool = next(t for t in tools if t.name == run_names[0]) assert run_tool.inputSchema["properties"]["note"]["type"] == "string" + assert run_tool.inputSchema["required"] == ["note"] + assert "default" not in run_tool.inputSchema["properties"]["note"] assert "governed" in (run_tool.description or "") + assert run_tool.annotations.readOnlyHint is False + assert run_tool.annotations.destructiveHint is True + list_tool = next(t for t in tools if t.name == "list_needs_attention") + assert list_tool.annotations.readOnlyHint is True def test_server_read_only_when_run_not_allowed(bundles_root, runner_config): @@ -38,4 +52,167 @@ async def list_tools(): return [t.name for t in result.root.tools] names = anyio.run(list_tools) - assert names == ["list_workflows", "get_workflow", "get_run_report"] + assert names == [ + "list_workflows", + "get_workflow", + "get_run_report", + "list_needs_attention", + "get_attention_item", + ] + + +def test_bridge_refusals_are_mcp_error_results(bundles_root, runner_config): + runner_config.runs_dir.mkdir() + bridge = AgentBridge(bundles_root, runner_config) + server = build_server(bridge) + + async def call_missing_item(): + handler = server.request_handlers[types.CallToolRequest] + return await handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="get_attention_item", + arguments={"attention_id": "0" * 24}, + ) + ) + ) + + result = anyio.run(call_missing_item).root + assert result.isError is True + assert "no current attention item" in result.content[0].text + + +def test_unexpected_local_exception_text_never_crosses_mcp( + monkeypatch, + bundles_root, + runner_config, +): + secret = "Jane Roe MRN-9911 sk_live_secret /private/protected/path" + bridge = AgentBridge(bundles_root, runner_config) + server = build_server(bridge) + + def fail(_name, _arguments): + raise RuntimeError(secret) + + monkeypatch.setattr(bridge, "dispatch", fail) + + async def call(): + handler = server.request_handlers[types.CallToolRequest] + return await handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name="list_workflows", + arguments={}, + ) + ) + ) + + result = anyio.run(call).root + assert result.isError is True + assert secret not in result.content[0].text + assert "failed safely" in result.content[0].text + + +def test_blocking_calls_leave_the_mcp_event_loop(monkeypatch, bundles_root, runner_config): + bridge = AgentBridge( + bundles_root, + runner_config, + allow_attended_actions=True, + attended_service=object(), + ) + server = build_server(bridge) + seen: dict[str, int] = {} + + def dispatch(name, _arguments): + seen[name] = threading.get_ident() + return {"ok": True} + + monkeypatch.setattr(bridge, "dispatch", dispatch) + + async def confirm(_server, _name): + return None + + monkeypatch.setattr(mcp_mod, "_confirm_attended_action", confirm) + + async def call_both(): + event_thread = threading.get_ident() + handler = server.request_handlers[types.CallToolRequest] + calls = { + "continue_attention": { + "attention_id": "0" * 24, + "capability_digest": "sha256:" + "0" * 64, + "idempotency_key": "stable-thread-test-0001", + "human_completed": True, + }, + "list_needs_attention": {}, + } + for name, arguments in calls.items(): + await handler( + types.CallToolRequest( + params=types.CallToolRequestParams( + name=name, + arguments=arguments, + ) + ) + ) + return event_thread + + event_thread = anyio.run(call_both) + assert seen["continue_attention"] != event_thread + assert seen["list_needs_attention"] != event_thread + + +class ElicitationSession: + def __init__(self, *, action="accept", confirmed=True, supported=True): + self.client_params = SimpleNamespace( + capabilities=SimpleNamespace( + elicitation=(SimpleNamespace(form=object()) if supported else None) + ) + ) + self.result = SimpleNamespace( + action=action, + content={"confirmed": confirmed}, + ) + self.calls = [] + + async def elicit_form(self, message, schema, related_request_id=None): + self.calls.append((message, schema, related_request_id)) + return self.result + + +class ElicitationServer: + def __init__(self, session): + self.request_context = SimpleNamespace( + session=session, + request_id="request-123", + ) + + +def test_attended_action_requires_protocol_native_human_confirmation(): + session = ElicitationSession() + anyio.run( + _confirm_attended_action, + ElicitationServer(session), + "continue_attention", + ) + message, schema, request_id = session.calls[0] + assert "person must already have completed" in message + assert schema["properties"]["confirmed"]["type"] == "boolean" + assert request_id == "request-123" + + +@pytest.mark.parametrize( + ("session", "error"), + [ + (ElicitationSession(supported=False), "form elicitation"), + (ElicitationSession(action="decline"), "declined or cancelled"), + (ElicitationSession(confirmed=False), "declined or cancelled"), + ], +) +def test_attended_action_refuses_missing_or_declined_human_confirmation(session, error): + with pytest.raises(BridgeError, match=error): + anyio.run( + _confirm_attended_action, + ElicitationServer(session), + "continue_attention", + ) diff --git a/tests/test_runner.py b/tests/test_runner.py index e7d3399..510b1b5 100644 --- a/tests/test_runner.py +++ b/tests/test_runner.py @@ -9,6 +9,7 @@ import openadapt_agent.runner as runner_mod from openadapt_agent.runner import FlowRunner, RunnerConfig, classify_outcome +from openadapt_agent.runner import RunOutcome def _run(monkeypatch, runner_config, stub, **kwargs): @@ -106,9 +107,7 @@ def test_url_override_refused_without_flag(monkeypatch, runner_config, bundle_di assert stub.calls == [] # nothing executed -def test_url_override_honoured_with_flag( - monkeypatch, tmp_path, bundle_dir, success_report -): +def test_url_override_honoured_with_flag(monkeypatch, tmp_path, bundle_dir, success_report): config = RunnerConfig( flow_cli=("openadapt-flow-stub",), runs_dir=tmp_path / "runs", @@ -151,3 +150,35 @@ def test_other_nonzero_exit_is_halt_with_evidence_pointer(exit_code): outcome = classify_outcome("w", exit_code, None, stdout="boom", stderr="") assert outcome.status == "halt" assert outcome.to_dict()["success"] is False + + +@pytest.mark.parametrize( + "status", + ["success", "halt", "refused", "timeout", "error"], +) +def test_public_outcome_projection_never_exports_protected_cli_or_report_text( + status, +): + secret = "Jane Roe MRN-9911 sk_live_secret /private/protected/path" + outcome = RunOutcome( + status=status, + workflow="workflow_" + "a" * 24, + run_id="run-" + "b" * 24, + run_dir=secret, + report_path=secret, + exit_code=1, + detail=secret, + halt={"reason": secret, "observed_texts": [secret]}, + summary={"steps_total": 2, "steps_ok": 1, "diagnostic": secret}, + stdout_tail=secret, + stderr_tail=secret, + ) + + public = outcome.to_dict() + assert secret not in str(public) + assert "protected" not in public + assert public["status"] == status + assert public["success"] is (status == "success") + + opted_in = outcome.to_dict(include_protected=True) + assert secret in str(opted_in["protected"]) diff --git a/tests/test_skill.py b/tests/test_skill.py index fbf4f82..f6282e6 100644 --- a/tests/test_skill.py +++ b/tests/test_skill.py @@ -2,8 +2,10 @@ from __future__ import annotations +import re from pathlib import Path +from openadapt_agent.bundles import load_workflow_info from openadapt_agent.skill import emit_agent_skill GOLDEN = Path(__file__).parent / "golden" / "skill_appendix.md" @@ -22,15 +24,24 @@ def test_emit_skill_wraps_flow_and_appends_guidance(bundle_dir, tmp_path): # Our appendix is present, exactly as the golden file specifies. golden = GOLDEN.read_text() - assert text.endswith(golden) + normalized = re.sub( + r"run_workflow_[0-9a-f]{24}", + "run_", + text, + ) + assert normalized.endswith(golden) def test_appendix_names_the_mcp_tool_and_params(bundle_dir, tmp_path): skill_dir = emit_agent_skill(bundle_dir, tmp_path / "skills") text = (skill_dir / "SKILL.md").read_text() - assert "run_demo_triage" in text + public_id = load_workflow_info(skill_dir / "bundle").public_id + assert f"run_{public_id}" in text assert "`note`" in text + assert "every declared parameter is required" in text + assert "Follow-up in 2 weeks" not in text.split("## Invoking via MCP", 1)[1] # Honesty requirements: halt is never success; refusal executes nothing. assert "NOT a success" in text assert "NOTHING was executed" in text - assert "Experimental" in text + assert "continue_attention" in text + assert "never performs that completed action again" in text