diff --git a/README.md b/README.md index 91d206a..fca7047 100644 --- a/README.md +++ b/README.md @@ -100,7 +100,7 @@ Frozen Python engine sidecar (capture, review, auth, sync, FlowBridge, | isolated subprocess mode in the same signed executable v openadapt-flow - record -> compile -> lint/certify -> replay -> halt/repair/teach + record -> compile -> qualify propose/accept -> lint/certify -> replay -> halt/repair/teach ``` The engine owns consent, OS permissions, recording and review, hosted @@ -122,7 +122,7 @@ exact runtime. |---|---| | `record`, `list`, `info` | Capture a session and inspect its metadata | | `scrub`, `review`, `approve`, `dismiss` | Drive the local review state machine; a dismissal keeps the raw data local | -| `compile`, `replay`, `run` | Call the bundled pinned Flow runtime on a capture or a bundle | +| `compile`, `qualify`, `replay`, `run` | Call the bundled Flow runtime on a capture or a bundle. `qualify --recording` fills application, environment, identity, and effect pins from the demo; missing pins HALT. | | `login`, `credential`, `rotate`, `push`, `report-break` | Authenticate to the control plane, check or renew the stored credential, push a bundle, report a halted run | | `backends`, `upload` | Inspect the legacy customer-owned storage adapters. Hosted uses governed `push`; customer-owned upload is paused behind a fail-closed gate. | | `storage`, `health`, `cleanup` | Inspect and maintain local storage | diff --git a/engine/cli.py b/engine/cli.py index 6727d8c..bd462c4 100644 --- a/engine/cli.py +++ b/engine/cli.py @@ -19,6 +19,7 @@ openadapt-desktop config openadapt-desktop capabilities [--json] openadapt-desktop doctor + openadapt-desktop qualify BUNDLE --recording REC [--accept] [--admit-local] """ from __future__ import annotations @@ -371,6 +372,37 @@ def cmd_compile(args: argparse.Namespace, engine: types.SimpleNamespace) -> None print(f" Bundle: {out}") +def cmd_qualify(args: argparse.Namespace, engine: types.SimpleNamespace) -> None: + """Fill production-shaped pins from a recording. Missing pins HALT.""" + + from engine.flow_bridge import FlowBridge, FlowNotAvailableError + + bundle = Path(args.bundle) + recording = Path(args.recording) + try: + if args.accept: + result = FlowBridge().qualify_from_demo( + bundle, + recording, + policy_pack=args.policy_pack, + admit_local=args.admit_local, + ) + else: + result = FlowBridge().qualify_propose( + bundle, + recording, + policy_pack=args.policy_pack, + ) + except FlowNotAvailableError as exc: + print(str(exc)) + sys.exit(1) + text = (result.stdout or result.stderr or "").strip() + if text: + print(text) + if not result.ok: + sys.exit(result.returncode or 1) + + def cmd_replay(args: argparse.Namespace, engine: types.SimpleNamespace) -> None: """Replay a bundle locally (delegates to openadapt-flow).""" from engine.flow_bridge import FlowBridge, FlowNotAvailableError @@ -797,6 +829,7 @@ def cmd_doctor(args: argparse.Namespace, engine: types.SimpleNamespace) -> None: "rotate": cmd_rotate, "push": cmd_push, "compile": cmd_compile, + "qualify": cmd_qualify, "replay": cmd_replay, "run": cmd_run, "report-break": cmd_report_break, @@ -877,6 +910,28 @@ def main(argv: list[str] | None = None) -> None: p.add_argument("recording", help="Recording directory") p.add_argument("--out", default=None, help="Output bundle directory") + p = subparsers.add_parser( + "qualify", + help="Fill qualification pins from the recording that produced a bundle", + ) + p.add_argument("bundle", help="Compiled workflow bundle directory") + p.add_argument("--recording", required=True, help="Recording directory") + p.add_argument( + "--policy-pack", + choices=("community", "cloud", "regulated"), + default="community", + ) + p.add_argument( + "--accept", + action="store_true", + help="Confirm every proposed pin. Refusing is the default HALT path.", + ) + p.add_argument( + "--admit-local", + action="store_true", + help="Sign a MockMed/local-dev admission that production trust maps refuse", + ) + # replay p = subparsers.add_parser("replay", help="Replay a flow bundle locally") p.add_argument("bundle", help="Bundle directory") diff --git a/engine/dispatch.py b/engine/dispatch.py index f02997a..cbec8c3 100644 --- a/engine/dispatch.py +++ b/engine/dispatch.py @@ -257,6 +257,7 @@ def _register(self) -> None: # qualification cockpit (canonical Flow graph/policy/manifests) "get_qualification": self.get_qualification, "initialize_qualification": self.initialize_qualification, + "propose_qualification_from_demo": self.propose_qualification_from_demo, "set_qualification_risk": self.set_qualification_risk, "arm_qualification_identity": self.arm_qualification_identity, "set_qualification_identity": self.set_qualification_identity, @@ -2112,6 +2113,42 @@ def get_qualification(self, **params: Any) -> dict: except Exception as exc: return {"ok": False, "workflow_id": workflow_id, "error": str(exc)} + def propose_qualification_from_demo(self, **params: Any) -> dict: + """Fill pins from the recording that produced this bundle.""" + + from engine.qualification import propose_qualification_from_demo + + workflow_id = str(params.get("workflow_id") or "") + recording_dir = params.get("recording_dir") + try: + if not recording_dir: + row = self.services.db.get_bundle(workflow_id) + capture_id = row.get("capture_id") if row else None + capture = ( + self.services.db.get_capture(str(capture_id)) + if capture_id + else None + ) + recording_dir = capture and ( + capture.get("capture_path") or capture.get("capture_dir") + ) + if not recording_dir: + raise ValueError( + "recording_dir is required when this bundle has no " + "retained capture" + ) + return propose_qualification_from_demo( + self._qualification_bundle_dir(workflow_id), + workflow_id=workflow_id, + recording_dir=Path(str(recording_dir)), + policy_pack=str(params.get("policy_pack") or "community"), + accept=bool(params.get("accept", True)), + admit_local=bool(params.get("admit_local", False)), + bundle_key=self._qualification_bundle_key(workflow_id), + ) + except Exception as exc: + return {"ok": False, "workflow_id": workflow_id, "error": str(exc)} + def initialize_qualification(self, **params: Any) -> dict: """Create Flow's versioned project for one explicit environment boundary.""" diff --git a/engine/flow_bridge.py b/engine/flow_bridge.py index 99fa835..57a05d0 100644 --- a/engine/flow_bridge.py +++ b/engine/flow_bridge.py @@ -718,6 +718,51 @@ def compile( ] return self._run(args, out_dir=out_dir, timeout=timeout) + def qualify_propose( + self, + bundle_dir: Path, + recording_dir: Path | None = None, + *, + policy_pack: str = "community", + out: Path | None = None, + timeout: float | None = None, + ) -> FlowResult: + """Draft qualification pins from a compiled demo. Missing pins HALT.""" + + args = ["qualify", "propose", str(bundle_dir), "--policy-pack", policy_pack] + if recording_dir is not None: + args += ["--recording", str(recording_dir)] + if out is not None: + args += ["--out", str(out)] + return self._run(args, out_dir=bundle_dir, timeout=timeout) + + def qualify_from_demo( + self, + bundle_dir: Path, + recording_dir: Path, + *, + policy_pack: str = "community", + admit_local: bool = False, + out: Path | None = None, + timeout: float | None = None, + ) -> FlowResult: + """Propose pins from the recording and accept them in one Flow call.""" + + args = [ + "qualify", + "from-demo", + str(bundle_dir), + "--recording", + str(recording_dir), + "--policy-pack", + policy_pack, + ] + if admit_local: + args.append("--admit-local") + if out is not None: + args += ["--out", str(out)] + return self._run(args, out_dir=bundle_dir, timeout=timeout) + def replay( self, bundle_dir: Path, @@ -833,9 +878,7 @@ def supports_command(self, command: str) -> bool: try: result = self._run([command, "--help"], timeout=15) if command == "push": - self._push_json_support = result.ok and "--json" in ( - result.stdout or "" - ) + self._push_json_support = result.ok and "--json" in (result.stdout or "") return result.ok except Exception: if command == "push": diff --git a/engine/main.py b/engine/main.py index ec03311..5149412 100644 --- a/engine/main.py +++ b/engine/main.py @@ -64,6 +64,7 @@ def _print_embedded_flow_help() -> None: resume Resume from the last verified checkpoint lint Inspect bundle coverage certify Enforce a bundle safety policy + qualify Draft and confirm qualification pins from a demo console Open the local operator console Run a command with --help for its complete options.""" diff --git a/engine/qualification.py b/engine/qualification.py index bb48ada..d9cbe48 100644 --- a/engine/qualification.py +++ b/engine/qualification.py @@ -1029,6 +1029,81 @@ def initialize_qualification( ) +def propose_qualification_from_demo( + bundle_dir: Path, + *, + workflow_id: str, + recording_dir: Path, + policy_pack: str = "community", + accept: bool = True, + admit_local: bool = False, + policy_source: str = DEFAULT_QUALIFICATION_POLICY, + bundle_key: str | None = None, +) -> dict: + """Fill production-shaped pins from the recording. Missing pins HALT. + + Uses Flow's ``qualify propose`` / ``accept`` APIs when the installed + runtime has them. Older Flow builds get a clear update error instead of + a guessed environment. + """ + + try: + from openadapt_flow.qualification_proposal import ( + QualificationProposalError, + accept_proposal, + admit_local_dev, + propose_qualification, + save_accepted_bundle, + ) + except ImportError as exc: + raise QualificationError( + "This Desktop build needs OpenAdapt Flow with " + "`qualify propose`. Update Flow, then qualify this recording." + ) from exc + + workflow = _load(bundle_dir, key=bundle_key) + try: + proposal = propose_qualification( + workflow, + recording_dir=recording_dir, + policy_pack=policy_pack, + ) + except (ValueError, TypeError) as exc: + raise QualificationError(str(exc)) from exc + payload = proposal.model_dump(mode="json") + if proposal.status == "halted" or not accept: + return { + "ok": proposal.status != "halted", + "workflow_id": workflow_id, + "proposal": payload, + "error": proposal.halt_reason or "", + } + try: + accepted = accept_proposal(workflow, proposal, replace=True) + local = None + if admit_local: + local = admit_local_dev( + workflow, accepted, bundle_dir=bundle_dir + ) + save_accepted_bundle( + workflow, + bundle_dir, + proposal=accepted, + local_admission=local, + ) + except QualificationProposalError as exc: + raise QualificationError(str(exc)) from exc + inspected = inspect_bundle( + bundle_dir, + workflow_id=workflow_id, + policy_source=policy_source, + bundle_key=bundle_key, + ) + inspected["proposal"] = accepted.model_dump(mode="json") + inspected["ok"] = True + return inspected + + def set_action_risk( bundle_dir: Path, *, diff --git a/src/lib/engine.ts b/src/lib/engine.ts index 81d5c6c..438bcef 100644 --- a/src/lib/engine.ts +++ b/src/lib/engine.ts @@ -39,6 +39,7 @@ export const CMD = { TEACH_FIX: "teach_fix", GET_QUALIFICATION: "get_qualification", INITIALIZE_QUALIFICATION: "initialize_qualification", + PROPOSE_QUALIFICATION_FROM_DEMO: "propose_qualification_from_demo", SET_QUALIFICATION_RISK: "set_qualification_risk", ARM_QUALIFICATION_IDENTITY: "arm_qualification_identity", SET_QUALIFICATION_IDENTITY: "set_qualification_identity", diff --git a/src/screens/Qualification.test.tsx b/src/screens/Qualification.test.tsx index 29db5ae..6f45817 100644 --- a/src/screens/Qualification.test.tsx +++ b/src/screens/Qualification.test.tsx @@ -165,6 +165,35 @@ describe("Qualification effect requirements", () => { ); }); + it("fills pins from the recording that produced the bundle", async () => { + const draft = projectWithTiers({ review: 3, submit: 2 }); + draft.draft_environment = true; + mockedEngineInvoke + .mockResolvedValueOnce(draft) + .mockResolvedValueOnce({ + ...draft, + draft_environment: false, + migration_required: false, + }); + + render( {}} />); + fireEvent.click( + await screen.findByRole("button", { + name: "Fill pins from this recording", + }), + ); + + await waitFor(() => + expect(mockedEngineInvoke).toHaveBeenCalledWith( + CMD.PROPOSE_QUALIFICATION_FROM_DEMO, + expect.objectContaining({ + workflow_id: "wf-1", + accept: true, + }), + ), + ); + }); + it("saves the selected action's tier without carrying the prior action's value", async () => { mockedEngineInvoke .mockResolvedValueOnce(projectWithTiers({ review: 3, submit: 2 })) diff --git a/src/screens/Qualification.tsx b/src/screens/Qualification.tsx index 717c2ee..3fb21a5 100644 --- a/src/screens/Qualification.tsx +++ b/src/screens/Qualification.tsx @@ -399,6 +399,33 @@ export function Qualification({ } } + async function qualifyThisRecording() { + setBusy("propose"); + setError(""); + try { + const response = await engineInvoke( + CMD.PROPOSE_QUALIFICATION_FROM_DEMO, + { + workflow_id: workflowId, + policy_pack: "community", + accept: true, + }, + ); + if (!response.ok) { + setError( + response.error || + "A pin is missing. Flow HALTed instead of guessing.", + ); + return; + } + setProject(response); + } catch (reason) { + setError(String(reason)); + } finally { + setBusy(""); + } + } + async function setRisk(stepId: string, risk: QualificationRisk) { setBusy(stepId); setError(""); @@ -759,9 +786,22 @@ export function Qualification({ + +

+ Or type the pins by hand if you already know the application + version and environment identifier. +

diff --git a/tests/test_engine/test_cli.py b/tests/test_engine/test_cli.py index 5a76978..e40edb7 100644 --- a/tests/test_engine/test_cli.py +++ b/tests/test_engine/test_cli.py @@ -121,6 +121,27 @@ def test_doctor_checks_flow(self, cli_config: EngineConfig, capsys) -> None: captured = capsys.readouterr() assert "openadapt-flow (loop engine)" in captured.out + def test_qualify_propose_calls_flow( + self, cli_config: EngineConfig, tmp_path: Path, capsys + ) -> None: + from engine.flow_bridge import FlowResult + + rec = tmp_path / "rec" + bundle = tmp_path / "bundle" + rec.mkdir() + bundle.mkdir() + result = FlowResult( + ok=True, returncode=0, stdout='{"status":"draft"}', stderr="" + ) + with patch("engine.cli.EngineConfig", return_value=cli_config), patch( + "engine.flow_bridge.FlowBridge" + ) as bridge_cls: + bridge_cls.return_value.qualify_propose.return_value = result + main(["qualify", str(bundle), "--recording", str(rec)]) + captured = capsys.readouterr() + assert '"status":"draft"' in captured.out + bridge_cls.return_value.qualify_propose.assert_called_once() + def test_login_success(self, cli_config: EngineConfig, capsys) -> None: """login should dispatch to engine.auth.login and report the org.""" cred = {"kind": "ingest_token", "token": "t", "refresh_token": None, diff --git a/tests/test_engine/test_flow_bridge.py b/tests/test_engine/test_flow_bridge.py index cfdbc42..3560ec6 100644 --- a/tests/test_engine/test_flow_bridge.py +++ b/tests/test_engine/test_flow_bridge.py @@ -77,9 +77,21 @@ def test_compile_builds_args(self, tmp_path: Path, monkeypatch) -> None: assert "--name" in command assert command[command.index("--name") + 1] == "bundle" - def test_report_break_keeps_token_out_of_argv( - self, tmp_path: Path, monkeypatch - ) -> None: + def test_qualify_from_demo_builds_args(self, tmp_path: Path, monkeypatch) -> None: + monkeypatch.setattr("engine.flow_bridge.shutil.which", lambda _: "/usr/bin/openadapt-flow") + calls: list = [] + bridge = FlowBridge(runner=_runner(calls, stdout="{}")) + rec = tmp_path / "rec" + bundle = tmp_path / "bundle" + result = bridge.qualify_from_demo(bundle, rec, admit_local=True) + assert result.ok + command, _env = calls[0] + assert command[1:3] == ["qualify", "from-demo"] + assert "--recording" in command + assert "--admit-local" in command + assert "--policy-pack" in command + + def test_report_break_keeps_token_out_of_argv(self, tmp_path: Path, monkeypatch) -> None: monkeypatch.setattr("engine.flow_bridge.shutil.which", lambda _: "/usr/bin/openadapt-flow") calls: list = [] bridge = FlowBridge(runner=_runner(calls, stdout="Nothing emitted: no halt")) @@ -96,9 +108,7 @@ def test_report_break_keeps_token_out_of_argv( assert "--token" not in command assert env["OPENADAPT_INGEST_TOKEN"] == "secret-value" - def test_push_keeps_token_and_local_name_out_of_argv( - self, tmp_path: Path, monkeypatch - ) -> None: + def test_push_keeps_token_and_local_name_out_of_argv(self, tmp_path: Path, monkeypatch) -> None: monkeypatch.setattr("engine.flow_bridge.shutil.which", lambda _: "/usr/bin/openadapt-flow") calls: list = [] bridge = FlowBridge(runner=_runner(calls, stdout="--json\nok"))