feat: emit qualified entity decision tasks v2

openadapt_flow/__main__.py

@@ -2054,17 +2054,22 @@ def _cmd_qualify(args: argparse.Namespace) -> int:
         QualificationCaseKind,
         QualificationCaseResult,
         QualificationOutcome,
+        QualifiedEntityLabel,
         RequalificationCondition,
         VerificationTier,
         add_case,
         add_requalification_condition,
         certify_project,
+        entity_label_options,
         init_project,
+        list_entity_labels,
         project_schema,
         record_case_results,
+        remove_entity_label,
         save_qualified_workflow,
         set_action_classification,
         set_effect_policy,
+        set_entity_label,
         set_identity_policy,
         set_trusted_runner_key,
     )
@@ -2076,6 +2081,42 @@ def _cmd_qualify(args: argparse.Namespace) -> int:
 
     workflow = _qualification_workflow(args)
 
+    if verb == "label":
+        if args.label_cmd == "list":
+            print(
+                json.dumps(
+                    [
+                        label.model_dump(mode="json")
+                        for label in list_entity_labels(workflow)
+                    ],
+                    indent=2,
+                )
+            )
+            return 0
+        changed = False
+        if args.label_cmd == "set":
+            label = QualifiedEntityLabel(
+                step_id=args.step, label=args.label, fallback=args.fallback
+            )
+            assert workflow.qualification is not None
+            changed = workflow.qualification.entity_labels.get(args.step) != label
+            set_entity_label(workflow, label)
+        elif args.label_cmd == "remove":
+            assert workflow.qualification is not None
+            changed = args.step in workflow.qualification.entity_labels
+            remove_entity_label(workflow, args.step)
+        else:  # pragma: no cover - argparse requires a known command.
+            raise SystemExit(f"unknown qualification label command {args.label_cmd!r}")
+        save_qualified_workflow(workflow, args.bundle)
+        print(workflow.qualification.model_dump_json(indent=2))
+        if changed:
+            print(
+                "Certification invalidated. Run `openadapt-flow qualify certify "
+                "<bundle> --evidence-root <path>` before production V2 tasks.",
+                file=sys.stderr,
+            )
+        return 0
+
     if verb == "init":
         environment = EnvironmentBoundary(
             target_kind=args.target,
@@ -2112,6 +2153,7 @@ def _cmd_qualify(args: argparse.Namespace) -> int:
                     else None
                 ),
                 "report": report.model_dump(mode="json"),
+                "entity_label_options": entity_label_options(),
             }
             print(json.dumps(payload, indent=2, sort_keys=True))
         else:
@@ -3876,6 +3918,37 @@ def build_parser() -> argparse.ArgumentParser:
     q = qsub.add_parser("schema", help="Print the qualification-project JSON Schema")
     q.set_defaults(func=_cmd_qualify)
 
+    q = qsub.add_parser(
+        "label",
+        help="Set, remove, or list qualification-owned entity labels",
+    )
+    from openadapt_flow.qualification import REMOTE_SAFE_ENTITY_LABELS
+
+    label_sub = q.add_subparsers(dest="label_cmd", required=True)
+    label = label_sub.add_parser("set", help="Set a label for one qualified step")
+    label.add_argument("bundle", help="Workflow bundle directory")
+    label.add_argument("--step", required=True, help="Exact qualified workflow step ID")
+    label.add_argument(
+        "--label",
+        required=True,
+        choices=REMOTE_SAFE_ENTITY_LABELS,
+        help="Qualification-approved class label, for example: insurance claim",
+    )
+    label.add_argument(
+        "--fallback",
+        choices=("record", "item"),
+        required=True,
+        help="Neutral fallback if a consumer cannot render the class label",
+    )
+    label.set_defaults(func=_cmd_qualify)
+    label = label_sub.add_parser("remove", help="Remove a step entity label")
+    label.add_argument("bundle", help="Workflow bundle directory")
+    label.add_argument("--step", required=True, help="Exact qualified workflow step ID")
+    label.set_defaults(func=_cmd_qualify)
+    label = label_sub.add_parser("list", help="List qualification-owned entity labels")
+    label.add_argument("bundle", help="Workflow bundle directory")
+    label.set_defaults(func=_cmd_qualify)
+
     q = qsub.add_parser("init", help="Initialize a bundle's qualification project")
     q.add_argument("bundle", help="Workflow bundle directory")
     q.add_argument(

openadapt_flow/console/human_decisions.py

@@ -33,8 +33,10 @@
 
 from openadapt_types import (
     HUMAN_DECISION_TASK_SCHEMA,
+    HUMAN_DECISION_TASK_V2_SCHEMA,
     HumanDecisionReceiptV1,
     HumanDecisionTaskV1,
+    HumanDecisionTaskV2,
 )
 from pydantic import BaseModel, ConfigDict, Field
 
@@ -232,7 +234,7 @@ class RemoteDecisionProjection(BaseModel):
     schema_version: Literal["openadapt.remote-decision-projection/v1"] = (
         "openadapt.remote-decision-projection/v1"
     )
-    task: HumanDecisionTaskV1
+    task: HumanDecisionTaskV1 | HumanDecisionTaskV2
     task_digest: str = Field(pattern=r"^sha256:[0-9a-f]{64}$")
     phase: Literal["paused"] = "paused"
     event_sequence: int = Field(ge=1)
@@ -389,6 +391,104 @@ def _remote_delivery_tier(
         raise AttendedActionRefused(str(exc)) from exc
 
 
+def _qualified_entity_v2_fields(
+    run_dir: Path,
+    *,
+    step_id: Optional[str],
+    capability: Any,
+) -> Optional[dict[str, Any]]:
+    """Read a V2 entity only from the current integrity-sealed project.
+
+    This deliberately reads no screenshots, OCR, accessibility observation,
+    parameters, application name, or model output.  Any missing, unreadable,
+    or mismatched capability/bundle binding falls back to V1. This gate
+    requires both bundle integrity and current qualification certification.
+    """
+
+    if step_id is None or step_id != capability.step_id:
+        return None
+    try:
+        from openadapt_flow.runtime.durable.checkpoint import CheckpointStore
+        from openadapt_flow.runtime.durable.program_checkpoint import bundle_version
+
+        manifest = CheckpointStore(run_dir).read_manifest()
+        if manifest is None:
+            return None
+        if bundle_version(manifest.bundle_dir) != capability.bundle_version:
+            return None
+        workflow, _ = data.load_workflow_safe(Path(manifest.bundle_dir))
+        project = workflow.qualification if workflow is not None else None
+        certification = project.last_certification if project is not None else None
+        if (
+            workflow is None
+            or project is None
+            or certification is None
+            or not certification.passed
+        ):
+            return None
+        from openadapt_flow.qualification import (
+            current_certification_matches,
+            entity_label_options,
+            workflow_contract_sha256,
+        )
+
+        # Certification is the existing authority for qualification approval.
+        # Check its direct bindings before asking it to independently recompute
+        # the signed-case and policy decision from this exact sealed workflow.
+        if (
+            certification.project_revision != project.revision
+            or certification.project_contract_sha256 != project.contract_sha256()
+            or certification.workflow_contract_sha256
+            != workflow_contract_sha256(workflow)
+            or certification.policy_contract_sha256 is None
+            or not current_certification_matches(
+                workflow,
+                policy_contract_digest=certification.policy_contract_sha256,
+            )
+        ):
+            return None
+        entity = project.entity_labels.get(step_id)
+        if (
+            entity is None
+            or {
+                "label": entity.label,
+                "fallback": entity.fallback,
+            }
+            not in entity_label_options()
+        ):
+            return None
+        return {
+            "qualification_project_id": project.project_id,
+            "qualification_revision_id": f"qualification_revision_{project.revision}",
+            "qualification_contract_digest": "sha256:" + project.contract_sha256(),
+            "qualification_step_id": step_id,
+            "entity": entity.model_dump(mode="json", exclude={"step_id"}),
+        }
+    except (OSError, ValueError, TypeError, AttendedActionRefused):
+        return None
+
+
+def _peer_negotiated_v2(deployment: Optional[DeploymentConfig]) -> bool:
+    """Return true only for an explicit authenticated-peer V2 declaration."""
+
+    return bool(
+        deployment is not None
+        and deployment.human_decisions.remote.enabled
+        and HUMAN_DECISION_TASK_V2_SCHEMA
+        in deployment.human_decisions.remote.peer_task_schemas
+    )
+
+
+def _task_model(
+    task: dict[str, Any],
+) -> type[HumanDecisionTaskV1 | HumanDecisionTaskV2]:
+    return (
+        HumanDecisionTaskV2
+        if task.get("schema_version") == HUMAN_DECISION_TASK_V2_SCHEMA
+        else HumanDecisionTaskV1
+    )
+
+
 def _task_and_presentation(
     run_dir: Path,
     item: AttentionItem,
@@ -540,8 +640,22 @@ def _task_and_presentation(
         ),
         "signature_algorithm": "hmac-sha256",
     }
-    task = AttendedActionStore(run_dir).seal_human_decision_task(unsigned)
-    task_digest = HumanDecisionTaskV1.model_validate(task).digest
+    qualified_v2 = (
+        _qualified_entity_v2_fields(
+            run_dir,
+            step_id=failed.step_id if failed is not None else None,
+            capability=capability,
+        )
+        if _peer_negotiated_v2(deployment)
+        else None
+    )
+    if qualified_v2 is not None:
+        unsigned.update(qualified_v2)
+        unsigned["schema_version"] = HUMAN_DECISION_TASK_V2_SCHEMA
+        task = AttendedActionStore(run_dir).seal_human_decision_task_v2(unsigned)
+    else:
+        task = AttendedActionStore(run_dir).seal_human_decision_task(unsigned)
+    task_digest = _task_model(task).model_validate(task).digest
     return task, task_digest, presentation
 
 
@@ -605,7 +719,7 @@ def portable_remote_decision_task(
         raise AttendedActionRefused(
             "the run has no current signed human decision task or pause capability"
         )
-    task = HumanDecisionTaskV1.model_validate(task_raw)
+    task = _task_model(task_raw).model_validate(task_raw)
     # The rest of `presentation` -- the screenshot artifact ids, the composed
     # question, the gated control label inside `halt` -- is still discarded.
     # Only the closed-vocabulary re-projection of `halt` may cross, and only at

openadapt_flow/deployment.py

@@ -446,6 +446,14 @@ class RemoteHumanDecisionConfig(BaseModel):
     context_tier: Literal["remote_closed_context", "remote_identifiers"] = (
         "remote_closed_context"
     )
+    #: Exact schemas the authenticated remote peer advertised for this
+    #: deployment.  Empty means no V2 negotiation occurred, so Flow emits V1.
+    peer_task_schemas: list[
+        Literal[
+            "openadapt.human-decision-task/v1",
+            "openadapt.human-decision-task/v2",
+        ]
+    ] = Field(default_factory=list)
 
     @model_validator(mode="after")
     def _require_exact_remote_scope(self) -> "RemoteHumanDecisionConfig":
@@ -454,6 +462,8 @@ def _require_exact_remote_scope(self) -> "RemoteHumanDecisionConfig":
                 "human_decisions.remote.enabled requires exact tenant_id and "
                 "runner_id bindings"
             )
+        if len(self.peer_task_schemas) != len(set(self.peer_task_schemas)):
+            raise ValueError("human decision peer task schemas must be unique")
         return self