Headless One-Shot Recipe Dispatch CLI

src/autoskillit/cli/fleet/AGENTS.md

@@ -6,8 +6,9 @@ Fleet campaign CLI subcommands for multi-issue dispatch orchestration.
 
 | File | Purpose |
 |------|---------|
-| `__init__.py` | Main module: `fleet_app` Cyclopts sub-app with `campaign`, `dispatch`, `list`, `status` commands |
+| `__init__.py` | Main module: `fleet_app` Cyclopts sub-app with `campaign`, `dispatch`, `list`, `run`, `status` commands |
 | `_fleet_display.py` | Status display: `_render_status_display()`, `_watch_loop()`, `_STATUS_COLUMNS`, `render_fleet_error()` |
 | `_fleet_lifecycle.py` | Thin wrapper delegating reap to `fleet._dispatch_reaper`; `_pick_resume_campaign()` |
 | `_fleet_preview.py` | Pre-launch dispatch preview: `_build_dispatch_recipe_table()`, `_print_dispatch_preview()` |
+| `_fleet_run.py` | `fleet_run` command body + `_fleet_run_error` JSON envelope helper (extracted from `__init__.py`) |
 | `_fleet_session.py` | `_launch_fleet_session()` — builds Claude interactive session for fleet campaigns |

src/autoskillit/cli/fleet/__init__.py

@@ -31,6 +31,7 @@
     _FLEET_DISPATCH_GREETINGS,
     _print_dispatch_preview,
 )
+from autoskillit.cli.fleet._fleet_run import fleet_run
 from autoskillit.cli.fleet._fleet_session import _launch_fleet_session
 from autoskillit.core import TerminalColumn, get_logger, is_feature_enabled
 
@@ -408,3 +409,6 @@ def fleet_status(
             return
 
         print(_render_terminal_table(columns, rows_list))
+
+
+fleet_run = fleet_app.command(name="run")(fleet_run)

src/autoskillit/cli/fleet/_fleet_run.py

@@ -0,0 +1,216 @@
+"""Headless one-shot fleet dispatch command."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import TYPE_CHECKING, Annotated, NoReturn
+
+from cyclopts import Parameter
+
+from autoskillit.core import get_logger, is_feature_enabled
+
+if TYPE_CHECKING:
+    from autoskillit.config import AutomationConfig
+    from autoskillit.core import CodingAgentBackend
+    from autoskillit.fleet import DispatchResult
+
+logger = get_logger(__name__)
+
+
+def _fleet_run_error(error: str, message: str, exit_code: int = 1) -> NoReturn:
+    envelope = {"success": False, "error": error, "user_visible_message": message}
+    print(json.dumps(envelope))
+    raise SystemExit(exit_code)
+
+
+async def _execute_fleet_run(
+    cfg: AutomationConfig,
+    recipe: str,
+    task: str,
+    ingredients: dict[str, str] | None,
+    timeout_sec: int | None,
+    dispatch_backend: CodingAgentBackend | None,
+    resume_session_id: str | None,
+    prior_dispatch_id: str | None,
+    disable_quota_guard: bool,
+) -> DispatchResult:
+    import functools
+
+    from autoskillit.core import detect_autoskillit_mcp_prefix
+    from autoskillit.fleet import _build_food_truck_prompt, execute_dispatch
+    from autoskillit.server import make_context
+
+    ctx = make_context(cfg, project_dir=Path.cwd())
+
+    effective_backend = dispatch_backend or ctx.backend
+    has_ufa = (
+        effective_backend.capabilities.has_unguarded_filesystem_access
+        if effective_backend
+        else False
+    )
+    prompt_builder = functools.partial(
+        _build_food_truck_prompt,
+        mcp_prefix=detect_autoskillit_mcp_prefix(),
+        has_unguarded_filesystem_access=has_ufa,
+    )
+
+    if disable_quota_guard:
+
+        async def quota_checker(_cfg: object) -> dict[str, object]:
+            return {"should_sleep": False}
+    else:
+        from autoskillit.execution import check_and_sleep_if_needed
+
+        _supports_quota = (
+            effective_backend.capabilities.anthropic_provider_capable
+            if effective_backend
+            else True
+        )
+
+        async def quota_checker(_cfg: object) -> dict[str, object]:
+            return await check_and_sleep_if_needed(
+                _cfg, provider="anthropic" if _supports_quota else ""
+            )
+
+    async def quota_refresher(_cfg: object) -> None:
+        pass
+
+    return await execute_dispatch(
+        tool_ctx=ctx,
+        recipe=recipe,
+        task=task,
+        ingredients=ingredients,
+        dispatch_name=None,
+        timeout_sec=timeout_sec,
+        prompt_builder=prompt_builder,
+        quota_checker=quota_checker,
+        quota_refresher=quota_refresher,
+        cache_invalidator=None,
+        resume_session_id=resume_session_id,
+        prior_dispatch_id=prior_dispatch_id,
+        dispatch_backend=dispatch_backend,
+    )
+
+
+def fleet_run(
+    recipe: str,
+    *,
+    task: Annotated[str, Parameter(name=["--task", "-t"])] = "",
+    ingredient: Annotated[tuple[str, ...], Parameter(name=["--ingredient", "-i"])] = (),
+    backend: Annotated[str | None, Parameter(name=["--backend"])] = None,
+    timeout_sec: Annotated[int | None, Parameter(name=["--timeout-sec"])] = None,
+    resume_session_id: Annotated[str | None, Parameter(name=["--resume-session-id"])] = None,
+    prior_dispatch_id: Annotated[str | None, Parameter(name=["--prior-dispatch-id"])] = None,
+    disable_quota_guard: Annotated[bool, Parameter(name=["--disable-quota-guard"])] = False,
+) -> None:
+    """One-shot headless recipe dispatch (experimental).
+
+    Dispatches a single recipe run non-interactively. Prints the dispatch
+    result envelope as JSON on stdout. Exit 0 on SUCCESS, nonzero otherwise.
+    """
+    # --- Session-type guard (retained for headless — prevents recursive dispatch) ---
+    if os.environ.get("AUTOSKILLIT_SESSION_TYPE") in ("skill", "leaf"):
+        _fleet_run_error(
+            "FLEET_SESSION_TYPE_BLOCKED",
+            "'fleet run' cannot run inside a skill or leaf session.",
+        )
+    # NOTE: CLAUDECODE guard intentionally NOT applied — REQ-HFD-006
+
+    # --- Config + feature gates ---
+    from autoskillit.config import load_config
+
+    try:
+        cfg = load_config(Path.cwd())
+    except Exception as exc:
+        logger.error("fleet run: failed to load config", exc_info=True)
+        _fleet_run_error("FLEET_CONFIG_ERROR", f"Failed to load config: {exc}")
+
+    # Fleet base feature check (equivalent to _require_fleet but with JSON output)
+    if not is_feature_enabled(
+        "fleet", cfg.features, experimental_enabled=cfg.experimental_enabled
+    ):
+        _fleet_run_error(
+            "FLEET_FEATURE_DISABLED",
+            "The 'fleet' feature is not enabled.\n"
+            "Enable with: features.experimental_enabled: true in your config\n"
+            "Or set: AUTOSKILLIT_FEATURES__FLEET=true",
+        )
+
+    # Headless run feature check
+    if not is_feature_enabled(
+        "fleet_headless_run",
+        cfg.features,
+        experimental_enabled=cfg.experimental_enabled,
+    ):
+        _fleet_run_error(
+            "FLEET_FEATURE_DISABLED",
+            "The 'fleet_headless_run' feature is not enabled.\n"
+            "Enable with: features.experimental_enabled: true\n"
+            "Or: features.fleet_headless_run: true\n"
+            "Or: AUTOSKILLIT_FEATURES__FLEET_HEADLESS_RUN=true",
+        )
+
+    # --- Parse ingredients ---
+    ingredients: dict[str, str] | None = None
+    if ingredient:
+        ingredients = {}
+        for item in ingredient:
+            if "=" not in item:
+                _fleet_run_error(
+                    "FLEET_INVALID_ARGUMENT",
+                    f"Ingredient must be key=value, got: {item!r}",
+                )
+            k, v = item.split("=", 1)
+            ingredients[k] = v
+
+    # --- Resolve backend ---
+    dispatch_backend = None
+    if backend is not None:
+        from autoskillit.server import resolve_backend_override
+
+        try:
+            dispatch_backend = resolve_backend_override(backend)
+        except ValueError as exc:
+            _fleet_run_error("FLEET_INVALID_BACKEND", str(exc))
+
+    # --- Run dispatch ---
+    import asyncio
+
+    from autoskillit.fleet import DispatchRejected, DispatchStatus
+
+    try:
+        result = asyncio.run(
+            _execute_fleet_run(
+                cfg=cfg,
+                recipe=recipe,
+                task=task,
+                ingredients=ingredients,
+                timeout_sec=timeout_sec,
+                dispatch_backend=dispatch_backend,
+                resume_session_id=resume_session_id,
+                prior_dispatch_id=prior_dispatch_id,
+                disable_quota_guard=disable_quota_guard,
+            )
+        )
+    except KeyboardInterrupt as exc:
+        logger.error("fleet run: dispatch interrupted: %s", exc)
+        _fleet_run_error(
+            "FLEET_DISPATCH_INTERRUPTED", "Dispatch interrupted by signal.", exit_code=1
+        )
+    except Exception as exc:
+        logger.error("fleet run: dispatch crashed", exc_info=True)
+        _fleet_run_error("FLEET_L3_STARTUP_OR_CRASH", str(exc))
+
+    # --- Output result envelope ---
+    print(result.outcome.to_envelope())
+
+    # --- Exit code (DispatchRejected has no .success/.dispatch_status — check it first) ---
+    if isinstance(result.outcome, DispatchRejected):
+        raise SystemExit(3)
+    if result.outcome.success:
+        raise SystemExit(0)
+    if result.outcome.dispatch_status == DispatchStatus.RESUMABLE:
+        raise SystemExit(2)
+    raise SystemExit(1)

src/autoskillit/core/types/_type_constants_features.py

@@ -60,6 +60,17 @@ class FeatureDef:
         default_enabled=False,
         since_version="0.9.119",
     ),
+    "fleet_headless_run": FeatureDef(
+        lifecycle=FeatureLifecycle.EXPERIMENTAL,
+        description="Headless one-shot recipe dispatch via CLI (fleet run)",
+        tool_tags=frozenset(),
+        skill_categories=frozenset(),
+        import_package=None,
+        tier=1,
+        default_enabled=False,
+        depends_on=frozenset({"fleet"}),
+        since_version="0.10.845",
+    ),
     "planner": FeatureDef(
         lifecycle=FeatureLifecycle.EXPERIMENTAL,
         description=(

src/autoskillit/server/__init__.py

@@ -56,6 +56,7 @@
     # Public utilities consumed by CLI and tests
     "version_info",
     "make_context",
+    "resolve_backend_override",
     # Wire-format compatibility middleware
     "ClaudeCodeCompatMiddleware",
     # Session-type visibility dispatcher (callable by tests)
@@ -77,6 +78,7 @@
     _notify,
 )
 from autoskillit.server._factory import make_context  # noqa: E402, F401
+from autoskillit.server._misc import resolve_backend_override  # noqa: E402, F401
 from autoskillit.server._session_type import _apply_session_type_visibility  # noqa: E402, F401
 from autoskillit.server.tools import (  # noqa: E402, F401
     tools_agents as _tools_agents,

tests/arch/_rules.py

@@ -63,6 +63,7 @@ class RuleDescriptor:
         "_fleet_display.py",
         "_fleet_lifecycle.py",
         "_fleet_preview.py",
+        "_fleet_run.py",
         "_fleet_session.py",
         "_session_picker.py",
         "_fleet.py",

tests/arch/test_feature_markers.py

@@ -31,6 +31,7 @@
         _TESTS_ROOT / "cli" / "test_fleet_campaign.py",
         _TESTS_ROOT / "cli" / "test_fleet_status.py",
         _TESTS_ROOT / "cli" / "test_fleet_list.py",
+        _TESTS_ROOT / "cli" / "test_fleet_run.py",
         _TESTS_ROOT / "server" / "test_tools_dispatch.py",
         _TESTS_ROOT / "server" / "test_tools_dispatch_validation.py",
         _TESTS_ROOT / "server" / "test_tools_dispatch_params.py",

tests/arch/test_feature_registry.py

@@ -345,6 +345,32 @@ def test_build_features_dict_franchise_raises_config_schema_error():
         AutomationConfig._build_features_dict({"franchise": True})
 
 
+# ── fleet_headless_run feature registry tests ─────────────────────────────────
+
+
+def test_fleet_headless_run_in_feature_registry():
+    """fleet_headless_run is registered with correct FeatureDef field values."""
+    from autoskillit.core.types._type_constants_features import FEATURE_REGISTRY
+    from autoskillit.core.types._type_enums import FeatureLifecycle
+
+    entry = FEATURE_REGISTRY["fleet_headless_run"]
+    assert entry.lifecycle == FeatureLifecycle.EXPERIMENTAL
+    assert entry.tier == 1
+    assert entry.default_enabled is False
+    assert entry.tool_tags == frozenset()
+    assert entry.skill_categories == frozenset()
+    assert entry.depends_on == frozenset({"fleet"})
+    assert entry.import_package is None
+
+
+def test_fleet_headless_run_promoted_by_experimental_blanket():
+    """fleet_headless_run has no requires_backend_alignment, experimental blanket promotes it."""
+    from autoskillit.core.feature_flags import is_feature_enabled
+
+    assert is_feature_enabled("fleet_headless_run", {}, experimental_enabled=True) is True
+    assert is_feature_enabled("fleet_headless_run", {}, experimental_enabled=False) is False
+
+
 # ── Providers feature registry tests ─────────────────────────────────────────
 
 

tests/cli/AGENTS.md

@@ -43,6 +43,7 @@ CLI command, subcommand, and interactive workflow tests.
 | `test_fleet_campaign_preview.py` | Tests: fleet_campaign shows preview + confirmation before launch |
 | `test_fleet_dispatch.py` | Tests: fleet CLI dispatch command |
 | `test_fleet_list.py` | Tests: fleet CLI list command |
+| `test_fleet_run.py` | Tests: fleet CLI run command gates — session-type guard, feature gates, CLAUDECODE relaxation |
 | `test_fleet_session.py` | Tests: _launch_fleet_session forwards ingredients_table to prompt builder |
 | `test_fleet_split.py` | Structural guard for fleet test split |
 | `test_fleet_status.py` | Tests: fleet CLI status command |

tests/cli/test_fleet_list.py

@@ -97,7 +97,7 @@ def test_fleet_campaign_command_registered(self) -> None:
 
         assert "campaign" in _subcommand_names(fleet_app)
 
-    def test_fleet_run_command_not_registered(self) -> None:
+    def test_fleet_run_command_registered(self) -> None:
         from autoskillit.cli.fleet import fleet_app
 
-        assert "run" not in _subcommand_names(fleet_app)
+        assert "run" in _subcommand_names(fleet_app)