Add reusable dry-run execution controls

python-tools/app/discovery.py

@@ -29,6 +29,7 @@ class Task:
     inputs: tuple["InputField", ...]
     wait_for_result: bool = False
     download_artifacts: bool = False
+    execution: "ExecutionMode | None" = None
 
 
 @dataclass(frozen=True)
@@ -64,6 +65,15 @@ class InputField:
     sensitive: bool = False
 
 
+@dataclass(frozen=True)
+class ExecutionMode:
+    """A task-declared report/apply mode exposed as a generated choice field."""
+
+    field: InputField
+    dry_run_value: str
+    execute_value: str
+
+
 @dataclass(frozen=True)
 class Group:
     id: str
@@ -162,9 +172,14 @@ class TaskCatalog:
                     path=path,
                     display_name=display_name(path.stem),
                     description=read_description(path),
-                    inputs=declaration.inputs if declaration else (),
+                    inputs=(
+                        (*declaration.inputs, declaration.execution.field)
+                        if declaration and declaration.execution
+                        else declaration.inputs if declaration else ()
+                    ),
                     wait_for_result=declaration.wait_for_result if declaration else False,
                     download_artifacts=declaration.download_artifacts if declaration else False,
+                    execution=declaration.execution if declaration else None,
                 )
             )
         return tasks
@@ -287,6 +302,7 @@ class TaskDeclaration:
     inputs: tuple[InputField, ...]
     wait_for_result: bool
     download_artifacts: bool
+    execution: ExecutionMode | None
 
 
 def parse_task_declaration(
@@ -296,7 +312,7 @@ def parse_task_declaration(
 ) -> TaskDeclaration:
     if (
         not isinstance(declaration, dict)
-        or set(declaration) - {"inputs", "wait_for_result", "download_artifacts"}
+        or set(declaration) - {"inputs", "wait_for_result", "download_artifacts", "execution"}
     ):
         raise ValueError(f"{manifest_path} task {filename} has an invalid declaration.")
     raw_inputs = declaration.get("inputs", [])
@@ -306,6 +322,7 @@ def parse_task_declaration(
     names = [field.name for field in fields]
     if len(names) != len(set(names)):
         raise ValueError(f"{manifest_path} task {filename} has duplicate input names.")
+    execution = parse_execution_mode(manifest_path, filename, declaration.get("execution"), set(names))
     wait_for_result = declaration.get("wait_for_result", False)
     if not isinstance(wait_for_result, bool):
         raise ValueError(f"{manifest_path} task {filename} wait_for_result must be true or false.")
@@ -318,6 +335,52 @@ def parse_task_declaration(
         inputs=fields,
         wait_for_result=wait_for_result,
         download_artifacts=download_artifacts,
+        execution=execution,
+    )
+
+
+def parse_execution_mode(
+    manifest_path: Path,
+    filename: str,
+    raw: Any,
+    input_names: set[str],
+) -> ExecutionMode | None:
+    if raw is None:
+        return None
+    required_keys = {
+        "field", "label", "dry_run_value", "dry_run_label", "execute_value", "execute_label",
+    }
+    if not isinstance(raw, dict) or set(raw) != required_keys:
+        raise ValueError(f"{manifest_path} task {filename} has an invalid execution declaration.")
+    values = {key: raw[key] for key in required_keys}
+    if not all(isinstance(value, str) and value.strip() for value in values.values()):
+        raise ValueError(f"{manifest_path} task {filename} execution values must be non-empty text.")
+    field_name = values["field"]
+    if not FIELD_NAME.fullmatch(field_name) or field_name in input_names:
+        raise ValueError(f"{manifest_path} task {filename} has an invalid execution field.")
+    if values["dry_run_value"] == values["execute_value"]:
+        raise ValueError(f"{manifest_path} task {filename} execution values must differ.")
+    field = InputField(
+        name=field_name,
+        label=values["label"],
+        type="choice",
+        required=True,
+        default=values["dry_run_value"],
+        minimum=None,
+        maximum=None,
+        step=None,
+        pattern=None,
+        options=(
+            InputOption(values["dry_run_value"], values["dry_run_label"]),
+            InputOption(values["execute_value"], values["execute_label"]),
+        ),
+        accept=(),
+        maximum_bytes=None,
+    )
+    return ExecutionMode(
+        field=field,
+        dry_run_value=values["dry_run_value"],
+        execute_value=values["execute_value"],
     )