#800 - Local workflow execution

.gitignore

@@ -29,3 +29,7 @@ docs/**/domain_config.js
 .ruff_cache
 
 .lycheecache
+
+.venv/
+build/
+*.egg-info

AGENTS.md

@@ -120,6 +120,8 @@ Entry point: `service/core/service.py`. Framework: FastAPI + Uvicorn + OpenTelem
 | `utils/job/` | `Task`, `FrontendJob`, `K8sObjectFactory`, `PodGroupTopologyBuilder` | Workflow execution framework. Task → K8s spec generation. Gang scheduling via PodGroup. Topology constraints. Backend job definitions. |
 | `utils/connectors/` | `ClusterConnector`, `PostgresConnector`, `RedisConnector` | K8s API wrapper, PostgreSQL operations, Redis job queue management. |
 | `utils/secret_manager/` | `SecretManager` | JWE-based secret encryption/decryption. MEK/UEK key management. |
+| `utils/standalone_executor.py` | `StandaloneExecutor`, `run_workflow_standalone` | Standalone Docker-based workflow execution. Runs workflow specs without Kubernetes by mapping tasks to `docker run` commands with volume mounts for data flow. Supports DAG scheduling, resume (`--from-step`), and GPU passthrough. |
+| `utils/compose_executor.py` | `ComposeExecutor`, `run_workflow_compose` | Docker Compose-based parallel workflow execution. Extends StandaloneExecutor with wave-parallel scheduling, `{{host:taskname}}` DNS resolution via shared Compose networks, and GPU support via deploy resource reservations. |
 | `utils/progress_check/` | — | Liveness/progress tracking for long-running services. |
 | `utils/metrics/` | — | Prometheus metrics collection and export. |
 
@@ -139,6 +141,8 @@ Entry point: `cli.py` → `main_parser.py` (argparse). Subcommand modules:
 | `login.py`                                                                                                     | Authentication                   |
 | `pool.py`, `resources.py`, `user.py`, `credential.py`, `access_token.py`, `bucket.py`, `task.py`, `version.py` | Supporting commands              |
 | `backend.py`                                                                                                   | Backend cluster management       |
+| `standalone.py`                                                                                                | Standalone workflow execution via Docker (`osmo standalone run`) |
+| `docker_compose.py`                                                                                            | Parallel workflow execution via Docker Compose (`osmo docker-compose run`) |
 
 Features: Tab completion (shtab), response formatting (`formatters.py`), spec editor (`editor.py`), PyInstaller packaging (`cli_builder.py`, `packaging/`).
 

cookbook/tutorials/BUILD

@@ -0,0 +1,5 @@
+filegroup(
+    name = "tutorial_specs",
+    srcs = glob(["*.yaml"]),
+    visibility = ["//src/utils/tests:__pkg__"],
+)

src/cli/BUILD

@@ -37,6 +37,8 @@ osmo_py_library(
         "dataset.py",
         "editor.py",
         "formatters.py",
+        "docker_compose.py",
+        "standalone.py",
         "login.py",
         "main_parser.py",
         "pool.py",
@@ -73,6 +75,8 @@ osmo_py_library(
         "//src/lib/utils:validation",
         "//src/lib/utils:version",
         "//src/lib/utils:workflow",
+        "//src/utils:compose_executor",
+        "//src/utils:standalone_executor",
     ],
 )
 

src/cli/docker_compose.py

@@ -0,0 +1,127 @@
+"""
+SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.  # pylint: disable=line-too-long
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+import argparse
+import os
+import sys
+
+import shtab
+
+from src.utils import compose_executor
+
+
+def setup_parser(parser: argparse._SubParsersAction):
+    """Register the 'docker-compose' subcommand for parallel workflow execution."""
+    dc_parser = parser.add_parser(
+        'docker-compose',
+        help='Run workflows using Docker Compose for parallel execution '
+             '(no Kubernetes cluster required).')
+    subparsers = dc_parser.add_subparsers(dest='command')
+    subparsers.required = True
+
+    run_parser = subparsers.add_parser(
+        'run',
+        help='Execute a workflow spec using Docker Compose for parallel task execution.')
+    run_parser.add_argument(
+        '-f', '--file',
+        required=True,
+        dest='workflow_file',
+        help='Path to the workflow YAML spec file.').complete = shtab.FILE
+    run_parser.add_argument(
+        '--work-dir',
+        dest='work_dir',
+        default=None,
+        help='Directory for task inputs/outputs and the generated docker-compose.yml. '
+             'Defaults to a temporary directory.')
+    run_parser.add_argument(
+        '--keep',
+        action='store_true',
+        default=False,
+        help='Keep the work directory after execution (always kept on failure).')
+    run_parser.add_argument(
+        '--compose-cmd',
+        dest='compose_cmd',
+        default='docker compose',
+        help='Docker Compose command to use (e.g. "docker-compose" for V1). '
+             'Default: "docker compose".')
+    run_parser.add_argument(
+        '--credential',
+        nargs='+',
+        default=[],
+        help='Map credential names to local directories. '
+             'Format: "<name>=<path>". The directory is bind-mounted read-only '
+             'into the container at the path declared in the spec. '
+             'Example: --credential hf-token=$HOME/.hf')
+    run_parser.add_argument(
+        '--set',
+        nargs='+',
+        default=[],
+        help='Override default-values in the workflow spec. '
+             'Format: "<field>=<value>". Values are cast as int or float if '
+             'applicable, otherwise kept as strings.')
+    run_parser.add_argument(
+        '--set-string',
+        dest='set_string',
+        nargs='+',
+        default=[],
+        help='Override default-values in the workflow spec, forcing string type. '
+             'Format: "<field>=<value>".')
+    run_parser.add_argument(
+        '--shm-size',
+        dest='shm_size',
+        default=None,
+        help='Shared memory size for GPU containers (e.g. 16g, 32g). '
+             'Defaults to 16g for tasks that request GPUs.')
+    run_parser.set_defaults(func=_run_compose)
+
+
+def _parse_credentials(raw_credentials: list[str]) -> dict[str, str]:
+    """Parse --credential name=path arguments into a dict."""
+    result: dict[str, str] = {}
+    for entry in raw_credentials:
+        if '=' not in entry:
+            raise ValueError(
+                f'--credential value "{entry}" is incorrectly formatted (expected name=/path)')
+        name, path = entry.split('=', 1)
+        if not os.path.isdir(path):
+            raise ValueError(
+                f'Credential path for "{name}" does not exist or is not a directory: {path}')
+        result[name] = path
+    return result
+
+
+def _run_compose(service_client, args: argparse.Namespace):
+    """Execute a workflow via Docker Compose using the parsed CLI arguments."""
+    try:
+        credentials = _parse_credentials(args.credential)
+        success = compose_executor.run_workflow_compose(
+            spec_path=args.workflow_file,
+            work_dir=args.work_dir,
+            keep_work_dir=args.keep,
+            compose_cmd=args.compose_cmd,
+            shm_size=args.shm_size,
+            set_variables=args.set,
+            set_string_variables=args.set_string,
+            credentials=credentials,
+        )
+    except (ValueError, FileNotFoundError, PermissionError) as error:
+        print(f'Error: {error}', file=sys.stderr)
+        sys.exit(1)
+
+    if not success:
+        sys.exit(1)

src/cli/main_parser.py

@@ -28,6 +28,8 @@
     credential,
     data,
     dataset,
+    docker_compose,
+    standalone,
     login,
     pool,
     profile,
@@ -55,7 +57,9 @@
     profile.setup_parser,
     pool.setup_parser,
     user.setup_parser,
-    config.setup_parser
+    config.setup_parser,
+    standalone.setup_parser,
+    docker_compose.setup_parser,
 )
 
 

src/cli/standalone.py

@@ -0,0 +1,140 @@
+"""
+SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.  # pylint: disable=line-too-long
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+"""
+
+import argparse
+import os
+import sys
+
+import shtab
+
+from src.utils import standalone_executor
+
+
+def setup_parser(parser: argparse._SubParsersAction):
+    """Register the 'standalone' subcommand and its nested 'run' action with the CLI argument parser."""
+    standalone_parser = parser.add_parser(
+        'standalone',
+        help='Run workflows in standalone mode using Docker containers (no Kubernetes cluster required).')
+    subparsers = standalone_parser.add_subparsers(dest='command')
+    subparsers.required = True
+
+    run_parser = subparsers.add_parser(
+        'run',
+        help='Execute a workflow spec in standalone mode using Docker containers.')
+    run_parser.add_argument(
+        '-f', '--file',
+        required=True,
+        dest='workflow_file',
+        help='Path to the workflow YAML spec file.').complete = shtab.FILE
+    run_parser.add_argument(
+        '--work-dir',
+        dest='work_dir',
+        default=None,
+        help='Directory for task inputs/outputs. Defaults to a temporary directory.')
+    run_parser.add_argument(
+        '--keep',
+        action='store_true',
+        default=False,
+        help='Keep the work directory after execution (always kept on failure).')
+    run_parser.add_argument(
+        '--docker',
+        dest='docker_cmd',
+        default='docker',
+        help='Docker-compatible command to use (e.g. podman). Default: docker.')
+    run_parser.add_argument(
+        '--resume',
+        action='store_true',
+        default=False,
+        help='Resume a previous run, skipping tasks that already completed successfully. '
+             'Requires --work-dir pointing to the previous run directory.')
+    run_parser.add_argument(
+        '--from-step',
+        dest='from_step',
+        default=None,
+        help='Resume from a specific task, re-running it and all downstream tasks. '
+             'Tasks upstream of the specified step are skipped if they completed '
+             'successfully. Requires --work-dir pointing to the previous run directory.')
+    run_parser.add_argument(
+        '--credential',
+        nargs='+',
+        default=[],
+        help='Map credential names to local directories. '
+             'Format: "<name>=<path>". The directory is bind-mounted read-only '
+             'into the container at the path declared in the spec. '
+             'Example: --credential hf-token=$HOME/.hf')
+    run_parser.add_argument(
+        '--set',
+        nargs='+',
+        default=[],
+        help='Override default-values in the workflow spec. '
+             'Format: "<field>=<value>". Values are cast as int or float if '
+             'applicable, otherwise kept as strings.')
+    run_parser.add_argument(
+        '--set-string',
+        dest='set_string',
+        nargs='+',
+        default=[],
+        help='Override default-values in the workflow spec, forcing string type. '
+             'Format: "<field>=<value>".')
+    run_parser.add_argument(
+        '--shm-size',
+        dest='shm_size',
+        default=None,
+        help='Shared memory size for GPU containers (e.g. 16g, 32g). '
+             'Defaults to 16g for tasks that request GPUs. '
+             'PyTorch DataLoader workers require large shared memory.')
+    run_parser.set_defaults(func=_run_standalone)
+
+
+def _parse_credentials(raw_credentials: list[str]) -> dict[str, str]:
+    """Parse --credential name=path arguments into a dict."""
+    result: dict[str, str] = {}
+    for entry in raw_credentials:
+        if '=' not in entry:
+            raise ValueError(
+                f'--credential value "{entry}" is incorrectly formatted (expected name=/path)')
+        name, path = entry.split('=', 1)
+        if not os.path.isdir(path):
+            raise ValueError(
+                f'Credential path for "{name}" does not exist or is not a directory: {path}')
+        result[name] = path
+    return result
+
+
+def _run_standalone(service_client, args: argparse.Namespace):
+    """Execute a workflow in standalone mode via Docker using the parsed CLI arguments."""
+    try:
+        credentials = _parse_credentials(args.credential)
+        success = standalone_executor.run_workflow_standalone(
+            spec_path=args.workflow_file,
+            work_dir=args.work_dir,
+            keep_work_dir=args.keep,
+            resume=args.resume,
+            from_step=args.from_step,
+            docker_cmd=args.docker_cmd,
+            shm_size=args.shm_size,
+            set_variables=args.set,
+            set_string_variables=args.set_string,
+            credentials=credentials,
+        )
+    except (ValueError, FileNotFoundError, PermissionError) as error:
+        print(f'Error: {error}', file=sys.stderr)
+        sys.exit(1)
+
+    if not success:
+        sys.exit(1)

src/utils/BUILD

@@ -126,3 +126,26 @@ osmo_py_library(
     ],
     visibility = ["//visibility:public"],
 )
+
+osmo_py_library(
+    name = "standalone_executor",
+    srcs = ["standalone_executor.py"],
+    deps = [
+        requirement("jinja2"),
+        requirement("pyyaml"),
+        "//src/lib/utils:workflow",
+        "//src/utils/job",
+    ],
+    visibility = ["//visibility:public"],
+)
+
+osmo_py_library(
+    name = "compose_executor",
+    srcs = ["compose_executor.py"],
+    deps = [
+        requirement("pyyaml"),
+        "//src/utils:standalone_executor",
+        "//src/utils/job",
+    ],
+    visibility = ["//visibility:public"],
+)