Workflows: Add UI for taskmaestro workflows

Discover taskmaestro workflows under ~/.taskmaestro/workflows/ and surface
them as a "Workflows" collection in the project tree alongside Scripts.
Each workflow's task input schema is auto-generated by introspecting the
Pydantic models via a Python helper (rips.taskmaestro_helper) and rendered
as PDM field bindings - scalars for str/int/float/bool, and PdmPtrField
pickers for fields typed as rips.Case / rips.WellPath / rips.View.

The Run button on a RimWorkflow writes input.yaml from current bindings
(rips object references serialized as __resinsight_ref__ maps), spawns
the helper's run subcommand, and streams output to a modeless log dialog.
The helper resolves __resinsight_ref__ markers back to live rips objects
via the running ResInsight's gRPC server before driving taskmaestro.

GrpcInterface/Python/rips/taskmaestro_helper/__init__.py

@@ -0,0 +1,6 @@
+"""Bridge between ResInsight and the taskmaestro workflow library.
+
+ResInsight invokes this package as a subprocess to introspect workflow
+schemas and to execute workflows. The Python side exists so ResInsight
+itself does not need to parse YAML or know about Pydantic.
+"""

GrpcInterface/Python/rips/taskmaestro_helper/__main__.py

@@ -0,0 +1,34 @@
+"""Command-line entry point: `python -m rips.taskmaestro_helper <subcommand>`."""
+
+from __future__ import annotations
+
+import sys
+
+
+def main() -> int:
+    if len(sys.argv) < 2:
+        print(
+            "usage: python -m rips.taskmaestro_helper <introspect|run> ...",
+            file=sys.stderr,
+        )
+        return 2
+
+    subcommand = sys.argv[1]
+    rest = sys.argv[2:]
+
+    if subcommand == "introspect":
+        from . import introspect
+
+        return introspect.main(rest)
+
+    if subcommand == "run":
+        from . import run
+
+        return run.main(rest)
+
+    print(f"unknown subcommand: {subcommand}", file=sys.stderr)
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

GrpcInterface/Python/rips/taskmaestro_helper/introspect.py

@@ -0,0 +1,186 @@
+"""Introspect a taskmaestro workflow directory and emit a JSON schema.
+
+The schema is consumed by ResInsight to auto-generate property-editor
+fields for each task's config inputs. Only fields listed in the workflow's
+`config_fields` are inspected, so models that also reference non-Pydantic
+types (e.g. live `rips` objects passed via `depends_on`) do not break
+introspection.
+"""
+
+from __future__ import annotations
+
+import datetime
+import json
+import pathlib
+import sys
+from pathlib import Path
+from types import UnionType
+from typing import Annotated, Any, Union, get_args, get_origin
+
+from pydantic import TypeAdapter
+from pydantic.fields import FieldInfo
+
+from .refs import annotation_to_resinsight_type
+
+
+_SCALAR_TYPE_MAP: dict[type, str] = {
+    str: "string",
+    bool: "boolean",
+    int: "integer",
+    float: "number",
+}
+
+
+def _format_for(field_info: FieldInfo) -> str | None:
+    """Ask Pydantic for the JSON-schema `format` of a field, if any.
+
+    Pydantic stores e.g. `pydantic.DirectoryPath` (which is
+    `Annotated[Path, PathType('dir')]`) as `annotation=Path` and
+    `metadata=[PathType('dir')]`. We reassemble both before asking
+    TypeAdapter, so format markers like `directory-path` survive.
+    """
+    annotation = field_info.annotation
+    if field_info.metadata:
+        annotation = Annotated[tuple([annotation, *field_info.metadata])]
+    try:
+        return TypeAdapter(annotation).json_schema().get("format")
+    except Exception:
+        return None
+
+
+def _annotation_type(annotation: object) -> str:
+    """Map a Pydantic field annotation to a JSON-schema-style type label."""
+    if isinstance(annotation, type) and annotation in _SCALAR_TYPE_MAP:
+        return _SCALAR_TYPE_MAP[annotation]
+
+    # Optional[T] / T | None — peel the union and map the non-None arm
+    origin = get_origin(annotation)
+    if origin in (Union, UnionType):
+        non_none = [a for a in get_args(annotation) if a is not type(None)]
+        if len(non_none) == 1:
+            return _annotation_type(non_none[0])
+
+    if origin in (list, tuple, set, frozenset):
+        return "array"
+
+    if isinstance(annotation, type):
+        # Recognised rips object types are reported as "object" with a separate marker.
+        if annotation_to_resinsight_type(annotation) is not None:
+            return "object"
+
+    return "string"
+
+
+_SENTINEL = object()
+
+
+def _serialize_default(value: Any) -> Any:
+    """Convert a Python value into a JSON-friendly form for the schema's `default`."""
+    if isinstance(value, (datetime.date, datetime.datetime)):
+        return value.isoformat()
+    if isinstance(value, pathlib.PurePath):
+        return str(value)
+    return value
+
+
+def _field_schema(
+    field_name: str, field_info: FieldInfo, input_value: Any = _SENTINEL
+) -> dict[str, Any]:
+    entry: dict[str, Any] = {
+        "name": field_name,
+        "type": _annotation_type(field_info.annotation),
+        "required": field_info.is_required(),
+    }
+    if field_info.description:
+        entry["description"] = field_info.description
+
+    # Prefer the value the user already has in input.yaml; fall back to the
+    # Pydantic-side default so the UI is pre-filled in either case.
+    if input_value is not _SENTINEL and input_value is not None:
+        entry["default"] = _serialize_default(input_value)
+    elif not field_info.is_required():
+        default = field_info.get_default(call_default_factory=False)
+        if default is not None:
+            entry["default"] = _serialize_default(default)
+
+    fmt = _format_for(field_info)
+    if fmt is not None:
+        entry["format"] = fmt
+    ri_type = annotation_to_resinsight_type(field_info.annotation)
+    if ri_type is not None:
+        entry["resinsight_type"] = ri_type
+    return entry
+
+
+def collect_schema(workflow_dir: Path) -> dict[str, Any]:
+    """Load the workflow at `workflow_dir` and produce its UI schema."""
+    workflow_yaml = workflow_dir / "workflow.yaml"
+    input_yaml = workflow_dir / "input.yaml"
+    if not workflow_yaml.is_file():
+        raise FileNotFoundError(f"Missing workflow.yaml in {workflow_dir}")
+
+    # Workflow source files (e.g. pipeline.py) live next to workflow.yaml and
+    # are referenced as plain dotted paths in workflow.yaml — make them importable.
+    workflow_dir_str = str(workflow_dir)
+    if workflow_dir_str not in sys.path:
+        sys.path.insert(0, workflow_dir_str)
+
+    from taskmaestro.task import get_input_type
+    from taskmaestro.yaml_config import _load_workflow_only
+
+    wf, jc = _load_workflow_only(
+        workflow_yaml,
+        input_yaml if input_yaml.is_file() else None,
+    )
+
+    tasks: list[dict[str, Any]] = []
+    for task_name, task_cls in wf.topological_order():
+        config_field_names = wf.get_config_fields(task_name)
+        if not config_field_names:
+            continue
+
+        task_config = jc.get_config_for_task(task_name) if jc is not None else {}
+
+        input_type = get_input_type(task_cls)
+        config_fields: list[dict[str, Any]] = []
+        for field_name in sorted(config_field_names):
+            field_info = input_type.model_fields.get(field_name)
+            if field_info is None:
+                config_fields.append(
+                    {
+                        "name": field_name,
+                        "type": "string",
+                        "required": True,
+                        "error": "field not found in input model",
+                    }
+                )
+                continue
+            input_value = task_config.get(field_name, _SENTINEL)
+            config_fields.append(
+                _field_schema(field_name, field_info, input_value=input_value)
+            )
+
+        tasks.append({"name": task_name, "config_fields": config_fields})
+
+    return {
+        "name": wf.name,
+        "description": "",
+        "tasks": tasks,
+    }
+
+
+def main(argv: list[str]) -> int:
+    if len(argv) != 1:
+        print("usage: introspect <workflow_dir>", file=sys.stderr)
+        return 2
+    workflow_dir = Path(argv[0]).resolve()
+    try:
+        schema = collect_schema(workflow_dir)
+    except Exception as exc:
+        print(
+            json.dumps({"error": str(exc), "type": type(exc).__name__}), file=sys.stderr
+        )
+        return 1
+    json.dump(schema, sys.stdout, indent=2)
+    sys.stdout.write("\n")
+    return 0

GrpcInterface/Python/rips/taskmaestro_helper/refs.py

@@ -0,0 +1,56 @@
+"""Mapping between Pydantic field annotations and ResInsight object types.
+
+The whitelist below is the single source of truth for which `rips` types
+the helper recognises as object pickers. It is referenced both by
+`introspect` (to add the `resinsight_type` marker on field schemas) and
+by `run` (to resolve `__resinsight_ref__` values back to live `rips`
+objects).
+"""
+
+from __future__ import annotations
+
+# Maps rips class name -> resinsight_type label sent to the C++ side.
+# The label is also the C++ binding subclass selector (RimWorkflow{Label}Binding).
+# Rips classes live in `rips.generated.generated_classes` but are re-exported as
+# `rips.<Name>`; we match by class name only since the module is generated.
+RIPS_TYPE_WHITELIST: dict[str, str] = {
+    "Case": "EclipseCase",
+    "EclipseCase": "EclipseCase",
+    "WellPath": "WellPath",
+    "View": "View",
+    "EclipseView": "View",
+}
+
+_RIPS_MODULE_PREFIXES = ("rips.", "rips")
+
+# Marker used inside input.yaml to denote a ResInsight object reference.
+REF_MARKER = "__resinsight_ref__"
+
+
+def annotation_to_resinsight_type(annotation: object) -> str | None:
+    """Return the resinsight_type label for a Pydantic field annotation, or None.
+
+    Handles both bare rips classes and ObjectModel[rips.X] wrappers used by
+    the example workflow (e.g. `class GridCase(ObjectModel[rips.EclipseCase])`).
+    """
+    candidates: list[type] = []
+    if isinstance(annotation, type):
+        candidates.append(annotation)
+        # Walk the MRO for ObjectModel[T] parameterizations: pydantic stores the
+        # type argument on the intermediate class in __pydantic_generic_metadata__.
+        for cls in annotation.__mro__:
+            md = getattr(cls, "__pydantic_generic_metadata__", None)
+            if not md:
+                continue
+            for arg in md.get("args", ()) or ():
+                if isinstance(arg, type):
+                    candidates.append(arg)
+
+    for cls in candidates:
+        module = getattr(cls, "__module__", "") or ""
+        if not module.startswith(_RIPS_MODULE_PREFIXES):
+            continue
+        label = RIPS_TYPE_WHITELIST.get(cls.__name__)
+        if label is not None:
+            return label
+    return None

GrpcInterface/Python/rips/taskmaestro_helper/run.py

@@ -0,0 +1,126 @@
+"""Execute a taskmaestro workflow against a running ResInsight.
+
+The helper:
+1. Connects back to ResInsight via the gRPC port passed by the caller.
+2. Reads input.yaml and walks the dict, replacing every
+   `{__resinsight_ref__: <type>, ...id...}` map with an `ObjectModel`-style
+   `{"value": <live rips object>}` so taskmaestro's Pydantic validation
+   accepts it as a wrapped object.
+3. Builds the workflow via taskmaestro's loader and runs it.
+4. Streams progress events as newline-delimited JSON to stdout for the
+   ResInsight log dialog to render.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from typing import Any, Callable
+
+import yaml
+
+from .refs import REF_MARKER
+
+
+def _emit(event: str, **fields: Any) -> None:
+    sys.stdout.write(json.dumps({"event": event, **fields}) + "\n")
+    sys.stdout.flush()
+
+
+def resolve_refs(data: Any, resolver: Callable[[str, dict[str, Any]], Any]) -> Any:
+    """Recursively replace `{__resinsight_ref__: ..., ...}` maps in `data`.
+
+    The replacement is a `{"value": <resolved>}` dict, so Pydantic's
+    ObjectModel validator can accept it directly.
+    """
+    if isinstance(data, dict):
+        if REF_MARKER in data:
+            ri_type = data[REF_MARKER]
+            extras = {k: v for k, v in data.items() if k != REF_MARKER}
+            return {"value": resolver(ri_type, extras)}
+        return {k: resolve_refs(v, resolver) for k, v in data.items()}
+    if isinstance(data, list):
+        return [resolve_refs(v, resolver) for v in data]
+    return data
+
+
+def make_rips_resolver(rips_instance: Any) -> Callable[[str, dict[str, Any]], Any]:
+    """Resolve a `__resinsight_ref__` map to a live `rips` object."""
+
+    def resolve(ri_type: str, extras: dict[str, Any]) -> Any:
+        project = rips_instance.project
+        if ri_type == "EclipseCase":
+            return project.case(case_id=extras["case_id"])
+        if ri_type == "WellPath":
+            return project.well_path_by_name(extras["well_path_name"])
+        if ri_type == "View":
+            for v in project.views():
+                if getattr(v, "id", None) == extras["view_id"]:
+                    return v
+            raise LookupError(f"View id {extras['view_id']} not found")
+        raise LookupError(f"Unknown resinsight_type: {ri_type}")
+
+    return resolve
+
+
+def run_workflow(workflow_dir: Path, input_path: Path, grpc_port: int) -> int:
+    workflow_dir_str = str(workflow_dir)
+    if workflow_dir_str not in sys.path:
+        sys.path.insert(0, workflow_dir_str)
+
+    import rips
+
+    instance = rips.Instance.find(start_port=grpc_port, end_port=grpc_port + 1)
+    if instance is None:
+        _emit("error", message=f"Could not connect to ResInsight on port {grpc_port}")
+        return 1
+    _emit("connected", port=grpc_port)
+
+    raw_input = yaml.safe_load(input_path.read_text()) or {}
+    if not isinstance(raw_input, dict):
+        _emit("error", message="input.yaml must contain a mapping")
+        return 1
+
+    resolver = make_rips_resolver(instance)
+    try:
+        resolved = resolve_refs(raw_input, resolver)
+    except Exception as exc:
+        _emit("error", message=f"Failed to resolve references: {exc}")
+        return 1
+
+    from taskmaestro import EmptyConfig, ExecutionContext, Job, JobConfiguration, Runner
+    from taskmaestro.yaml_config import _load_workflow_only
+
+    workflow_yaml = workflow_dir / "workflow.yaml"
+    workflow, _ = _load_workflow_only(workflow_yaml)
+
+    job_config = JobConfiguration(resolved)
+    job: Job[Any] = Job(
+        workflow=workflow, config=EmptyConfig(), job_configuration=job_config
+    )
+
+    _emit("starting", workflow=workflow.name)
+    try:
+        result = Runner().run(job, ctx=ExecutionContext())
+    except Exception as exc:
+        _emit("error", message=str(exc))
+        return 1
+
+    _emit("finished", status=str(result.status), error=str(result.error or ""))
+    return 0 if result.error is None else 1
+
+
+def main(argv: list[str]) -> int:
+    import argparse
+
+    parser = argparse.ArgumentParser(
+        prog="run", description="Run a taskmaestro workflow."
+    )
+    parser.add_argument("workflow_dir", type=Path)
+    parser.add_argument("--input", type=Path, required=True)
+    parser.add_argument("--grpc-port", type=int, required=True)
+    args = parser.parse_args(argv)
+    return run_workflow(
+        args.workflow_dir.resolve(), args.input.resolve(), args.grpc_port
+    )

GrpcInterface/Python/rips/tests/test_taskmaestro_helper.py

@@ -0,0 +1,186 @@
+"""Unit tests for rips.taskmaestro_helper.
+
+These tests do not require a running ResInsight; they exercise the
+Pydantic introspection logic in isolation against a synthetic workflow
+written into a tmp_path so they stay self-contained.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+pytest.importorskip("taskmaestro")
+
+
+SYNTHETIC_PIPELINE = '''
+"""Synthetic workflow for taskmaestro_helper tests."""
+
+from __future__ import annotations
+
+import datetime
+import pathlib
+
+import pydantic
+from pydantic import BaseModel, Field
+
+from taskmaestro import EmptyConfig, ExecutionContext, Task
+
+
+class GreetInput(BaseModel):
+    name: str = Field(description="Person to greet", default="world")
+    times: int = Field(description="How many times", ge=1)
+    when: datetime.date = Field(description="Greet date", default=datetime.date(2024, 1, 1))
+    out_file: pathlib.Path = Field(description="Where to write greeting", default=pathlib.Path("/tmp/hi.txt"))
+    out_dir: pydantic.DirectoryPath = Field(description="Where to write logs")
+
+
+class GreetOutput(BaseModel):
+    message: str
+
+
+class Greet(Task[GreetInput, GreetOutput]):
+    name = "greet"
+
+    def run(self, input: GreetInput, ctx: ExecutionContext) -> GreetOutput:  # pragma: no cover
+        return GreetOutput(message=f"hi {input.name}" * input.times)
+
+
+class StartInput(BaseModel):
+    pass
+
+
+class StartOutput(BaseModel):
+    pass
+
+
+class Start(Task[StartInput, StartOutput]):
+    name = "start"
+
+    def run(self, input: StartInput, ctx: ExecutionContext) -> StartOutput:  # pragma: no cover
+        return StartOutput()
+'''
+
+
+SYNTHETIC_WORKFLOW_YAML = """workflow:
+  name: synthetic
+  tasks:
+    - task: pipeline.Start
+    - task: pipeline.Greet
+      depends_on: pipeline.Start
+      config_fields: [name, times, when, out_file, out_dir]
+"""
+
+
+SYNTHETIC_INPUT_YAML = """greet:
+  name: alice
+  times: 3
+  out_dir: /tmp
+"""
+
+
+@pytest.fixture
+def workflow_dir(tmp_path: Path) -> Path:
+    (tmp_path / "pipeline.py").write_text(SYNTHETIC_PIPELINE)
+    (tmp_path / "workflow.yaml").write_text(SYNTHETIC_WORKFLOW_YAML)
+    (tmp_path / "input.yaml").write_text(SYNTHETIC_INPUT_YAML)
+    return tmp_path
+
+
+def test_collect_schema_returns_workflow_name_and_tasks(workflow_dir: Path) -> None:
+    from rips.taskmaestro_helper.introspect import collect_schema
+
+    schema = collect_schema(workflow_dir)
+    assert schema["name"] == "synthetic"
+    task_names = [t["name"] for t in schema["tasks"]]
+    assert task_names == ["greet"]  # Start has no config_fields, so omitted
+
+
+def test_collect_schema_extracts_field_types_and_metadata(workflow_dir: Path) -> None:
+    from rips.taskmaestro_helper.introspect import collect_schema
+
+    schema = collect_schema(workflow_dir)
+    [greet] = schema["tasks"]
+    fields_by_name = {f["name"]: f for f in greet["config_fields"]}
+
+    # input.yaml has `name: alice`, so that wins over the Pydantic default "world"
+    assert fields_by_name["name"]["type"] == "string"
+    assert fields_by_name["name"]["default"] == "alice"
+    assert fields_by_name["name"]["description"] == "Person to greet"
+    assert fields_by_name["name"]["required"] is False
+
+    # input.yaml has `times: 3`; field has no Pydantic default but is required
+    assert fields_by_name["times"]["type"] == "integer"
+    assert fields_by_name["times"]["required"] is True
+    assert fields_by_name["times"]["default"] == 3
+    assert "resinsight_type" not in fields_by_name["times"]
+
+    # input.yaml has no `when:` block, so Pydantic default kicks in
+    assert fields_by_name["when"]["type"] == "string"
+    assert fields_by_name["when"]["format"] == "date"
+    assert fields_by_name["when"]["default"] == "2024-01-01"
+
+    # input.yaml has no out_file; Pydantic default flows through
+    assert fields_by_name["out_file"]["type"] == "string"
+    assert fields_by_name["out_file"]["format"] == "path"
+    assert fields_by_name["out_file"]["default"] == "/tmp/hi.txt"
+
+    # input.yaml has `out_dir: /tmp`; pre-fills even though field is required
+    assert fields_by_name["out_dir"]["type"] == "string"
+    assert fields_by_name["out_dir"]["format"] == "directory-path"
+    assert fields_by_name["out_dir"]["required"] is True
+    assert fields_by_name["out_dir"]["default"] == "/tmp"
+
+
+def test_resolve_refs_substitutes_object_model_value() -> None:
+    from rips.taskmaestro_helper.run import resolve_refs
+
+    sentinel_case = object()
+    sentinel_well = object()
+
+    def resolver(ri_type: str, extras: dict) -> object:
+        if ri_type == "EclipseCase" and extras == {"case_id": 0}:
+            return sentinel_case
+        if ri_type == "WellPath" and extras == {"well_path_name": "B-2H"}:
+            return sentinel_well
+        raise AssertionError(f"unexpected ref: {ri_type} {extras}")
+
+    raw = {
+        "load_model": {
+            "case": {"__resinsight_ref__": "EclipseCase", "case_id": 0},
+            "label": "x",
+        },
+        "load_well": {
+            "well": {"__resinsight_ref__": "WellPath", "well_path_name": "B-2H"},
+        },
+        "untouched": "stays",
+    }
+    out = resolve_refs(raw, resolver)
+    assert out["load_model"]["case"] == {"value": sentinel_case}
+    assert out["load_model"]["label"] == "x"
+    assert out["load_well"]["well"] == {"value": sentinel_well}
+    assert out["untouched"] == "stays"
+
+
+def test_main_emits_json_to_stdout(workflow_dir: Path) -> None:
+    rips_root = Path(__file__).resolve().parents[2]
+    proc = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "rips.taskmaestro_helper",
+            "introspect",
+            str(workflow_dir),
+        ],
+        cwd=rips_root,
+        capture_output=True,
+        text=True,
+        check=True,
+    )
+    payload = json.loads(proc.stdout)
+    assert payload["name"] == "synthetic"
+    assert {t["name"] for t in payload["tasks"]} == {"greet"}