The Orchestrator¶

@dataclass
class InitRequest:
    """Declarative intent from the caller for a scaffolding run.

    Attributes:
        template_blueprint: An optional pre-loaded template blueprint to apply.
        python_version: An optional Python version string (e.g. '3.13'). Informational;
            the modules list is already constructed with the resolved version.
        docker: If True, scaffolds container artifacts (.dockerignore, Dockerfile).
        force_merge: If True, bypasses collision prompts and forces a merge strategy.
        force_replace: If True, bypasses collision prompts and forces an overwrite strategy.
        metadata: Pre-resolved metadata dictionary to inject into the manifest.
        is_external: If True, the template was loaded from an external (untrusted) source.
        is_user_aliased: If True, the template was resolved via a trusted global config alias.
    """

    template_blueprint: TemplateBlueprint | None = None
    python_version: str | None = None
    docker: bool = False
    force_merge: bool = False
    force_replace: bool = False
    metadata: dict[str, Any] | None = field(default=None)
    is_external: bool = False
    is_user_aliased: bool = False

@dataclass(frozen=True)
class ExecutionResult:
    """Observed outcome returned by Orchestrator.execute().

    Attributes:
        touched_paths: Immutable set of relative paths written or created on disk.
        diagnostics: Ordered tuple of non-fatal diagnostic events emitted during execution.
    """

    touched_paths: frozenset[str]
    diagnostics: tuple[DiagnosticEvent, ...]

    def to_dict(self) -> dict[str, Any]:
        """Serializes the execution result to a JSON-safe dictionary.

        The ``touched_paths`` frozenset is emitted as a sorted list for deterministic
        output. Each diagnostic event is emitted as an explicit dict; the optional
        ``detail`` field is omitted when absent to keep payloads compact.

        Returns:
            A JSON-serializable dictionary representation.
        """
        diagnostics: list[dict[str, Any]] = []
        for event in self.diagnostics:
            entry: dict[str, Any] = {
                "phase": str(event.phase),
                "message": event.message,
                "severity": event.severity.value,
            }
            if event.detail is not None:
                entry["detail"] = event.detail
            diagnostics.append(entry)

        return {
            "touched_paths": sorted(self.touched_paths),
            "diagnostics": diagnostics,
        }

class Orchestrator:
    """Manages the lifecycle of the Python environment scaffolding process.

    The orchestrator provides a strict two-phase API:
    - plan(): Evaluates the workspace state and assembles a declarative
      EnvironmentManifest without mutating the filesystem.
    - execute(): Takes an already-built manifest and realizes it on disk.

    This separation guarantees that plan() is always safe to retry (it
    instantiates a fresh manifest on every call), and that execute() never
    performs planning, collision detection, or user interaction.
    """

    def __init__(
        self,
        modules: list[BootstrapModule],
        user_config: UserConfig,
        request: InitRequest | None = None,
    ) -> None:
        """Initializes the orchestrator with the requested modules and intent.

        Args:
            modules: The ordered stack of bootstrap layers to apply.
            user_config: The active UserConfig instance.
            request: Optional InitRequest describing caller intent. Defaults to a
                no-op InitRequest if omitted.
        """
        self.modules = modules
        self.user_config = user_config
        self.request = request or InitRequest()

    def plan(self) -> EnvironmentManifest:
        """Evaluates workspace state and assembles a declarative EnvironmentManifest.

        A fresh EnvironmentManifest is instantiated on every call, guaranteeing
        that retries (e.g. after a collision resolution) start from a clean slate.

        Raises:
            WorkspaceCollisionError: If collision markers exist on disk and no
                force flag (force_merge / force_replace) was provided in the request.
            MissingDependencyError: If a module pre-flight check fails.

        Returns:
            A populated EnvironmentManifest ready to be passed to execute().
        """
        req = self.request

        # Phase 1: Instantiate a fresh manifest for every plan() call
        manifest = EnvironmentManifest(
            force_merge=req.force_merge,
            force_replace=req.force_replace,
        )

        # Phase 2: Collision intercept (raises instead of prompting)
        collision_targets: set[Path] = set()
        for mod in self.modules:
            for marker in mod.collision_markers:
                if marker.exists():
                    collision_targets.add(marker)

        if collision_targets:
            if req.force_replace:
                logger.debug(
                    "--force-replace flag provided. Defaulting to OVERWRITE collision strategy."
                )
                manifest.collision_strategy = CollisionStrategy.OVERWRITE
            elif req.force_merge:
                logger.debug(
                    "--force-merge flag provided. Defaulting to MERGE collision strategy."
                )
                manifest.collision_strategy = CollisionStrategy.MERGE
            else:
                raise WorkspaceCollisionError(
                    paths=frozenset(collision_targets),
                )

        # Phase 3: Pre-flight verification
        missing_deps: dict[GlobalExecutable, MissingDependencyError] = {}
        for mod in self.modules:
            try:
                mod.pre_flight()
            except MissingDependencyError as e:
                missing_deps[e.dependency] = e

        if missing_deps:
            raise AggregatedDependencyError(tuple(missing_deps.values()))

        # Phase 4: Manifest aggregation
        if req.metadata:
            manifest.metadata.update(cast(ProjectMetadata, req.metadata))

        for mod in self.modules:
            mod.build(manifest)

        # Phase 5: Blueprint injection
        blueprint = req.template_blueprint
        if blueprint:
            logger.debug("Injecting blueprint structural fields into manifest.")

            for dep in blueprint.dependencies:
                manifest.dependencies.add(dep)

            for dep in blueprint.dev_dependencies:
                manifest.dependencies.add_dev(dep)

            for dep in blueprint.docs_dependencies:
                manifest.dependencies.add_docs(dep)

            for d in blueprint.directories:
                manifest.filesystem.add_directory(d)

            for ig in blueprint.vcs_ignores:
                manifest.filesystem.add_vcs_ignore(ig)

            for cmd in blueprint.system_tasks:
                manifest.tasks.add_system_task(cmd)

            for cmd in blueprint.post_install_tasks:
                manifest.tasks.add_post_install_task(cmd)

            if blueprint.pyproject_injections:
                logger.debug("Injecting pyproject.toml payloads from configuration.")
                for payload in blueprint.pyproject_injections.values():
                    manifest.filesystem.add_file_append("pyproject.toml", payload)

            if blueprint.appends:
                logger.debug("Injecting generic file appends from configuration.")
                for filepath, payloads in blueprint.appends.items():
                    for payload in payloads:
                        manifest.filesystem.add_file_append(filepath, payload)

            if blueprint.files:
                logger.debug("Injecting static files from configuration.")
                for filepath, content in blueprint.files.items():
                    manifest.filesystem.add_file_injection(filepath, content)

        return manifest

    def execute(self, manifest: EnvironmentManifest) -> ExecutionResult:
        """Realizes the pre-built manifest on disk.

        Takes an already-built manifest from plan() and executes it. Performs no
        planning, collision detection, template resolution, or user interaction.

        Args:
            manifest: The populated EnvironmentManifest to execute.

        Raises:
            PartialExecutionAbortedError: If the user interrupts execution after
                disk mutations have already begun.

        Returns:
            An ExecutionResult describing what was touched and any diagnostics.
        """
        executor = SystemExecutor(manifest, self.user_config, self.request.docker)
        try:
            executor.execute()
        except KeyboardInterrupt:
            raise PartialExecutionAbortedError(
                frozenset(executor.touched_paths)
            ) from None

        return ExecutionResult(
            touched_paths=frozenset(executor.touched_paths),
            diagnostics=tuple(executor.diagnostics),
        )