tm0/flow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
df8a19d6cc1e5d822e96376c1b381f15015e1899
flow/docs/superpowers/specs/2026-03-16-flow-architecture-redesign.md
Tomas Mirchev df8a19d6cc Fix all plan review issues, save implementation plan 
Plan fixes:
- detect_platform raises FlowError not RuntimeError
- TargetConfig lives in core/config.py only (remote domain imports it)
- plan_link handles source changes (remove_link + create_link)
- resolve_package_targets skips local files when mount_path is root
- LinkedState.from_dict guards on version mismatch
- Added missing test for parse_module_ref with absent ref
- Task 12 now has full tests and serialization format
- Task 13 uses spec signatures as truth, old code as reference
- Task 15 includes describe() examples and tests
- Task 24 has detailed test cases for ProjectService
- Note that conflicts.py is intentionally merged into planning.py
- Spec Section 12 example fixed to include filesystem_check arg

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-03-16 04:39:11 +02:00

1274 lines
40 KiB
Markdown
Raw Blame History

# 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. Domain functions that need filesystem information (directory listings, file existence) receive it as pre-built data or injected callbacks -- never by reading the filesystem directly.
- `services/` receives all dependencies via constructor (runtime, config, console). No module-level singletons. Cross-service calls use the shared `FlowContext` (e.g. `BootstrapService` constructs `DotfilesService(self.ctx)` internally). This is the intended pattern since all services share the same runtime context.
- `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, TargetConfig, SSHCommand
resolution.py # Target parsing, host template expansion
containers/
models.py # ContainerSpec, ContainerInfo, ImageRef
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.
```python
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
```python
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
```python
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
```python
@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.
```python
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
```python
@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
package_id: str # Qualified: "layer/name" (e.g. "_shared/nvim")
source_dir: Path # Absolute path in dotfiles repo
module: ModuleRef | None # If backed by external repo
local_files: list[tuple[Path, Path]] # (absolute_source, relative_to_package_root)
# Pre-walked by the service layer (no I/O in domain)
@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/<package_id>
module_files: list[tuple[Path, Path]] # (absolute_source, relative_to_cache_root)
# Pre-walked by the service layer
@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.
```python
# domain/dotfiles/resolution.py
# All functions are PURE: they operate on pre-built Package data (including
# pre-walked file lists), never touching the filesystem directly.
def resolve_package_targets(
package: Package,
home: Path,
skip: set[str],
) -> list[LinkTarget]:
"""Resolve all LinkTargets for a package, handling modules correctly.
Uses package.local_files and package.module.module_files (pre-walked
by service layer) to compute targets. No filesystem I/O.
"""
...
def resolve_all_targets(
packages: list[Package],
home: Path,
skip: set[str],
) -> list[LinkTarget]:
"""Resolve targets for all packages. Raises PlanConflict on duplicate targets
across packages (e.g. _shared/zsh and linux-work/zsh both targeting ~/.zshrc).
This is a pure cross-package collision check, not a filesystem conflict check.
"""
...
```
```python
# domain/dotfiles/modules.py
# Pure functions for module metadata. I/O (reading YAML, walking dirs) is done
# by the service layer, which passes parsed data into these functions.
def parse_module_ref(
raw: dict, # Pre-loaded YAML content (dict, not file path)
package_id: str,
mount_path: Path,
modules_base: Path,
) -> ModuleRef:
"""Build a ModuleRef from parsed _module.yaml content.
Args:
raw: The parsed YAML dict (service reads the file and passes content in).
package_id: Qualified name like "_shared/nvim".
mount_path: Relative path from package root to _module.yaml parent.
modules_base: Base directory for module caches (from core/paths.py).
"""
...
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_id: str, modules_base: Path) -> Path:
"""Compute cache dir for a module. Uses package_id (e.g. '_shared/nvim')
with '/' replaced by '--' to avoid collisions between same-named packages
in different layers."""
return modules_base / package_id.replace("/", "--")
def normalize_source(source: str) -> str:
"""github:org/repo -> https://github.com/org/repo.git"""
...
```
**I/O boundary note:** The service layer (`DotfilesService`) is responsible for:
1. Walking the dotfiles dir to find packages and `_module.yaml` files
2. Reading `_module.yaml` files and passing parsed dicts to `parse_module_ref`
3. Walking module cache dirs to build file lists
4. Constructing `Package` objects with pre-populated `local_files` and `ModuleRef.module_files`
5. Passing these fully-built objects to pure domain functions
### 4.3 Planning and Conflict Detection
Conflict detection and planning are integrated into a single flow. There are two kinds of conflicts:
1. **Cross-package collisions** (pure): two packages want the same target path. Detected by `resolve_all_targets` which raises `PlanConflict`.
2. **Filesystem conflicts** (requires I/O): a target path already exists on disk and is not managed by flow. Detected by `plan_link` via an injected callback.
```python
# domain/dotfiles/planning.py
def plan_link(
desired: list[LinkTarget],
current: LinkedState,
filesystem_check: Callable[[Path], str | None],
# ^^ Injected by service layer. Returns "file", "dir", "symlink", or None.
# This is the ONLY I/O dependency in the planning layer.
# Tests provide a fake (e.g., lambda p: None).
) -> LinkPlan:
"""Compare desired targets with current state, produce reconciliation plan.
- New targets: create_link ops
- Removed targets (in current but not desired): remove_link ops
- Existing correct symlinks: unchanged (no op)
- Filesystem conflicts: listed in plan.conflicts (target exists, not managed)
- Directory conflicts: listed in plan.conflicts (target is a dir, cannot overwrite)
The service checks plan.conflicts and raises PlanConflict if non-empty and
--force was not passed.
"""
...
def plan_unlink(
current: LinkedState,
packages: list[str] | None, # None = unlink all
) -> LinkPlan:
"""Plan removal of managed links."""
...
```
---
## 5. Packages Domain
Shared by `flow packages` and `flow setup`. Single source of truth.
### 5.1 Models
```python
@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
```python
# 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
```python
@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
setup_modules: list[SetupModuleDef] # Parsed 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 SetupModuleDef:
"""Configuration for a single setup module step, parsed from YAML.
Examples:
SetupModuleDef(type="hostname", config={"value": "my-host"})
SetupModuleDef(type="ssh-keygen", config={"keys": [{"type": "ed25519", ...}]})
SetupModuleDef(type="runcmd", config={"commands": ["sudo groupadd docker || true"]})
"""
type: str # "hostname" | "locale" | "shell" | "ssh-keygen" | "runcmd"
config: dict # Type-specific configuration from YAML
@dataclass(frozen=True)
class BootstrapAction:
"""A single step in the bootstrap plan."""
type: str # "install_packages" | "run_setup_module"
# | "link_dotfiles" | "set_shell"
description: str
payload: Any # Type-specific data:
# install_packages -> PackagePlan
# run_setup_module -> tuple[SetupModuleDef, list[str]]
# (the module def + pre-computed shell commands)
# link_dotfiles -> profile name (str)
# set_shell -> shell name (str)
critical: bool # Stop on failure?
@dataclass(frozen=True)
class BootstrapPlan:
profile: str
actions: list[BootstrapAction]
summary: str
```
**Note on env var validation:** `plan_bootstrap` validates required env vars eagerly and raises `ConfigError` if any are missing. This is a pure check (comparing `profile.requires` against the provided `env` dict) and happens before any actions are generated. It is not an action in the plan -- missing env vars are a precondition failure that prevents plan creation.
### 6.2 Setup Modules
Each module type is a small class with a plan method (pure) that returns shell commands to execute.
```python
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
```python
# 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
```yaml
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
```python
@dataclass(frozen=True)
class TargetConfig:
"""A target as defined in config YAML. Parsed from the 'targets' section.
Supports two YAML formats:
personal@orb: personal.orb # shorthand: key is namespace@platform, value is host
work@ec2: # dict form
host: work.ec2.internal
identity: ~/.ssh/id_work
"""
namespace: str
platform: str
host: str
identity: str | None = None
@dataclass(frozen=True)
class Target:
"""A fully resolved SSH target (after merging CLI args + config + templates)."""
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
```python
# 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
```python
@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
```python
# 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
```python
class DotfilesService:
def __init__(self, ctx: FlowContext): ...
def init(self, repo_url: str, *, dry_run: bool) -> None:
"""Clone dotfiles repo + discover and clone all module repos.
dry_run: prints repos that would be cloned without cloning."""
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. Read-only."""
def edit(self, target: str, *, no_commit: bool) -> None:
"""Pull relevant repo -> open editor -> commit + push.
Interactive command -- dry_run not applicable (editor is the point).
no_commit: skip the auto-commit/push after editing."""
def repos_list(self) -> None:
"""List all managed repos. Read-only."""
def repos_status(self, repo_filter: str | None) -> None:
"""Git status for repos. Read-only."""
def repos_pull(self, repo_filter: str | None, *, dry_run: bool) -> None:
"""Pull one or all repos. dry_run: shows what would be pulled."""
def repos_push(self, repo_filter: str | None, *, dry_run: bool) -> None:
"""Push one or all repos. dry_run: shows what would be pushed."""
```
#### Link flow detail
The service layer performs all I/O (directory walking, YAML reading, filesystem checks) and passes pre-built data structures to pure domain functions:
```python
def link(self, ...):
# I/O: walk dirs, read _module.yaml files, build Package objects
packages = self._discover_packages(profile)
# Pure: resolve file paths to LinkTargets
targets = resolve_all_targets(packages, home, skip)
# I/O: load persisted state
current = self._load_linked_state()
# Pure (with injected callback): build plan with filesystem checks
plan = plan_link(targets, current,
filesystem_check=self._check_path_on_disk)
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
```python
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
```python
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:
"""List available profiles with summary info. Read-only."""
def show(self, profile: str, *, dry_run: bool) -> None:
"""Show profile details. When dry_run=True (or always, effectively),
builds and prints the full BootstrapPlan without executing.
'setup show' is conceptually 'setup run --dry-run' scoped to one profile."""
```
#### Run flow detail
```python
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
```python
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
```python
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:
"""Interactive -- attach to tmux session. No dry_run."""
def exec(self, name: str, cmd: list[str] | None) -> None:
"""Interactive -- exec into container. No dry_run."""
def list(self) -> None:
"""Read-only."""
def stop(self, name: str, *, kill: bool, dry_run: bool) -> None: ...
def remove(self, name: str, *, force: bool, dry_run: bool) -> None: ...
def respawn(self, name: str, *, dry_run: bool) -> None: ...
```
### 9.6 ProjectService
```python
class ProjectService:
def __init__(self, ctx: FlowContext): ...
def check(self, *, fetch: bool) -> None:
"""Read-only (fetch is network I/O but does not mutate local state
beyond FETCH_HEAD -- acceptable for a check command)."""
def fetch(self) -> None:
"""Fetch all remotes. Network I/O only, no local mutations beyond
remote tracking refs. dry_run not applicable."""
def summary(self) -> None:
"""Read-only."""
```
---
## 10. CLI Entry Point
```python
# 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:
```python
# 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.
```yaml
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: two formats supported
# Shorthand: "namespace@platform: host [identity]"
# Dict form: "namespace@platform: {host: ..., identity: ...}"
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]
- name: docker
type: pkg
sources:
apt: docker-ce
allow-sudo: true # Allow sudo in post-install hook
post-install: | # Shell script, runs after install
sudo groupadd docker || true
sudo usermod -aG docker $USER
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"
packages:
- git
- fd
- binary/neovim
- name: docker
allow-sudo: true
```
### Path constants (`core/paths.py`)
All XDG-compliant, hardcoded in `core/paths.py`:
```python
CONFIG_DIR = XDG_CONFIG_HOME / "flow" # ~/.config/flow/
DATA_DIR = XDG_DATA_HOME / "flow" # ~/.local/share/flow/
STATE_DIR = XDG_STATE_HOME / "flow" # ~/.local/state/flow/
DOTFILES_DIR = DATA_DIR / "dotfiles" # Cloned dotfiles repo
MODULES_DIR = DATA_DIR / "modules" # Module cache (each module cloned here)
PACKAGES_DIR = DATA_DIR / "packages" # Binary package scratch
LINKED_STATE = STATE_DIR / "linked.json" # Current link state
INSTALLED_STATE = STATE_DIR / "installed.json" # Installed packages state
# Self-hosted config (from dotfiles repo, takes priority over CONFIG_DIR)
DOTFILES_FLOW_CONFIG = DOTFILES_DIR / "_shared" / "flow" / ".config" / "flow"
```
`MODULES_DIR` is the `modules_base` passed to `module_cache_dir()` and other domain functions.
### Target config normalization
The config parser in `core/config.py` normalizes both target formats into `TargetConfig` objects:
```python
# Shorthand: "personal@orb: personal.orb"
# -> TargetConfig(namespace="personal", platform="orb", host="personal.orb", identity=None)
# Dict: "work@ec2: {host: ..., identity: ...}"
# -> TargetConfig(namespace="work", platform="ec2", host="work.ec2.internal", identity="~/.ssh/id_work")
```
---
## 12. Testing Strategy
### Domain tests (majority of tests)
Pure function tests. No mocks, no filesystem, no subprocess. Fast.
```python
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, filesystem_check=lambda p: None)
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.
```python
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.
Reference in New Issue View Git Blame Copy Permalink