flow/docs/superpowers/specs/2026-03-16-flow-architecture-redesign.md

Flow CLI Architecture Redesign

Overview

Full rewrite of the flow CLI: a personal dev environment tool managing the host -> VM -> container workflow stack. Single Python binary (PyInstaller), context-aware (host/vm/container), no backward compatibility constraints.

Goals

• Correct abstractions with pure domain logic separated from I/O
• Plan-then-execute pattern for all mutating operations (native dry-run)
• No code duplication across domains
• Every domain unit independently testable without mocks
• Module path resolution works correctly for _module.yaml-backed packages

---

1. Layer Architecture

Four layers with strict dependency direction (each layer only imports from layers below):

commands/          Layer 4: CLI parsing + dispatch (no logic)
    |
services/          Layer 3: Orchestration + side effects (thin)
    |
domain/            Layer 2: Pure logic, models, planning (no I/O)
    |
core/              Layer 1: Runtime primitives + config loading

Rules

• domain/ never imports core/runtime, services/, or commands/. It only imports core/template.py, core/errors.py, core/paths.py (constants only), and stdlib.
• services/ receives all dependencies via constructor (runtime, config, console). No module-level singletons.
• commands/ are trivial: parse args into a typed namespace, call one service method.
• Every plan is a frozen dataclass returned by domain functions. Services decide whether to execute or dry-run print them.

File Layout

src/flow/
  __init__.py
  __main__.py
  cli.py                          # argparse, context creation, error handling

  core/
    runtime.py                    # CommandRunner, FileSystem, GitClient, SystemRuntime
    config.py                     # YAML loading, AppConfig, FlowContext
    platform.py                   # OS/arch detection, execution context (host/vm/container)
    paths.py                      # XDG path constants
    console.py                    # Output formatting with TTY detection, --no-color, --quiet
    errors.py                     # FlowError hierarchy
    template.py                   # Variable/template substitution (pure)

  domain/
    dotfiles/
      models.py                   # Package, ModuleRef, LinkTarget, LinkedState
      resolution.py               # Path resolution: package -> home-relative targets
      modules.py                  # Module source normalization, cache path computation
      planning.py                 # Compute LinkPlan from desired vs current state
      conflicts.py                # Conflict detection

    packages/
      models.py                   # PackageDef, ProfilePackageRef, InstalledState, InstalledPackage
      catalog.py                  # Manifest parsing, profile entry normalization, spec resolution
      resolution.py               # Source name resolution, platform mapping, URL building
      planning.py                 # Compute PackagePlan (install/remove operations)

    bootstrap/
      models.py                   # Profile, SetupModuleDef, BootstrapPlan
      planning.py                 # Profile -> ordered action list (packages + modules + dotfiles)
      modules.py                  # Built-in setup module definitions (hostname, locale, shell, ssh-keygen, runcmd)

    remote/
      models.py                   # Target, SSHCommand
      resolution.py               # Target parsing, host template expansion

    containers/
      models.py                   # ContainerSpec, ContainerState
      resolution.py               # Image ref parsing, name normalization, mount computation

  services/
    dotfiles.py                   # DotfilesService
    packages.py                   # PackageService
    bootstrap.py                  # BootstrapService
    remote.py                     # RemoteService
    containers.py                 # ContainerService
    projects.py                   # ProjectService

  commands/
    __init__.py
    remote.py
    dev.py
    dotfiles.py
    setup.py
    packages.py
    projects.py
    completion.py

---

2. Command Surface

Context awareness

Flow detects execution context via environment variables (DF_NAMESPACE, DF_PLATFORM) and restricts commands accordingly:

Context	Detection	Available commands
Host	No DF_NAMESPACE, no DF_PLATFORM	remote, dotfiles, setup, packages
VM	DF_NAMESPACE set, no container indicators	dev, projects, dotfiles, setup, packages
Container	Inside container (e.g., /.dockerenv exists)	dotfiles, setup, packages

Unavailable commands print a clear message (e.g., "flow remote is only available on the host machine").

Commands

Canonical names are plural where they manage collections. Short aliases in parentheses.

flow remote enter <target>              # Host only
flow remote list

flow dev create <name> -i <image>       # VM only
flow dev attach <name>
flow dev exec <name> [cmd...]
flow dev list
flow dev stop <name>
flow dev rm <name>
flow dev respawn <name>

flow dotfiles init --repo <url>         # (dot) Everywhere
flow dotfiles link [--profile p]
flow dotfiles unlink [packages...]
flow dotfiles status [packages...]
flow dotfiles edit <package>

flow dotfiles repos list                # (repo)
flow dotfiles repos status [--repo=x]
flow dotfiles repos pull [--repo=x]
flow dotfiles repos push [--repo=x]

flow setup run [--profile p]            # (bootstrap) Everywhere
flow setup list
flow setup show <profile>

flow packages install <name...>         # (package, pkg) Everywhere
flow packages list [--all]
flow packages remove <name...>

flow projects check [--fetch]           # (project) VM only
flow projects fetch
flow projects summary

Global flags

flow --version
flow --quiet           # Suppress info messages
flow --no-color        # Disable ANSI colors
flow --dry-run         # Available on all mutating commands (passed through to service)

--dry-run as a global flag means the plan-then-execute pattern works uniformly: every service method that mutates state accepts dry_run: bool and either executes the plan or prints it.

---

3. Core Layer

3.1 Errors

Small hierarchy. Every domain function raises FlowError subtypes. cli.py catches FlowError at the top level.

class FlowError(Exception):
    """Base for all user-facing errors."""

class ConfigError(FlowError):
    """Invalid config/manifest YAML."""

class PlanConflict(FlowError):
    """Conflicts detected during planning."""
    def __init__(self, message: str, conflicts: list[str]):
        super().__init__(message)
        self.conflicts = conflicts

class ExecutionError(FlowError):
    """A plan step failed during execution."""

No bare RuntimeError anywhere in the codebase.

3.2 Console

class Console:
    def __init__(self, *, quiet: bool = False, color: bool | None = None):
        # color=None means auto-detect via os.isatty(1)

    def info(self, msg: str) -> None: ...
    def warn(self, msg: str) -> None: ...
    def error(self, msg: str) -> None: ...
    def success(self, msg: str) -> None: ...
    def table(self, headers: list[str], rows: list[list[str]]) -> None: ...
    def print_plan(self, operations: list[Any], *, verb: str = "execute") -> None: ...

3.3 Runtime

class CommandRunner:
    def run(self, argv, *, cwd, env, capture_output, check, timeout) -> CompletedProcess: ...
    def run_shell(self, command, *, cwd, env, capture_output, check) -> CompletedProcess: ...
    def stream_shell(self, command, console, *, check) -> CompletedProcess: ...
    def require_binary(self, name) -> str: ...

class FileSystem:
    def ensure_dir(self, path, *, sudo, runner, mode) -> None: ...
    def remove_file(self, path, *, sudo, runner, missing_ok) -> None: ...
    def remove_tree(self, path) -> None: ...
    def copy_file(self, source, target, *, sudo, runner) -> None: ...
    def copy_tree(self, source, target) -> None: ...
    def create_symlink(self, source, target, *, sudo, runner) -> None: ...
    def same_symlink(self, target, source) -> bool: ...
    def read_text(self, path, *, default) -> str: ...
    def write_text(self, path, content) -> None: ...
    def read_json(self, path, *, default) -> Any: ...
    def write_json(self, path, data) -> None: ...

class GitClient:
    def __init__(self, runner: CommandRunner): ...
    def run(self, repo_dir, *args, capture_output, check) -> CompletedProcess: ...

@dataclass
class SystemRuntime:
    runner: CommandRunner
    fs: FileSystem
    git: GitClient

3.4 FlowContext

@dataclass
class FlowContext:
    config: AppConfig
    manifest: dict[str, Any]
    platform: PlatformInfo
    console: Console
    runtime: SystemRuntime

Created once in cli.py, passed to services via constructor. Never accessed as a global.

3.5 Template

Pure functions. No I/O.

def substitute(text: str, variables: dict[str, str]) -> str:
    """Replace $VAR and ${VAR} with values."""

def substitute_template(text: str, context: dict[str, Any]) -> str:
    """Replace {{expr}} placeholders."""

---

4. Dotfiles Domain

4.1 Models

@dataclass(frozen=True)
class Package:
    """A dotfiles package: a named set of files mapping to home-relative targets."""
    name: str                          # e.g. "zsh", "nvim"
    layer: str                         # "_shared" or profile name
    source_dir: Path                   # Absolute path in dotfiles repo
    module: ModuleRef | None           # If backed by external repo

@dataclass(frozen=True)
class ModuleRef:
    """An external git repo providing content for a package subtree."""
    source: str                        # Normalized git URL
    ref_type: str                      # "branch" | "tag" | "commit"
    ref_value: str                     # e.g. "main", "v1.0", "abc123"
    mount_path: Path                   # Relative path within package to _module.yaml parent
                                       # e.g. Path(".config/nvim")
    cache_dir: Path                    # ~/.local/share/flow/modules/<name>

@dataclass(frozen=True)
class LinkTarget:
    """A single file that should be linked into the filesystem."""
    source: Path                       # Where file lives (dotfiles repo or module cache)
    target: Path                       # Where symlink goes (e.g. ~/.config/nvim/init.lua)
    package: str                       # Owning package (e.g. "_shared/nvim")
    from_module: bool                  # Whether source is from a module repo
    needs_sudo: bool                   # True for _root/ targets outside $HOME

@dataclass(frozen=True)
class LinkOp:
    """A single operation in a link plan."""
    type: str                          # "create_link" | "remove_link" | "create_dir"
    target: Path
    source: Path | None
    package: str
    needs_sudo: bool

@dataclass(frozen=True)
class PlanSummary:
    added: int
    removed: int
    unchanged: int
    from_modules: int

@dataclass(frozen=True)
class LinkPlan:
    """Complete reconciliation plan."""
    operations: list[LinkOp]
    conflicts: list[str]
    summary: PlanSummary

@dataclass
class LinkedState:
    """Persisted to ~/.local/state/flow/linked.json."""
    links: dict[Path, LinkTarget]

    def as_dict(self) -> dict: ...
    @classmethod
    def from_dict(cls, data: dict) -> LinkedState: ...

@dataclass(frozen=True)
class RepoInfo:
    """A managed git repo (dotfiles or module)."""
    name: str                          # "dotfiles" or module package name
    path: Path                         # Local clone path
    source: str                        # Remote URL
    is_module: bool

4.2 Module Path Resolution

The core algorithm that fixes the current bug. Given a dotfiles repo layout:

_shared/
  nvim/                              # Package: "nvim", layer: "_shared"
    .config/nvim/
      _module.yaml                   # Module definition
    .local/bin/nvim-wrapper          # Normal local file

Resolution steps:

1. Discover packages: scan _shared/ and profile dirs for first-level subdirectories.
2. For each package, walk its directory tree:
   • Find _module.yaml if present. Its parent relative to the package root is the mount_path.
   • For _shared/nvim/.config/nvim/_module.yaml: mount_path = .config/nvim
3. Resolve files to LinkTargets:
   • Files outside mount_path come from the dotfiles repo directly.
     • .local/bin/nvim-wrapper -> LinkTarget(source=<dotfiles>/_shared/nvim/.local/bin/nvim-wrapper, target=~/.local/bin/nvim-wrapper)
   • Files inside mount_path come from the module cache.
     • The module repo is cloned to ~/.local/share/flow/modules/nvim.
     • Every file in the clone maps to ~/<mount_path>/<relative>.
     • modules/nvim/init.lua -> LinkTarget(source=<modules>/nvim/init.lua, target=~/.config/nvim/init.lua)
   • The _module.yaml file itself is never linked.

# domain/dotfiles/resolution.py

def discover_packages(dotfiles_dir: Path, profile: str | None) -> list[Package]: ...

def resolve_package_targets(
    package: Package,
    home: Path,
    skip: set[str],
) -> list[LinkTarget]:
    """Resolve all LinkTargets for a package, handling modules correctly."""
    ...

def resolve_all_targets(
    packages: list[Package],
    home: Path,
    skip: set[str],
) -> list[LinkTarget]:
    """Resolve targets for all packages. Raises on conflicts."""
    ...

# domain/dotfiles/modules.py

def parse_module_yaml(path: Path) -> ModuleRef: ...

def compute_mount_path(module_yaml: Path, package_dir: Path) -> Path:
    """The key function: relative path from package root to _module.yaml parent."""
    return module_yaml.parent.relative_to(package_dir)

def module_cache_dir(package_name: str, modules_base: Path) -> Path:
    return modules_base / package_name.replace("/", "--")

def normalize_source(source: str) -> str:
    """github:org/repo -> https://github.com/org/repo.git"""
    ...

4.3 Planning

# domain/dotfiles/planning.py

def plan_link(
    desired: list[LinkTarget],
    current: LinkedState,
) -> LinkPlan:
    """Compare desired targets with current state, produce reconciliation plan.

    - New targets: create_link ops
    - Removed targets: remove_link ops
    - Existing correct symlinks: unchanged (no op)
    - Conflicts (target exists, not managed): listed in plan.conflicts
    """
    ...

def plan_unlink(
    current: LinkedState,
    packages: list[str] | None,        # None = unlink all
) -> LinkPlan:
    """Plan removal of managed links."""
    ...

4.4 Conflict Detection

# domain/dotfiles/conflicts.py

def detect_conflicts(
    desired: list[LinkTarget],
    current: LinkedState,
    filesystem_check: Callable[[Path], str | None],
    # Returns "file", "dir", "symlink", or None
    # This is the ONLY I/O dependency, injected from service layer
) -> list[str]:
    """Return human-readable conflict descriptions."""
    ...

The filesystem_check callback is the boundary between pure domain and I/O. The service injects it. Tests provide a fake.

---

5. Packages Domain

Shared by flow packages and flow setup. Single source of truth.

5.1 Models

@dataclass(frozen=True)
class PackageDef:
    """A package as defined in the manifest."""
    name: str
    type: str                          # "pkg" | "binary" | "cask"
    sources: dict[str, str]            # {"apt": "fd-find", "brew": "fd"}
    source: str | None                 # Binary: "github:neovim/neovim"
    version: str | None                # Binary: "0.10.4"
    asset_pattern: str | None
    platform_map: dict[str, dict]
    extract_dir: str | None
    install: dict[str, list[str]]      # {"bin": ["bin/nvim"], "share": [...]}
    post_install: str | None
    allow_sudo: bool

@dataclass(frozen=True)
class ProfilePackageRef:
    """A package reference from a profile's package list."""
    name: str
    type: str | None
    allow_sudo: bool
    post_install: str | None

@dataclass(frozen=True)
class PkgInstallOp:
    type: str                          # "pm_update" | "pm_install" | "binary_install" | "run_hook"
    package: str | None
    command: str | None                # Shell command for pm_update/pm_install/run_hook
    download_url: str | None           # For binary_install
    install_map: dict[str, list[str]] | None
    description: str                   # Human-readable for dry-run

@dataclass(frozen=True)
class PkgRemoveOp:
    package: str
    files: list[Path]                  # Files to delete

@dataclass(frozen=True)
class PackagePlan:
    operations: list[PkgInstallOp | PkgRemoveOp]
    summary: str

@dataclass(frozen=True)
class InstalledPackage:
    name: str
    version: str
    type: str
    files: list[Path]                  # Track installed files for real removal

@dataclass
class InstalledState:
    packages: dict[str, InstalledPackage]

    def as_dict(self) -> dict: ...
    @classmethod
    def from_dict(cls, data: dict) -> InstalledState: ...

5.2 Pure Functions

# domain/packages/catalog.py
def parse_catalog(raw: Any) -> dict[str, PackageDef]: ...
def normalize_profile_entry(entry: Any) -> ProfilePackageRef: ...
def resolve_spec(catalog: dict[str, PackageDef], ref: ProfilePackageRef) -> PackageDef: ...

# domain/packages/resolution.py
def resolve_source_name(spec: PackageDef, pm: str) -> str: ...
def resolve_binary_asset(spec: PackageDef, platform: PlatformInfo) -> str: ...
def resolve_download_url(spec: PackageDef, asset: str, template_ctx: dict) -> str: ...
def detect_package_manager(os_name: str) -> str | None: ...
def pm_update_command(pm: str) -> str: ...
def pm_install_command(pm: str, packages: list[str], pkg_type: str) -> str: ...

# domain/packages/planning.py
def plan_install(
    specs: list[PackageDef],
    pm: str,
    platform: PlatformInfo,
    template_ctx: dict,
) -> PackagePlan: ...

def plan_remove(
    names: list[str],
    installed: InstalledState,
) -> PackagePlan: ...

---

6. Bootstrap Domain

Bootstrap is an orchestrator. It does not have unique domain logic -- it composes packages, dotfiles, and setup modules into an ordered plan.

6.1 Models

@dataclass(frozen=True)
class Profile:
    """A bootstrap profile from the manifest."""
    name: str
    os: str                            # "linux" | "macos"
    package_manager: str | None        # None = auto-detect
    packages: list[Any]                # Raw entries, normalized via packages domain
    modules: list[dict]                # Setup module definitions
    requires: list[str]                # Required env vars
    shell: str | None                  # e.g. "zsh"
    dotfiles_profile: str | None       # Profile name for dotfiles linking

@dataclass(frozen=True)
class BootstrapAction:
    """A single step in the bootstrap plan."""
    type: str                          # "validate_env" | "install_packages" | "run_module"
                                       # | "link_dotfiles" | "set_shell"
    description: str
    payload: Any                       # Type-specific data:
                                       #   install_packages -> PackagePlan
                                       #   run_module -> SetupModuleDef
                                       #   link_dotfiles -> profile name
                                       #   set_shell -> shell name
    critical: bool                     # Stop on failure?

@dataclass(frozen=True)
class BootstrapPlan:
    profile: str
    actions: list[BootstrapAction]
    summary: str

6.2 Setup Modules

Each module type is a small class with a plan method (pure) that returns shell commands to execute.

class SetupModule(Protocol):
    def plan(self, config: dict, template_ctx: dict) -> list[str]:
        """Return shell commands to execute. Pure -- no side effects."""
        ...

    def describe(self) -> str:
        """Human-readable description for dry-run output."""
        ...

Built-in modules in domain/bootstrap/modules.py:

• HostnameModule -- sets hostname via hostnamectl (linux) or scutil (macos)
• LocaleModule -- sets locale via locale-gen + update-locale (linux only)
• ShellModule -- installs shell if missing, adds to /etc/shells, runs chsh
• SSHKeygenModule -- generates SSH keys per spec
• RuncmdModule -- runs arbitrary shell commands from profile

6.3 Planning

# domain/bootstrap/planning.py

def parse_profile(name: str, raw: dict) -> Profile: ...

def plan_bootstrap(
    profile: Profile,
    catalog: dict[str, PackageDef],
    platform: PlatformInfo,
    env: dict[str, str],
    template_ctx: dict,
) -> BootstrapPlan:
    """Build the full ordered action list:
    1. Validate required env vars
    2. Setup modules (hostname, locale, etc.)
    3. Install packages (delegates to packages domain)
    4. Set shell
    5. SSH keygen
    6. Runcmd
    7. Link dotfiles
    """
    ...

6.4 Config Format

profiles:
  linux-work:
    os: linux
    shell: zsh
    dotfiles-profile: linux-work
    requires: [USER_EMAIL]
    modules:
      - type: hostname
        value: "{{ env.HOSTNAME }}"
      - type: locale
        value: en_US.UTF-8
      - type: ssh-keygen
        keys:
          - type: ed25519
            comment: "{{ env.USER_EMAIL }}"
      - type: runcmd
        commands:
          - "sudo groupadd docker || true"
          - "sudo usermod -aG docker $USER"
    packages:
      - git
      - fd
      - binary/neovim

---

7. Remote Domain

7.1 Models

@dataclass(frozen=True)
class Target:
    """A resolved SSH target."""
    user: str
    namespace: str
    platform: str
    host: str
    identity: str | None

@dataclass(frozen=True)
class SSHCommand:
    """A fully resolved SSH command ready to exec."""
    argv: list[str]
    destination: str
    tmux_session: str | None
    env_vars: dict[str, str]           # DF_NAMESPACE, DF_PLATFORM

7.2 Resolution

# domain/remote/resolution.py

HOST_TEMPLATES = {
    "orb": "<namespace>.orb",
    "utm": "<namespace>.utm.local",
    "core": "<namespace>.core.lan",
}

def parse_target(target: str) -> tuple[str | None, str | None, str | None]:
    """Parse [user@]namespace@platform."""
    ...

def resolve_target(
    parsed: tuple,
    config_targets: list[TargetConfig],
    default_user: str,
) -> Target: ...

def build_ssh_command(
    target: Target,
    *,
    tmux_session: str | None,
    no_tmux: bool,
) -> SSHCommand: ...

def terminfo_fix_command(term: str | None, destination: str) -> str | None: ...

---

8. Containers Domain

8.1 Models

@dataclass(frozen=True)
class ImageRef:
    """A resolved container image reference."""
    full_ref: str
    registry: str
    repo: str
    tag: str
    label: str

@dataclass(frozen=True)
class ContainerSpec:
    """Everything needed to create a container."""
    name: str                          # dev-<name>
    image: ImageRef
    project_path: Path | None
    mounts: list[tuple[str, str]]
    labels: dict[str, str]
    network: str                       # "host"

@dataclass(frozen=True)
class ContainerInfo:
    """Runtime state of an existing container."""
    name: str
    image: str
    project: str
    status: str

8.2 Resolution

# domain/containers/resolution.py

def parse_image_ref(
    image: str,
    *,
    default_registry: str,
    default_tag: str,
) -> ImageRef: ...

def container_name(name: str) -> str:
    """Ensure dev- prefix."""
    ...

def resolve_mounts(home: str) -> list[tuple[str, str]]:
    """Standard host mounts: .ssh (ro), .npmrc (ro), .npm, docker socket."""
    ...

def build_container_spec(
    name: str,
    image: ImageRef,
    project_path: Path | None,
    home: str,
) -> ContainerSpec: ...

---

9. Services Layer

Each service is a class receiving dependencies via constructor. All mutating methods accept dry_run: bool.

9.1 DotfilesService

class DotfilesService:
    def __init__(self, ctx: FlowContext): ...

    def init(self, repo_url: str) -> None:
        """Clone dotfiles repo + discover and clone all module repos."""

    def link(self, *, profile: str | None, packages: list[str] | None,
             force: bool, dry_run: bool) -> None:
        """Reconcile links: discover -> resolve -> plan -> execute."""

    def unlink(self, packages: list[str] | None, *, dry_run: bool) -> None:
        """Remove managed links."""

    def status(self, packages: list[str] | None) -> None:
        """Show package list, link health, module info."""

    def edit(self, target: str) -> None:
        """Pull relevant repo -> open editor -> commit + push."""

    def repos_list(self) -> None: ...
    def repos_status(self, repo_filter: str | None) -> None: ...
    def repos_pull(self, repo_filter: str | None) -> None: ...
    def repos_push(self, repo_filter: str | None) -> None: ...

Link flow detail

def link(self, ...):
    packages = discover_packages(dotfiles_dir, profile)     # domain
    targets = resolve_all_targets(packages, home, skip)     # domain
    current = self._load_linked_state()                     # I/O
    plan = plan_link(targets, current)                      # domain

    if plan.conflicts and not force:
        raise PlanConflict(...)

    if dry_run:
        self.console.print_plan(plan.operations)
        return

    self._execute_link_plan(plan)                           # I/O
    self._save_linked_state(...)                            # I/O

9.2 PackageService

class PackageService:
    def __init__(self, ctx: FlowContext): ...

    def install(self, names: list[str], *, dry_run: bool) -> None: ...
    def list(self, *, show_all: bool) -> None: ...
    def remove(self, names: list[str], *, dry_run: bool) -> None: ...

9.3 BootstrapService

class BootstrapService:
    def __init__(self, ctx: FlowContext): ...

    def run(self, *, profile: str | None, variables: dict[str, str],
            dry_run: bool) -> None:
        """Parse profile -> build plan -> execute actions sequentially."""

    def list(self) -> None: ...
    def show(self, profile: str) -> None: ...

Run flow detail

def run(self, ...):
    profile = parse_profile(name, raw)                      # domain
    catalog = parse_catalog(manifest["packages"])            # domain/packages
    plan = plan_bootstrap(profile, catalog, platform, ...)   # domain

    if dry_run:
        self.console.print_plan(plan.actions)
        return

    for action in plan.actions:
        match action.type:
            case "install_packages":
                self._execute_package_plan(action.payload)
            case "run_module":
                self._execute_setup_module(action.payload)
            case "link_dotfiles":
                DotfilesService(self.ctx).link(profile=action.payload, dry_run=False)
            case "set_shell":
                self._set_shell(action.payload)

9.4 RemoteService

class RemoteService:
    def __init__(self, ctx: FlowContext): ...

    def enter(self, target_str: str, *, user: str | None, namespace: str | None,
              platform: str | None, session: str, no_tmux: bool,
              dry_run: bool) -> None: ...

    def list(self) -> None: ...

9.5 ContainerService

class ContainerService:
    def __init__(self, ctx: FlowContext): ...

    def create(self, name: str, image: str, project: str | None, *,
               dry_run: bool) -> None: ...
    def attach(self, name: str) -> None: ...
    def exec(self, name: str, cmd: list[str] | None) -> None: ...
    def list(self) -> None: ...
    def stop(self, name: str, *, kill: bool) -> None: ...
    def remove(self, name: str, *, force: bool) -> None: ...
    def respawn(self, name: str) -> None: ...

9.6 ProjectService

class ProjectService:
    def __init__(self, ctx: FlowContext): ...

    def check(self, *, fetch: bool) -> None: ...
    def fetch(self) -> None: ...
    def summary(self) -> None: ...

---

10. CLI Entry Point

# cli.py

def main():
    parser = build_parser()
    args = parser.parse_args()

    console = Console(
        quiet=args.quiet,
        color=not args.no_color if args.no_color else None,
    )

    context_type = detect_context()
    validate_command_context(args.command, context_type, console)
    ensure_non_root(console)

    platform = detect_platform()
    config = load_config()
    manifest = load_manifest()
    runtime = SystemRuntime()

    ctx = FlowContext(config, manifest, platform, console, runtime)

    try:
        args.handler(ctx, args)
    except PlanConflict as e:
        for conflict in e.conflicts:
            console.error(conflict)
        console.error(str(e))
        sys.exit(1)
    except FlowError as e:
        console.error(str(e))
        sys.exit(1)

Command modules are trivial:

# commands/dotfiles.py

def register(subparsers):
    p = subparsers.add_parser("dotfiles", aliases=["dot"], help="Manage dotfiles")
    sub = p.add_subparsers(dest="dotfiles_command")

    link = sub.add_parser("link", help="Reconcile dotfile symlinks")
    link.add_argument("--profile")
    link.add_argument("packages", nargs="*")
    link.add_argument("--force", action="store_true")
    link.set_defaults(handler=_run_link)
    # ...

def _run_link(ctx, args):
    DotfilesService(ctx).link(
        profile=args.profile,
        packages=args.packages or None,
        force=args.force,
        dry_run=args.dry_run,
    )

---

11. Config Format

Single unified YAML format, loaded from ~/.config/flow/ or self-hosted from dotfiles repo.

repository:
  url: git@github.com:user/dotfiles.git
  branch: main

paths:
  projects: ~/projects

defaults:
  container-registry: registry.tomastm.com
  container-tag: latest
  tmux-session: default

targets:
  personal@orb: personal.orb
  work@ec2:
    host: work.ec2.internal
    identity: ~/.ssh/id_work

packages:
  - name: fd
    type: pkg
    sources:
      apt: fd-find
      brew: fd

  - name: neovim
    type: binary
    source: github:neovim/neovim
    version: "0.10.4"
    asset-pattern: "nvim-{{os}}-{{arch}}.tar.gz"
    platform-map:
      linux-x64: { os: linux, arch: x64 }
      darwin-arm64: { os: macos, arch: arm64 }
    extract-dir: "nvim-{{os}}64"
    install:
      bin: [bin/nvim]
      share: [share/nvim]

profiles:
  linux-work:
    os: linux
    shell: zsh
    dotfiles-profile: linux-work
    requires: [USER_EMAIL]
    modules:
      - type: hostname
        value: "{{ env.HOSTNAME }}"
      - type: locale
        value: en_US.UTF-8
      - type: ssh-keygen
        keys:
          - type: ed25519
            comment: "{{ env.USER_EMAIL }}"
    packages:
      - git
      - fd
      - binary/neovim

---

12. Testing Strategy

Domain tests (majority of tests)

Pure function tests. No mocks, no filesystem, no subprocess. Fast.

def test_module_mount_path():
    mount = compute_mount_path(
        module_yaml=Path("/dots/_shared/nvim/.config/nvim/_module.yaml"),
        package_dir=Path("/dots/_shared/nvim"),
    )
    assert mount == Path(".config/nvim")

def test_plan_link_creates_ops_for_new_targets():
    desired = [LinkTarget(source=Path("/a"), target=Path("/home/x/.zshrc"),
                          package="_shared/zsh", from_module=False, needs_sudo=False)]
    current = LinkedState(links={})
    plan = plan_link(desired, current)
    assert len(plan.operations) == 1
    assert plan.operations[0].type == "create_link"

def test_resolve_binary_url():
    spec = PackageDef(name="nvim", type="binary",
                      source="github:neovim/neovim", version="0.10.4", ...)
    url = resolve_download_url(spec, "nvim-linux-x64.tar.gz", {...})
    assert url == "https://github.com/neovim/neovim/releases/download/v0.10.4/nvim-linux-x64.tar.gz"

Service tests (integration)

Use tmp_path fixtures with real filesystem but fake CommandRunner that records calls.

class FakeRunner:
    def __init__(self):
        self.calls: list[list[str]] = []
    def run(self, argv, **kwargs):
        self.calls.append(list(argv))
        return CompletedProcess(argv, 0, stdout="", stderr="")

def test_dotfiles_link_creates_symlinks(tmp_path):
    # Set up real dotfiles dir, real home dir
    # Inject FakeRunner for git calls
    # Call DotfilesService.link()
    # Assert symlinks exist on disk

E2E tests (opt-in, container-based)

Same pattern as current test_dotfiles_e2e_container.py. Run with FLOW_RUN_E2E=1.

---

13. Migration Notes

This is a full rewrite. The approach is:

1. Build domain/ layer first with comprehensive tests
2. Build services/ layer on top
3. Build commands/ + cli.py last
4. Delete all old code

No incremental migration. The old code serves as reference but is not preserved.