feat(qualification): capture local institutional judgment cases

docs/DECISION_PORTAL.md

@@ -58,6 +58,31 @@ commit recorded in
 They use synthetic data and a public OpenEMR reference frame. They are not a
 customer run and do not claim that the complete workflow reached `VERIFIED`.
 
+## Add a human decision during qualification
+
+A workflow does not need a separate policy author or manual JSON edit. In the
+Desktop qualification cockpit, an operator can add a decision before an
+existing workflow step. The form records:
+
+- one reviewed question;
+- the roles that can answer it;
+- two to four finite answers;
+- the exact qualified successor for each answer;
+- optional local evidence that an answer requires;
+- the answer expiry; and
+- the live application check that must still pass before execution continues.
+
+Desktop sends these fields to Flow's canonical
+`openadapt.business-decision/v1` contract. Flow derives the guarded graph
+transitions. Saving the change advances the qualification revision and
+invalidates the prior certification. The qualification cases must pass again
+before production use.
+
+This decision is optional. A workflow that has no institutional-knowledge
+branch does not need one. One signed answer selects one qualified branch for
+one run. It does not prove the business effect and it does not change the
+workflow policy for later runs.
+
 ## Network boundary
 
 | Setting | Default | Meaning |

engine/dispatch.py

@@ -239,6 +239,8 @@ def _register(self) -> None:
             "bind_qualification_effect": self.bind_qualification_effect,
             "set_qualification_effect_verification": (self.set_qualification_effect_verification),
             "set_qualification_minimum_effect_tier": (self.set_qualification_minimum_effect_tier),
+            "author_qualification_business_decision": (self.author_qualification_business_decision),
+            "set_qualification_judgment_cases": (self.set_qualification_judgment_cases),
             "add_qualification_case": self.add_qualification_case,
             "run_qualification_case": self.run_qualification_case,
             "import_qualification_results": self.import_qualification_results,
@@ -1963,6 +1965,101 @@ def set_qualification_minimum_effect_tier(self, **params: Any) -> dict:
         except Exception as exc:
             return {"ok": False, "workflow_id": workflow_id, "error": str(exc)}
 
+    def author_qualification_business_decision(self, **params: Any) -> dict:
+        """Add or update one Flow-owned typed decision from form fields."""
+
+        from engine.qualification import (
+            DEFAULT_QUALIFICATION_POLICY,
+            author_business_decision,
+        )
+
+        workflow_id = str(params.get("workflow_id") or "")
+        policy = str(params.get("policy") or DEFAULT_QUALIFICATION_POLICY)
+        raw_roles = params.get("authorized_roles") or []
+        raw_options = params.get("options") or []
+        raw_requirements = params.get("evidence_requirements") or []
+        try:
+            if not isinstance(raw_roles, list):
+                raise ValueError("authorized_roles must be a list")
+            if not isinstance(raw_options, list) or not all(
+                isinstance(item, dict) for item in raw_options
+            ):
+                raise ValueError("options must be a list of decision option objects")
+            if not isinstance(raw_requirements, list) or not all(
+                isinstance(item, dict) for item in raw_requirements
+            ):
+                raise ValueError("evidence_requirements must be a list of evidence objects")
+            bundle = self._qualification_bundle_dir(workflow_id)
+            result = author_business_decision(
+                bundle,
+                workflow_id=workflow_id,
+                graph_id=str(params.get("graph_id") or ""),
+                state_id=str(params.get("state_id") or ""),
+                question=str(params.get("question") or ""),
+                authorized_roles=[str(item) for item in raw_roles],
+                output_param=str(params.get("output_param") or ""),
+                options=raw_options,
+                evidence_requirements=raw_requirements,
+                expires_after_s=int(params.get("expires_after_s", 3600)),
+                revalidation_kind=str(params.get("revalidation_kind") or ""),
+                revalidation_text=(
+                    str(params["revalidation_text"])
+                    if params.get("revalidation_text") is not None
+                    else None
+                ),
+                revalidation_state_id=(
+                    str(params["revalidation_state_id"])
+                    if params.get("revalidation_state_id") is not None
+                    else None
+                ),
+                insert_before_state_id=(
+                    str(params["insert_before_state_id"])
+                    if params.get("insert_before_state_id") is not None
+                    else None
+                ),
+                policy_source=policy,
+                bundle_key=self._qualification_bundle_key(workflow_id),
+            )
+            self.services.db.update_bundle(workflow_id, status="qualification_pending")
+            return result
+        except Exception as exc:
+            return {"ok": False, "workflow_id": workflow_id, "error": str(exc)}
+
+    def set_qualification_judgment_cases(self, **params: Any) -> dict:
+        """Save Flow-owned local judgment cases. No runtime decision is created."""
+
+        from engine.qualification import (
+            DEFAULT_QUALIFICATION_POLICY,
+            set_judgment_cases,
+        )
+
+        workflow_id = str(params.get("workflow_id") or "")
+        policy = str(params.get("policy") or DEFAULT_QUALIFICATION_POLICY)
+        raw_schemas = params.get("schemas") or []
+        raw_cases = params.get("cases") or []
+        try:
+            if not isinstance(raw_schemas, list) or not all(
+                isinstance(item, dict) for item in raw_schemas
+            ):
+                raise ValueError("schemas must be a list of local fact-schema objects")
+            if not isinstance(raw_cases, list) or not all(
+                isinstance(item, dict) for item in raw_cases
+            ):
+                raise ValueError("cases must be a list of local judgment-case objects")
+            bundle = self._qualification_bundle_dir(workflow_id)
+            result = set_judgment_cases(
+                bundle,
+                workflow_id=workflow_id,
+                schemas=raw_schemas,
+                cases=raw_cases,
+                policy_source=policy,
+                bundle_key=self._qualification_bundle_key(workflow_id),
+            )
+            self.services.db.update_bundle(workflow_id, status="qualification_pending")
+            return result
+        except Exception as exc:
+            return {"ok": False, "workflow_id": workflow_id, "error": str(exc)}
+
     def add_qualification_case(self, **params: Any) -> dict:
         """Add a typed case and keep its optional parameter fixture local."""