refactor: organize workflow CLI commands

design/cli.md

@@ -148,11 +148,20 @@ extensions/
 The nested package's `__init__.py` owns its Typer application and registration.
 Shared helpers for that nested surface can live in `_helpers.py`.
 
-Creating a nested CLI package does not transfer same-named domain behavior into
-that package. If an existing domain module collides with a new nested command
-namespace, keep the implementation in the parent domain package (or a focused
-domain module there). Preserve an established import path through thin
-compatibility exports from the nested package when required.
+A nested CLI hierarchy may also be the root of a bounded subdomain when its
+concept depends on the parent domain but owns a distinct resource and lifecycle
+that the parent commands do not cover. In that case, keep the subdomain's
+non-command modules and its `command_*.py` adapters together in the nested
+package. Storage, validation, composition, or distribution behavior specific to
+that resource are signals that the namespace is a domain root, not merely a CLI
+group.
+
+Creating a nested CLI package solely to group commands does not transfer
+same-named parent-domain behavior into that package. If an existing domain
+module merely collides with a new nested command namespace, keep the
+implementation in the parent domain package (or a focused domain module there).
+Preserve an established import path through thin compatibility exports from the
+nested package when required.
 
 Do not add a nested `_commands.py` merely for symmetry. Create one only when
 the nested group develops substantial shared command infrastructure that no

src/specify_cli/__init__.py

@@ -592,12 +592,11 @@ def _require_specify_project() -> Path:
 # Re-exported at the package root because bundler primitives import these
 # handlers via ``from specify_cli import workflow_*`` (and tests monkeypatch
 # ``specify_cli.workflow_add``). Keep these names resolvable from the root.
-from .workflows._commands import (  # noqa: E402,F401
-    workflow_add,
-    workflow_remove,
-    workflow_step_add,
-    workflow_step_remove,
-)
+from .workflows.command_add import workflow_add  # noqa: E402,F401
+from .workflows.command_remove import workflow_remove  # noqa: E402,F401
+from .workflows.step.command_add import workflow_step_add  # noqa: E402,F401
+from .workflows.step.command_remove import workflow_step_remove  # noqa: E402,F401
+
 
 def main():
     # On Windows the default stdout/stderr code page (e.g. cp1252) cannot encode

src/specify_cli/workflows/__init__.py

@@ -44,18 +44,18 @@ def get_step_type(type_key: str) -> StepBase | None:
 
 def _register_builtin_steps() -> None:
     """Register all built-in step types."""
-    from .steps.command import CommandStep
-    from .steps.do_while import DoWhileStep
-    from .steps.fan_in import FanInStep
-    from .steps.fan_out import FanOutStep
-    from .steps.gate import GateStep
-    from .steps.if_then import IfThenStep
-    from .steps.init import InitStep
-    from .steps.prompt import PromptStep
-    from .steps.shell import ShellStep
-    from .steps.slot import SlotStep
-    from .steps.switch import SwitchStep
-    from .steps.while_loop import WhileStep
+    from .step.command import CommandStep
+    from .step.do_while import DoWhileStep
+    from .step.fan_in import FanInStep
+    from .step.fan_out import FanOutStep
+    from .step.gate import GateStep
+    from .step.if_then import IfThenStep
+    from .step.init import InitStep
+    from .step.prompt import PromptStep
+    from .step.shell import ShellStep
+    from .step.slot import SlotStep
+    from .step.switch import SwitchStep
+    from .step.while_loop import WhileStep
 
     _register_step(CommandStep())
     _register_step(DoWhileStep())

src/specify_cli/workflows/_command_resume_state.py

@@ -0,0 +1,49 @@
+"""Resume-private installed-workflow owner state resolution."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+
+def _path_has_symlink_component(path: Path) -> bool:
+    """Return whether any component of an absolute path is a symlink."""
+    absolute = Path(os.path.abspath(path))
+    current = Path(absolute.anchor)
+    for part in absolute.parts[1:]:
+        current /= part
+        if current.is_symlink():
+            return True
+    return False
+
+
+def _resolve_run_owner_root(
+    installed_registry_root: str | None, project_root: Path
+) -> Path:
+    """Determine which project's registry gates resuming a run.
+
+    ``installed_registry_root`` is only ever persisted when the run's
+    installed workflow genuinely belongs to a *different* project than the
+    one whose ``runs/`` directory holds this run's own state (a direct
+    external workflow-file invocation) -- see ``workflow_run``. The common
+    case (an installed workflow run from its own project) stores ``None``,
+    so a later project rename/move is transparently picked up here by
+    falling back to the *current* ``project_root`` instead of a stale
+    absolute path baked in at run start.
+
+    A persisted cross-project root that no longer exists cannot be safely
+    rediscovered and must fail closed instead of consulting the unrelated
+    project that happens to store the run state.
+    """
+    if installed_registry_root:
+        candidate = Path(installed_registry_root)
+        if (
+            candidate.is_absolute()
+            and not _path_has_symlink_component(candidate)
+            and candidate.is_dir()
+        ):
+            return candidate
+        raise ValueError(
+            "Installed workflow owner is unavailable; cannot safely resume"
+        )
+    return project_root

src/specify_cli/workflows/_command_run_ownership.py

@@ -0,0 +1,187 @@
+"""Run-private installed-workflow ownership resolution."""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from . import _commands as cli
+
+
+def _same_existing_path(left: Path, right: Path) -> bool:
+    """Return whether two existing paths identify the same filesystem entry."""
+    try:
+        return os.path.samefile(left, right)
+    except OSError:
+        return left == right
+
+
+def _scan_for_workflow_owner(parts: tuple[str, ...]) -> int | None:
+    """Find the *nearest* (innermost) ``.specify/workflows/<id>`` owner in
+    *parts*, scanning from the end of the path.
+
+    Scanning from the end (rather than stopping at the first match from the
+    start) matters for a project nested beneath an unrelated outer path that
+    happens to reuse the same ``.specify``/``workflows`` segment names: the
+    first-from-start match would pick the outer directory and the wrong
+    workflow ID, silently missing the real (inner) owner's disabled check.
+
+    Returns the index of the owning ``.specify`` segment, or ``None`` if no
+    owner segment is present.
+    """
+    for i in range(len(parts) - 3, -1, -1):
+        if (
+            parts[i].casefold() == ".specify"
+            and parts[i + 1].casefold() == "workflows"
+        ):
+            return i
+    return None
+
+
+def _expand_first_symlink_target(path: Path) -> Path | None:
+    """Expand one symlink component while preserving the remaining path."""
+    parts = path.parts
+    current = Path(path.anchor) if path.is_absolute() else Path()
+    start = 1 if path.is_absolute() else 0
+    for index in range(start, len(parts)):
+        current = current / parts[index]
+        if not current.is_symlink():
+            continue
+        try:
+            target = Path(os.readlink(current))
+        except OSError:
+            return None
+        if not target.is_absolute():
+            target = current.parent / target
+        expanded = target.joinpath(*parts[index + 1 :])
+        return Path(os.path.normpath(str(expanded.absolute())))
+    return None
+
+
+def _resolve_installed_workflow_ownership(
+    source_path: Path, err
+) -> tuple[Path | None, str | None]:
+    """Map a direct ``workflow.yml`` *source_path* back to the installed
+    workflow (``registry_root``, ``registered_id``) it belongs to, if any.
+
+    A registered path can point at installed storage three ways, all of
+    which must receive the same registry disabled-check:
+
+    1. Lexically: the path's own (symlink-preserving) segments identify
+       ``.specify/workflows/<id>`` -- collapsing ``..``/``.`` but
+       never resolving symlinks, so a symlinked ``workflow.yml`` leaf (or
+       symlinked ``<id>`` directory) inside the owned tree is caught by the
+       inward-symlink refusal below rather than silently followed.
+    2. Via an intermediate alias target whose lexical path identifies
+       ``.specify/workflows/<id>`` before a symlinked storage ancestor is
+       resolved away.
+    3. Via an outward-pointing alias whose fully resolved target lands
+       inside legitimate installed storage, even though the raw invocation
+       path has no ownership segments.
+
+    Returns ``(None, None)`` when neither applies -- a genuinely standalone
+    external workflow file, which is allowed to run unchecked.
+    """
+    def ownership_for(candidate: Path) -> tuple[Path, str] | None:
+        parts = candidate.parts
+        i = _scan_for_workflow_owner(parts)
+        if i is None:
+            return None
+        registry_root = (
+            Path(*parts[:i]) if i else Path(candidate.anchor or ".")
+        )
+        candidate_specify = Path(*parts[: i + 1])
+        candidate_workflows = Path(*parts[: i + 2])
+        candidate_id_dir = Path(*parts[: i + 3])
+        canonical_specify = registry_root / ".specify"
+        canonical_workflows = canonical_specify / "workflows"
+        # The path-derived registry_root here may differ from the cwd's
+        # project_root already checked by _reject_unsafe_workflow_storage
+        # (e.g. this path points into another project entirely, or this
+        # project's own .specify is itself a symlink to an
+        # attacker-controlled tree) -- check it explicitly rather than
+        # trusting that cwd-scoped guard, and don't rely on
+        # WorkflowRegistry's own symlinked-parent handling as the safety
+        # signal here: it fails closed by raising OSError at construction
+        # time (see catalog.py's _load), but that surfaces as an opaque
+        # exception rather than this guard's clean, specific CLI error for
+        # the actual owning project root.
+        cli._reject_unsafe_dir(canonical_specify, ".specify")
+        cli._reject_unsafe_dir(canonical_workflows, ".specify/workflows")
+        cli._reject_unsafe_dir(candidate_specify, ".specify")
+        cli._reject_unsafe_dir(candidate_workflows, ".specify/workflows")
+        try:
+            if not os.path.samefile(candidate_specify, canonical_specify):
+                return None
+            if not os.path.samefile(
+                candidate_workflows, canonical_workflows
+            ):
+                return None
+        except OSError:
+            return None
+        registry = cli._open_workflow_registry(registry_root, err)
+        registered_id = None
+        for workflow_id in registry.list():
+            if (
+                not isinstance(workflow_id, str)
+                or workflow_id in cli._RESERVED_WORKFLOW_IDS
+                or not cli._WORKFLOW_ID_PATTERN.fullmatch(workflow_id)
+            ):
+                continue
+            try:
+                if os.path.samefile(
+                    candidate_id_dir,
+                    canonical_workflows / workflow_id,
+                ):
+                    registered_id = workflow_id
+                    break
+            except OSError:
+                continue
+        if registered_id is None:
+            return None
+        # A legitimately installed workflow's own directory tree never
+        # contains a symlink (workflow add/remove both refuse one at
+        # install time); one appearing here means the file actually loaded
+        # below would not be the file this ownership match is based on, so
+        # refuse rather than silently mismatch.
+        for k in range(i + 2, len(parts) + 1):
+            if Path(*parts[:k]).is_symlink():
+                err.print(
+                    "[red]Error:[/red] Refusing to run: "
+                    f".specify/workflows/{cli._escape_markup(registered_id)} "
+                    "contains a symlinked path component"
+                )
+                raise cli.typer.Exit(1)
+        return registry_root, registered_id
+
+    lexical = Path(os.path.normpath(str(source_path.absolute())))
+    ownership = ownership_for(lexical)
+    if ownership is not None:
+        return ownership
+
+    # Inspect each intermediate symlink target before fully resolving it.
+    # Full resolution can erase .specify/workflows ownership segments when
+    # one of those storage directories is itself a symlink.
+    candidate = lexical
+    seen = {candidate}
+    for _ in range(40):
+        expanded = _expand_first_symlink_target(candidate)
+        if expanded is None or expanded in seen:
+            break
+        ownership = ownership_for(expanded)
+        if ownership is not None:
+            return ownership
+        seen.add(expanded)
+        candidate = expanded
+
+    # A fully resolved target may still land in legitimate installed
+    # storage through an unrelated-looking alias.
+    try:
+        resolved = source_path.resolve(strict=False)
+    except (OSError, RuntimeError):
+        return None, None
+    if resolved == lexical:
+        # Nothing on this path is a symlink; already covered above.
+        return None, None
+    ownership = ownership_for(resolved)
+    return ownership if ownership is not None else (None, None)