feat: Qualify this recording fills Flow pins from the demo

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 |

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")

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."""
 

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":

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."""

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,
     *,

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",

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(<Qualification workflowId="wf-1" onBack={() => {}} />);
+    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 }))