gyxx-flow/src/gyxx_flow/cli.py

"""Operator-safe command-line entry point for GYXX Flow."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import signal
import sys
import threading
from collections.abc import Sequence
from contextlib import suppress
from datetime import date, timedelta
from pathlib import Path
from typing import TextIO

from dotenv import dotenv_values

from gyxx_flow.acceptance import build_acceptance_report
from gyxx_flow.adapters.native import DeferredModuleCommandStep
from gyxx_flow.catalog import CatalogError, WorkflowCatalog, WorkflowEntry
from gyxx_flow.core.artifacts import atomic_write_json
from gyxx_flow.core.config import Settings
from gyxx_flow.core.context import RunContext
from gyxx_flow.core.layout import DataLayout
from gyxx_flow.core.locks import LockManager
from gyxx_flow.core.records import RunJournal
from gyxx_flow.diagnostics import run_doctor
from gyxx_flow.modules import create_default_registry as create_module_registry
from gyxx_flow.modules.product_commerce import (
    TMALL_BAIBU_DEFAULT_IMPORT_MODE,
    TMALL_BAIBU_WORKFLOW_ID,
    validate_tmall_baibu_import_mode,
)
from gyxx_flow.ops import EffectLedger, EffectStateAmbiguous, RunIndex
from gyxx_flow.scheduler_service import (
    PythonScheduler,
    SchedulerInstanceLock,
    SubprocessWorkflowLauncher,
)
from gyxx_flow.script_catalog import ScriptCatalog, ScriptCatalogError
from gyxx_flow.source_sync.cli import add_sources_parser, run_sources_command
from gyxx_flow.workflow.engine import WorkflowEngine, WorkflowRunResult
from gyxx_flow.workflow.model import StepDefinition, WorkflowDefinition
from gyxx_flow.workflow.registry import WorkflowRegistry, WorkflowRegistryError
from gyxx_flow.workflow.selection import rerun_step, resume_from

EXIT_SUCCESS = 0
EXIT_WORKFLOW_FAILED = 1
EXIT_USAGE = 2
EXIT_CONFIGURATION = 3
EXIT_RUNTIME = 4
EXIT_ACCEPTANCE_INCOMPLETE = 1
MAX_BACKFILL_DAYS = 366


class CliConfigurationError(ValueError):
    """Safe operator-facing configuration or input error."""


class CliRuntimeError(RuntimeError):
    """Sanitized runtime failure."""


def build_parser() -> argparse.ArgumentParser:
    """Build the CLI parser without initializing runtime resources."""

    parser = argparse.ArgumentParser(
        prog="gyxx",
        description="GYXX Flow workflow orchestration",
    )
    commands = parser.add_subparsers(dest="command", required=True)

    add_sources_parser(commands)

    list_parser = commands.add_parser("list", help="list scheduled workflows")
    list_parser.add_argument("--json", action="store_true", help="emit machine-readable JSON")

    scripts_parser = commands.add_parser(
        "scripts", help="list and manually run explicitly registered commands"
    )
    script_commands = scripts_parser.add_subparsers(dest="scripts_command", required=True)
    scripts_list_parser = script_commands.add_parser("list", help="list runnable scripts")
    scripts_list_parser.add_argument("--module")
    scripts_list_parser.add_argument("--json", action="store_true")
    scripts_run_parser = script_commands.add_parser("run", help="run one local script")
    scripts_run_parser.add_argument("script_id")
    scripts_run_parser.add_argument(
        "--date", required=True, help="business date (YYYY-MM-DD)"
    )
    scripts_run_parser.add_argument("--execute", action="store_true")
    scripts_run_parser.add_argument("--shadow", action="store_true")
    scripts_run_parser.add_argument(
        "--arg",
        dest="script_args",
        action="append",
        default=[],
        help="one argument passed to the script; repeat for multiple arguments",
    )

    accounts_parser = commands.add_parser(
        "accounts",
        help="manage account-level browser login states shared by scripts",
    )
    account_commands = accounts_parser.add_subparsers(
        dest="accounts_command", required=True
    )
    accounts_list_parser = account_commands.add_parser("list", help="list accounts")
    accounts_list_parser.add_argument("--json", action="store_true")
    accounts_login_parser = account_commands.add_parser(
        "login", help="open the account login browser and save the vault state"
    )
    accounts_login_parser.add_argument("account_id")
    accounts_login_parser.add_argument(
        "--timeout", type=int, default=600, help="login wait timeout in seconds"
    )
    accounts_login_parser.add_argument("--target-url", help="page to open for login")
    accounts_login_parser.add_argument("--force", action="store_true")
    accounts_login_parser.add_argument("--json", action="store_true")
    accounts_sync_parser = account_commands.add_parser(
        "sync", help="push account vault cookies into member scripts"
    )
    accounts_sync_parser.add_argument("account_id", nargs="?")
    accounts_sync_parser.add_argument("--json", action="store_true")
    accounts_seed_parser = account_commands.add_parser(
        "seed", help="import one existing script browser state into an account vault"
    )
    accounts_seed_parser.add_argument("account_id")
    accounts_seed_parser.add_argument("--source-binding", required=True)
    accounts_seed_parser.add_argument("--force", action="store_true")
    accounts_seed_parser.add_argument("--json", action="store_true")

    effects_parser = commands.add_parser(
        "effects", help="inspect or reconcile guarded production effects"
    )
    effect_commands = effects_parser.add_subparsers(
        dest="effects_command", required=True
    )
    reconcile_parser = effect_commands.add_parser(
        "reconcile", help="resolve one verified ambiguous effect receipt"
    )
    reconcile_parser.add_argument("workflow_id")
    reconcile_parser.add_argument("--date", required=True, dest="business_date")
    reconcile_parser.add_argument("--step", required=True, dest="step_id")
    reconcile_parser.add_argument("--expected-run-id", required=True)
    reconcile_parser.add_argument(
        "--action", required=True, choices=("retry", "applied")
    )
    reconcile_parser.add_argument("--operator", required=True)
    reconcile_parser.add_argument("--reason", required=True)
    reconcile_parser.add_argument("--evidence", required=True)
    reconcile_parser.add_argument(
        "--execute",
        action="store_true",
        help="confirm the audited receipt mutation",
    )

    run_parser = commands.add_parser("run", help="run one workflow business date")
    _add_execution_arguments(run_parser)
    run_parser.add_argument(
        "--date", required=True, help="business date (YYYY-MM-DD)"
    )

    backfill_parser = commands.add_parser(
        "backfill", help="run an inclusive business-date range"
    )
    _add_execution_arguments(backfill_parser)
    backfill_parser.add_argument(
        "--from", dest="from_date", required=True, help="first business date"
    )
    backfill_parser.add_argument(
        "--to", dest="to_date", required=True, help="last business date"
    )

    schedule_parser = commands.add_parser(
        "schedule", help="run the project-owned resident scheduler"
    )
    schedule_commands = schedule_parser.add_subparsers(
        dest="schedule_command", required=True
    )
    run_schedule_parser = schedule_commands.add_parser(
        "run", help="run the cross-platform resident Python scheduler"
    )
    run_schedule_parser.add_argument("--python-executable", type=Path)
    run_schedule_parser.add_argument("--poll-seconds", type=float, default=15.0)
    run_schedule_parser.add_argument("--misfire-grace-seconds", type=int, default=21600)
    run_schedule_parser.add_argument("--shutdown-timeout-seconds", type=float, default=300.0)
    run_schedule_parser.add_argument("--once", action="store_true")
    run_schedule_parser.add_argument("--dry-run", action="store_true")
    run_schedule_parser.add_argument(
        "--env-file",
        action="append",
        default=[],
        type=Path,
        help="explicit runtime environment file; repeatable, existing process values win",
    )
    schedule_commands.add_parser("status", help="show persisted Python scheduler state")

    console_parser = commands.add_parser(
        "console", help="serve the workflow operator web console"
    )
    console_parser.add_argument("--host", default="127.0.0.1")
    console_parser.add_argument("--port", type=int, default=8765)
    console_parser.add_argument(
        "--env-file",
        action="append",
        default=[],
        type=Path,
        help="explicit runtime environment file; repeatable, existing process values win",
    )

    doctor_parser = commands.add_parser("doctor", help="run environment preflight checks")
    doctor_parser.add_argument("--json", action="store_true")

    acceptance_parser = commands.add_parser(
        "acceptance", help="inspect acceptance evidence"
    )
    acceptance_commands = acceptance_parser.add_subparsers(
        dest="acceptance_command", required=True
    )
    status_parser = acceptance_commands.add_parser("status")
    status_parser.add_argument("--json", action="store_true")
    status_parser.add_argument("--output", type=Path)
    return parser


def _add_execution_arguments(parser: argparse.ArgumentParser) -> None:
    parser.add_argument("workflow_id")
    parser.add_argument(
        "--execute",
        action="store_true",
        help="execute registered steps; omitted means no-side-effect dry-run",
    )
    parser.add_argument("--shadow", action="store_true")
    parser.add_argument(
        "--tmall-baibu-import-mode",
        choices=("all", "without_hyperlinks"),
        help="Tmall Baibu scope: all entries or only entries without hyperlinks",
    )
    recovery = parser.add_mutually_exclusive_group()
    recovery.add_argument("--resume-from", metavar="STEP_ID")
    recovery.add_argument("--rerun-step", metavar="STEP_ID")


def build_default_registry(
    settings: Settings,
    *,
    tmall_baibu_import_mode: str | None = None,
) -> WorkflowRegistry:
    """Compose only workflows explicitly implemented by migrated modules."""

    catalog = WorkflowCatalog.load(settings.project_root / "config")
    registry = WorkflowRegistry(catalog)
    modules = create_module_registry(
        catalog=catalog,
        tmall_baibu_import_mode=tmall_baibu_import_mode,
    )
    for workflow_id in modules.workflow_ids:
        registry.register(modules.workflow(workflow_id))
    return registry


def main(
    argv: Sequence[str] | None = None,
    *,
    registry: WorkflowRegistry | None = None,
    settings: Settings | None = None,
    stdout: TextIO | None = None,
    stderr: TextIO | None = None,
) -> int:
    """Execute a CLI command and return a stable process exit code."""

    output = stdout or sys.stdout
    errors = stderr or sys.stderr
    parser = build_parser()
    try:
        arguments = parser.parse_args(argv)
    except SystemExit as exc:
        return int(exc.code)

    try:
        _load_runtime_environment_files(arguments)
        resolved_settings = settings or Settings.from_env()
        if arguments.command == "doctor":
            return _doctor_command(arguments, resolved_settings, output)
        if arguments.command == "console":
            return _console_command(arguments, resolved_settings, output)
        if arguments.command == "sources":
            return run_sources_command(
                arguments,
                project_root=resolved_settings.project_root,
                output=output,
            )
        if (
            arguments.command == "acceptance"
            and arguments.acceptance_command == "status"
        ):
            return _acceptance_status_command(arguments, resolved_settings, output)
        if arguments.command == "scripts" and arguments.scripts_command == "list":
            return _list_scripts(arguments, output)
        if arguments.command == "scripts" and arguments.scripts_command == "run":
            return _run_script(arguments, resolved_settings, output)
        if (
            arguments.command == "effects"
            and arguments.effects_command == "reconcile"
        ):
            return _effects_reconcile_command(
                arguments, resolved_settings, output
            )
        if arguments.command == "accounts":
            if arguments.accounts_command == "list":
                return _accounts_list_command(arguments, resolved_settings, output)
            if arguments.accounts_command == "login":
                return _accounts_login_command(arguments, resolved_settings, output)
            if arguments.accounts_command == "sync":
                return _accounts_sync_command(arguments, resolved_settings, output)
            if arguments.accounts_command == "seed":
                return _accounts_seed_command(arguments, resolved_settings, output)
            raise CliConfigurationError(
                f"unsupported accounts command: {arguments.accounts_command}"
            )
        tmall_baibu_import_mode: str | None = None
        if arguments.command in {"run", "backfill"}:
            raw_import_mode = getattr(arguments, "tmall_baibu_import_mode", None)
            if raw_import_mode is not None and arguments.workflow_id != TMALL_BAIBU_WORKFLOW_ID:
                raise CliConfigurationError(
                    "--tmall-baibu-import-mode 仅适用于 product.tmall_baibu_apply"
                )
            if arguments.workflow_id == TMALL_BAIBU_WORKFLOW_ID:
                tmall_baibu_import_mode = validate_tmall_baibu_import_mode(
                    raw_import_mode or TMALL_BAIBU_DEFAULT_IMPORT_MODE
                )
        resolved_registry = registry or build_default_registry(
            resolved_settings,
            tmall_baibu_import_mode=tmall_baibu_import_mode,
        )
        if arguments.command == "list":
            return _list_workflows(arguments, resolved_registry, output)
        if arguments.command == "run":
            return _run_command(arguments, resolved_registry, resolved_settings, output)
        if arguments.command == "backfill":
            return _backfill_command(arguments, resolved_registry, resolved_settings, output)
        if arguments.command == "schedule" and arguments.schedule_command == "run":
            return _schedule_run_command(arguments, resolved_registry.catalog, resolved_settings, output)
        if arguments.command == "schedule" and arguments.schedule_command == "status":
            return _schedule_status_command(resolved_registry.catalog, resolved_settings, output)
        raise CliConfigurationError(f"unsupported command: {arguments.command}")
    except (
        CatalogError,
        ScriptCatalogError,
        WorkflowRegistryError,
        CliConfigurationError,
        ValueError,
    ) as exc:
        errors.write(f"error: {exc}\n")
        return EXIT_CONFIGURATION
    except CliRuntimeError as exc:
        errors.write(f"error: {exc}\n")
        return EXIT_RUNTIME


def _load_runtime_environment_files(arguments: argparse.Namespace) -> None:
    """Load only explicitly named service environment files without leaking values."""

    raw_paths = getattr(arguments, "env_file", ()) or ()
    for raw_path in raw_paths:
        path = Path(raw_path).expanduser().resolve()
        if not path.is_file():
            raise CliConfigurationError(f"runtime environment file not found: {path}")
        try:
            values = dotenv_values(path, encoding="utf-8")
        except (OSError, UnicodeError) as exc:
            raise CliConfigurationError(
                f"cannot read runtime environment file: {path}"
            ) from exc
        for key, value in values.items():
            if not isinstance(key, str) or not key or value is None:
                continue
            # A service wrapper may export an empty placeholder for a secret
            # or database setting.  Treat that as unset so the explicitly
            # requested runtime file can supply it; a non-empty process value
            # remains authoritative.
            if not os.environ.get(key, "").strip():
                os.environ[key] = value


def _doctor_command(
    arguments: argparse.Namespace, settings: Settings, output: TextIO
) -> int:
    report = run_doctor(settings)
    if arguments.json:
        _write_json(output, report.as_dict())
    else:
        for check in report.checks:
            marker = "PASS" if check.passed else "FAIL"
            output.write(f"{marker}\t{check.name}\t{check.message}\n")
    return EXIT_SUCCESS if report.is_healthy else EXIT_RUNTIME


def _effects_reconcile_command(
    arguments: argparse.Namespace,
    settings: Settings,
    output: TextIO,
) -> int:
    if not arguments.execute:
        raise CliConfigurationError("effect reconciliation requires --execute")
    try:
        result = EffectLedger(settings.data_root).reconcile(
            workflow_id=arguments.workflow_id,
            business_date=arguments.business_date,
            step_id=arguments.step_id,
            expected_run_id=arguments.expected_run_id,
            action=arguments.action,
            operator=arguments.operator,
            reason=arguments.reason,
            evidence=arguments.evidence,
        )
    except EffectStateAmbiguous as exc:
        raise CliConfigurationError(str(exc)) from exc
    _write_json(output, result)
    return EXIT_SUCCESS


def _accounts_list_command(
    arguments: argparse.Namespace,
    settings: Settings,
    output: TextIO,
) -> int:
    from gyxx_flow.accounts import list_accounts

    payload = list_accounts(settings)
    if arguments.json:
        _write_json(output, payload)
    else:
        for item in payload["accounts"]:
            state = "valid" if item["vault_valid"] else (
                "missing" if not item["vault_exists"] else "invalid"
            )
            output.write(
                f"{item['account_id']}\tcdp={item['cdp_port']}\tstate={state}"
                f"\tmembers={len(item['members'])}\n"
            )
    return EXIT_SUCCESS


def _accounts_login_command(
    arguments: argparse.Namespace,
    settings: Settings,
    output: TextIO,
) -> int:
    from gyxx_flow.accounts import login_account

    try:
        result = login_account(
            settings,
            arguments.account_id,
            output=output,
            timeout_seconds=arguments.timeout,
            target_url=arguments.target_url,
            force=arguments.force,
        )
    except TimeoutError as exc:
        raise CliRuntimeError(str(exc)) from exc
    if arguments.json:
        _write_json(output, result)
    return EXIT_SUCCESS


def _accounts_sync_command(
    arguments: argparse.Namespace,
    settings: Settings,
    output: TextIO,
) -> int:
    from gyxx_flow.accounts import sync_account

    try:
        result = sync_account(settings, arguments.account_id, output)
    except ValueError as exc:
        raise CliConfigurationError(str(exc)) from exc
    if arguments.json:
        _write_json(output, result)
    return EXIT_SUCCESS


def _accounts_seed_command(
    arguments: argparse.Namespace,
    settings: Settings,
    output: TextIO,
) -> int:
    from gyxx_flow.accounts import seed_account

    try:
        result = seed_account(
            settings,
            arguments.account_id,
            arguments.source_binding,
            output,
            force=arguments.force,
        )
    except ValueError as exc:
        raise CliConfigurationError(str(exc)) from exc
    if arguments.json:
        _write_json(output, result)
    return EXIT_SUCCESS


def _console_command(
    arguments: argparse.Namespace,
    settings: Settings,
    output: TextIO,
) -> int:
    from gyxx_flow.console import serve_console

    try:
        return serve_console(
            settings,
            host=arguments.host,
            port=arguments.port,
            output=output,
            env_files=arguments.env_file,
        )
    except OSError as exc:
        raise CliRuntimeError("cannot start workflow console") from exc


def _acceptance_status_command(
    arguments: argparse.Namespace, settings: Settings, output: TextIO
) -> int:
    report = build_acceptance_report(settings)
    payload = report.as_dict()
    if arguments.output is not None:
        atomic_write_json(arguments.output, payload)
    if arguments.json:
        _write_json(output, payload)
    else:
        summary = payload["summary"]
        output.write(
            f"acceptance: {summary['completed']}/{summary['total']} complete; "
            f"pending={summary['pending']}\n"
        )
    return EXIT_SUCCESS if report.is_complete else EXIT_ACCEPTANCE_INCOMPLETE


def _list_workflows(
    arguments: argparse.Namespace,
    registry: WorkflowRegistry,
    output: TextIO,
) -> int:
    rows = [
        {
            "id": entry.workflow_id,
            "module": entry.module,
            "registered": registry.is_registered(entry.workflow_id),
            "trigger": entry.trigger,
            "enabled": registry.catalog.schedule_for(entry.workflow_id).enabled,
        }
        for entry in registry.catalog.scheduled_workflows()
    ]
    if arguments.json:
        _write_json(output, rows)
    else:
        for row in rows:
            state = (
                "not-registered"
                if not row["registered"]
                else "ready" if row["enabled"] else "disabled"
            )
            output.write(
                f"{row['id']}\t{row['module']}\t{row['trigger']}\t{state}\n"
            )
    return EXIT_SUCCESS


def _list_scripts(arguments: argparse.Namespace, output: TextIO) -> int:
    catalog = ScriptCatalog.load_default()
    rows = [
        {
            "id": script.command_id,
            "command_id": script.command_id,
            "legacy_script_id": script.legacy_script_id,
            "module": script.module,
            "entry": script.entry,
            "kind": script.kind,
            "default_args": list(script.default_args),
            "notification_workflow_id": script.notification_workflow_id,
        }
        for script in catalog.scripts
        if arguments.module is None or script.module == arguments.module
    ]
    if arguments.module is not None and not rows:
        raise CliConfigurationError(f"unknown module or no runnable scripts: {arguments.module}")
    if arguments.json:
        _write_json(output, rows)
    else:
        for row in rows:
            output.write(f"{row['id']}\t{row['kind']}\n")
    return EXIT_SUCCESS


def _run_script(
    arguments: argparse.Namespace,
    settings: Settings,
    output: TextIO,
) -> int:
    script = ScriptCatalog.load_default().get(arguments.script_id)
    business_date = _parse_date(arguments.date)
    script_args = (
        *_render_script_default_args(script.default_args, business_date),
        *arguments.script_args,
    )
    invocation_identity = json.dumps(
        [script.command_id, *script_args],
        ensure_ascii=False,
        separators=(",", ":"),
    )
    suffix = hashlib.sha256(invocation_identity.encode("utf-8")).hexdigest()[:12]
    workflow_id = f"script.{script.module}.{suffix}"
    entry = WorkflowEntry(
        workflow_id=workflow_id,
        module=script.module,
        trigger="manual",
        entry=script.entry,
        args=script_args,
        notification_workflow_id=script.notification_workflow_id,
    )
    workflow = WorkflowDefinition(
        workflow_id,
        (
            StepDefinition(
                "module_script",
                DeferredModuleCommandStep(entry),
                timeout_seconds=6 * 60 * 60,
                max_attempts=1,
                resources=(f"module:{script.module}",),
                production_sink=True,
            ),
        ),
    )
    exit_code, payload = _execute_once(
        workflow,
        business_date=business_date,
        shadow=arguments.shadow,
        dry_run=not arguments.execute,
        settings=settings,
    )
    payload["command_id"] = script.command_id
    payload["script_id"] = script.legacy_script_id
    _write_json(output, payload)
    return exit_code


def _render_script_default_args(
    arguments: tuple[str, ...], business_date: date
) -> tuple[str, ...]:
    return tuple(
        argument.replace("{business_date}", business_date.isoformat())
        for argument in arguments
    )


def _run_command(
    arguments: argparse.Namespace,
    registry: WorkflowRegistry,
    settings: Settings,
    output: TextIO,
) -> int:
    registered = registry.resolve(arguments.workflow_id)
    workflow = _select_workflow(registered.definition, arguments)
    business_date = _parse_date(arguments.date)
    dry_run = not arguments.execute
    exit_code, payload = _execute_once(
        workflow,
        business_date=business_date,
        shadow=arguments.shadow,
        dry_run=dry_run,
        settings=settings,
    )
    _write_json(output, payload)
    return exit_code


def _python_scheduler(
    catalog: WorkflowCatalog,
    settings: Settings,
    *,
    python_executable: Path | None = None,
    misfire_grace_seconds: int = 21600,
    dry_run: bool = False,
    enable_account_keepalive: bool = False,
) -> PythonScheduler:
    launcher = SubprocessWorkflowLauncher(
        settings.project_root, settings.data_root, python_executable=python_executable
    )
    maintenance = None
    if enable_account_keepalive and not dry_run:
        from gyxx_flow.accounts import AccountKeepaliveManager

        maintenance = AccountKeepaliveManager(settings).tick
    return PythonScheduler(
        catalog,
        settings.data_root,
        launcher=launcher,
        misfire_grace_seconds=misfire_grace_seconds,
        dry_run=dry_run,
        maintenance=maintenance,
        catalog_loader=lambda: WorkflowCatalog.load(
            settings.project_root / "config"
        ),
    )


def _schedule_run_command(
    arguments: argparse.Namespace,
    catalog: WorkflowCatalog,
    settings: Settings,
    output: TextIO,
) -> int:
    if arguments.poll_seconds <= 0:
        raise CliConfigurationError("poll seconds must be positive")
    if arguments.shutdown_timeout_seconds < 0:
        raise CliConfigurationError("shutdown timeout must be non-negative")
    scheduler = _python_scheduler(
        catalog,
        settings,
        python_executable=arguments.python_executable,
        misfire_grace_seconds=arguments.misfire_grace_seconds,
        dry_run=arguments.dry_run,
        enable_account_keepalive=True,
    )
    lock_path = settings.data_root / "state" / "scheduler" / "service.lock"
    try:
        with SchedulerInstanceLock(lock_path):
            first = scheduler.tick()
            if arguments.once or arguments.dry_run:
                if not arguments.dry_run:
                    scheduler.wait_for_active(arguments.shutdown_timeout_seconds)
                _write_json(output, {
                    "mode": "dry-run" if arguments.dry_run else "once",
                    "due_or_started": first,
                    "state": scheduler.status(),
                })
                return EXIT_SUCCESS

            stop = threading.Event()
            previous_handlers: dict[int, object] = {}

            def request_stop(_signum, _frame) -> None:  # type: ignore[no-untyped-def]
                stop.set()

            for signum in (signal.SIGINT, signal.SIGTERM):
                previous_handlers[signum] = signal.getsignal(signum)
                signal.signal(signum, request_stop)
            try:
                while not stop.wait(arguments.poll_seconds):
                    scheduler.tick()
            finally:
                for signum, handler in previous_handlers.items():
                    signal.signal(signum, handler)
                scheduler.wait_for_active(arguments.shutdown_timeout_seconds)
            _write_json(output, {"mode": "service", "state": scheduler.status()})
            return EXIT_SUCCESS
    except RuntimeError as exc:
        raise CliRuntimeError(str(exc)) from exc


def _schedule_status_command(
    catalog: WorkflowCatalog, settings: Settings, output: TextIO
) -> int:
    scheduler = _python_scheduler(catalog, settings)
    _write_json(output, {
        "timezone": catalog.timezone,
        "scheduled_count": len(catalog.scheduled_workflows()),
        "state": scheduler.status(),
    })
    return EXIT_SUCCESS


def _backfill_command(
    arguments: argparse.Namespace,
    registry: WorkflowRegistry,
    settings: Settings,
    output: TextIO,
) -> int:
    registered = registry.resolve(arguments.workflow_id)
    workflow = _select_workflow(registered.definition, arguments)
    start = _parse_date(arguments.from_date)
    end = _parse_date(arguments.to_date)
    if end < start:
        raise CliConfigurationError("backfill end date must not precede start date")
    day_count = (end - start).days + 1
    if day_count > MAX_BACKFILL_DAYS:
        raise CliConfigurationError(
            f"backfill range exceeds safety limit of {MAX_BACKFILL_DAYS} days"
        )

    runs: list[dict[str, object]] = []
    final_exit = EXIT_SUCCESS
    for offset in range(day_count):
        business_date = start + timedelta(days=offset)
        exit_code, payload = _execute_once(
            workflow,
            business_date=business_date,
            shadow=arguments.shadow,
            dry_run=not arguments.execute,
            settings=settings,
        )
        runs.append(payload)
        if exit_code != EXIT_SUCCESS:
            final_exit = EXIT_WORKFLOW_FAILED
    _write_json(
        output,
        {
            "workflow_id": workflow.workflow_id,
            "from": start.isoformat(),
            "to": end.isoformat(),
            "dry_run": not arguments.execute,
            "status": "success" if final_exit == EXIT_SUCCESS else "failed",
            "runs": runs,
        },
    )
    return final_exit


def _select_workflow(workflow, arguments: argparse.Namespace):  # type: ignore[no-untyped-def]
    if arguments.rerun_step:
        return rerun_step(workflow, arguments.rerun_step)
    if arguments.resume_from:
        return resume_from(workflow, arguments.resume_from)
    return workflow


def _execute_once(
    workflow,  # type: ignore[no-untyped-def]
    *,
    business_date: date,
    shadow: bool,
    dry_run: bool,
    settings: Settings,
) -> tuple[int, dict[str, object]]:
    context = RunContext.create(
        workflow.workflow_id,
        business_date.isoformat(),
        shadow=shadow,
    )
    layout = DataLayout(settings.data_root)
    journal: RunJournal | None = None
    try:
        journal = RunJournal.create(
            layout,
            context,
            mode="dry_run" if dry_run else "execute",
        )
        RunIndex(settings.data_root).index_journal(journal)
        result = WorkflowEngine(
            LockManager(settings.data_root / "state" / "locks"),
            effect_ledger=EffectLedger(settings.data_root),
        ).execute(workflow, context=context, journal=journal, dry_run=dry_run)
        RunIndex(settings.data_root).index_journal(journal)
    except Exception as exc:
        if journal is not None:
            with suppress(OSError, ValueError, KeyError, TypeError, json.JSONDecodeError):
                journal.finalize("failed", error="runtime infrastructure failure")
                RunIndex(settings.data_root).index_journal(journal)
        raise CliRuntimeError(
            f"runtime failure while executing {workflow.workflow_id}"
        ) from exc
    return result.exit_code, _result_payload(
        context,
        result,
        dry_run=dry_run,
        journal_path=journal.path.relative_to(settings.data_root).as_posix(),
    )


def _result_payload(
    context: RunContext,
    result: WorkflowRunResult,
    *,
    dry_run: bool,
    journal_path: str,
) -> dict[str, object]:
    return {
        "workflow_id": result.workflow_id,
        "run_id": context.run_id,
        "business_date": context.business_date.isoformat(),
        "shadow": context.shadow,
        "dry_run": dry_run,
        "status": result.status,
        "exit_code": result.exit_code,
        "journal_path": journal_path,
        "steps": {
            step_id: step.status for step_id, step in result.steps.items()
        },
        "warnings": list(result.warnings),
    }


def _parse_date(value: str) -> date:
    try:
        parsed = date.fromisoformat(value)
    except (TypeError, ValueError) as exc:
        raise CliConfigurationError(f"invalid business date: {value!r}") from exc
    if parsed.isoformat() != value:
        raise CliConfigurationError(f"invalid business date: {value!r}")
    return parsed


def _write_json(output: TextIO, payload: object) -> None:
    json.dump(payload, output, ensure_ascii=False, sort_keys=True)
    output.write("\n")