feat(qualification): capture local institutional judgment cases

engine/dispatch.py

@@ -240,6 +240,7 @@ def _register(self) -> None:
             "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,
@@ -2024,6 +2025,41 @@ def author_qualification_business_decision(self, **params: Any) -> dict:
         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."""
 

engine/qualification.py

@@ -123,6 +123,10 @@ def _flow_api() -> dict[str, Any]:
         "set_identity_policy": set_identity_policy,
         "set_minimum_effect_tier": set_minimum_effect_tier,
         "set_business_decision": getattr(flow_qualification, "set_business_decision", None),
+        "set_judgment_cases": getattr(flow_qualification, "set_judgment_cases", None),
+        "evaluate_judgment_case_qualification": getattr(
+            flow_qualification, "evaluate_judgment_case_qualification", None
+        ),
         "workflow_contract_sha256": workflow_contract_sha256,
     }
 
@@ -398,6 +402,7 @@ def _qualification_controls(workflow, graph: dict[str, Any]) -> dict[str, Any]:
         "parameters": parameters,
         "actions": actions,
         "business_decisions": _business_decision_controls(workflow),
+        "judgment_cases": _judgment_case_controls(workflow),
     }
 
 
@@ -524,6 +529,83 @@ def _business_decision_controls(workflow) -> dict[str, Any]:
     }
 
 
+def _judgment_case_controls(workflow) -> dict[str, Any]:
+    """Project Flow-owned local judgment cases without exposing their artifacts.
+
+    This capability is additive. Older embedded Flow builds retain all direct
+    decision authoring controls and return a precise upgrade state here.
+    """
+
+    api = _flow_api()
+    setter = api.get("set_judgment_cases")
+    evaluator = api.get("evaluate_judgment_case_qualification")
+    if setter is None or evaluator is None or workflow.qualification is None:
+        return {
+            "available": False,
+            "required_flow_capability": "qualification.set_judgment_cases",
+            "contexts": [],
+            "report": None,
+        }
+    try:
+        from openadapt_flow.ir import StateKind, lift_to_program
+    except (ImportError, AttributeError):
+        return {
+            "available": False,
+            "required_flow_capability": "qualification.set_judgment_cases",
+            "contexts": [],
+            "report": None,
+        }
+
+    project = workflow.qualification
+    schemas = {
+        (item.graph_id, item.state_id): item.fact_schema
+        for item in project.judgment_fact_schemas
+    }
+    cases = list(project.judgment_cases)
+    workflow_digest = api["workflow_contract_sha256"](workflow)
+    main_program = workflow.program or lift_to_program(workflow)
+    contexts = []
+    for graph_id, graph in [("__program__", main_program), *workflow.subflows.items()]:
+        if graph is None:
+            continue
+        for state_id, state in graph.states.items():
+            if state.kind is not StateKind.BUSINESS_DECISION or state.decision is None:
+                continue
+            schema = schemas.get((graph_id, state_id))
+            if schema is None:
+                continue
+            contexts.append(
+                {
+                    "decision": {
+                        "graph_id": graph_id,
+                        "state_id": state_id,
+                        "workflow_contract_sha256": workflow_digest,
+                        "decision_contract_sha256": state.decision.contract_sha256(),
+                    },
+                    "fact_schema": schema.model_dump(mode="json"),
+                    "fact_schema_sha256": schema.contract_sha256(),
+                    "options": [item.model_dump(mode="json") for item in state.decision.options],
+                    "authorized_roles": list(state.decision.authorized_roles),
+                    "cases": [
+                        item.model_dump(mode="json")
+                        for item in cases
+                        if item.decision.graph_id == graph_id
+                        and item.decision.state_id == state_id
+                    ],
+                }
+            )
+    try:
+        report = evaluator(workflow).model_dump(mode="json")
+    except (ValueError, TypeError) as exc:
+        raise QualificationError(f"Cannot evaluate local judgment cases: {exc}") from exc
+    return {
+        "available": True,
+        "required_flow_capability": "qualification.set_judgment_cases",
+        "contexts": contexts,
+        "report": report,
+    }
+
+
 def _capability_coverage(
     bundle_dir: Path,
     *,
@@ -1157,6 +1239,54 @@ def author_business_decision(
     )
 
 
+def set_judgment_cases(
+    bundle_dir: Path,
+    *,
+    workflow_id: str,
+    schemas: list[dict[str, Any]],
+    cases: list[dict[str, Any]],
+    policy_source: str = DEFAULT_QUALIFICATION_POLICY,
+    bundle_key: str | None = None,
+) -> dict:
+    """Persist only Flow-validated local judgment cases and fact schemas.
+
+    The caller supplies reviewed local references. This function never accepts
+    screenshot bytes, record values outside declared facts, a generated rule,
+    or a runtime answer.
+    """
+
+    api = _flow_api()
+    setter = api.get("set_judgment_cases")
+    if setter is None:
+        raise QualificationError(
+            "This Desktop build needs a Flow runtime with "
+            "qualification.set_judgment_cases before it can save judgment cases."
+        )
+    try:
+        from openadapt_flow.judgment_cases import (
+            JudgmentCaseV1,
+            JudgmentFactSchemaBindingV1,
+        )
+    except ImportError as exc:
+        raise QualificationError(
+            "The bundled Flow runtime does not include local judgment-case qualification."
+        ) from exc
+    try:
+        workflow = _load(bundle_dir, key=bundle_key)
+        exact_schemas = [JudgmentFactSchemaBindingV1.model_validate(item) for item in schemas]
+        exact_cases = [JudgmentCaseV1.model_validate(item) for item in cases]
+        setter(workflow, schemas=exact_schemas, cases=exact_cases)
+        _save(workflow, bundle_dir, key=bundle_key)
+    except (QualificationError, ValueError, TypeError) as exc:
+        raise QualificationError(str(exc)) from exc
+    return inspect_bundle(
+        bundle_dir,
+        workflow_id=workflow_id,
+        policy_source=policy_source,
+        bundle_key=bundle_key,
+    )
+
+
 def bind_action_effect(
     bundle_dir: Path,
     *,

src/lib/engine.ts

@@ -46,6 +46,7 @@ export const CMD = {
     "set_qualification_minimum_effect_tier",
   AUTHOR_QUALIFICATION_BUSINESS_DECISION:
     "author_qualification_business_decision",
+  SET_QUALIFICATION_JUDGMENT_CASES: "set_qualification_judgment_cases",
   ADD_QUALIFICATION_CASE: "add_qualification_case",
   RUN_QUALIFICATION_CASE: "run_qualification_case",
   IMPORT_QUALIFICATION_RESULTS: "import_qualification_results",

src/lib/types.ts

@@ -296,6 +296,105 @@ export interface QualificationBusinessDecisionControls {
   }[];
 }
 
+/**
+ * Local-only qualification evidence for an institutional-judgment case.
+ *
+ * A reference never contains a screenshot, a record value, or free text. The
+ * Desktop and Flow resolve it inside the customer boundary.
+ */
+export interface LocalEvidenceRefV1 {
+  relative_path: string;
+  sha256: string;
+  kind: "frame" | "recording" | "report" | "document" | "system_read";
+}
+
+export type JudgmentFactTypeV1 =
+  | "boolean"
+  | "integer"
+  | "number"
+  | "string"
+  | "enum";
+
+export interface JudgmentFactFieldV1 {
+  type: JudgmentFactTypeV1;
+  allowed_values?: string[];
+}
+
+export interface JudgmentFactSchemaV1 {
+  schema_version: "openadapt.judgment-fact-schema/v1";
+  fields: Record<string, JudgmentFactFieldV1>;
+}
+
+export interface JudgmentFactSchemaBindingV1 {
+  graph_id: string;
+  state_id: string;
+  fact_schema: JudgmentFactSchemaV1;
+}
+
+export interface JudgmentCaseProvenanceV1 {
+  source: string;
+  source_ref_sha256: string;
+  reviewer_role: string;
+  reviewer_principal_ref_sha256: string;
+}
+
+export interface JudgmentDecisionBindingV1 {
+  graph_id: string;
+  state_id: string;
+  workflow_contract_sha256: string;
+  decision_contract_sha256: string;
+}
+
+export type JudgmentDispositionV1 =
+  | "automatic_rule"
+  | "human_node"
+  | "more_evidence_required";
+
+/** The exact local qualification payload owned and validated by Flow. */
+export interface JudgmentCaseV1 {
+  id: string;
+  decision: JudgmentDecisionBindingV1;
+  fact_schema_sha256: string;
+  facts: Record<string, boolean | number | string>;
+  local_evidence: LocalEvidenceRefV1[];
+  review_note_ref?: LocalEvidenceRefV1 | null;
+  provenance: JudgmentCaseProvenanceV1;
+  disposition: JudgmentDispositionV1;
+  reviewed_rule_id?: string | null;
+  option_id?: string | null;
+  contrast_case_ids: string[];
+}
+
+/**
+ * Flow supplies this local read model. It is intentionally not a portable
+ * decision-task type and it cannot authorize a runtime answer.
+ */
+export interface JudgmentCaseCaptureContextV1 {
+  decision: JudgmentDecisionBindingV1;
+  fact_schema: JudgmentFactSchemaV1;
+  fact_schema_sha256: string;
+  options: { id: string; label: string }[];
+  authorized_roles: string[];
+  cases: JudgmentCaseV1[];
+}
+
+export interface JudgmentCaseQualificationReportV1 {
+  schema_version: "openadapt.judgment-case-report/v1";
+  workflow_contract_sha256: string;
+  passed: boolean;
+  case_count: number;
+  automatic_case_count: number;
+  retained_human_authority_count: number;
+  findings: { code: string; case_id?: string | null; message: string }[];
+}
+
+export interface QualificationJudgmentCaseControls {
+  available: boolean;
+  required_flow_capability: "qualification.set_judgment_cases";
+  contexts: JudgmentCaseCaptureContextV1[];
+  report: JudgmentCaseQualificationReportV1 | null;
+}
+
 export interface QualificationViolation {
   rule: string;
   step_id?: string | null;
@@ -432,6 +531,7 @@ export interface QualificationProject {
     parameters: QualificationParameter[];
     actions: Record<string, QualificationActionControls>;
     business_decisions: QualificationBusinessDecisionControls;
+    judgment_cases: QualificationJudgmentCaseControls;
   };
 }